Integrated Variant¶

def feed_line(self, line: str) -> None:
    """Ingest one output line, assign its order, and schedule a flush.

    Args:
        line (str): The process output line to broadcast.
    """
    with self._lock:
        self._seq += 1
        item = {"seq": self._seq, "t": time.time(), "line": line}
        self._history.append(item)
        self._pending.append(item)
        if not self._flush_scheduled:
            self._flush_scheduled = True
            try:
                self._sio.start_background_task(self._flush_after_delay)
            except Exception:
                # Un-wedge the scheduler: the line stays pending and the
                # next feed_line retries; a stuck True flag would silence
                # the stream forever.
                self._flush_scheduled = False
                raise

def history(self) -> list[dict]:
    """Return a snapshot of buffered lines for late-joining clients.

    Returns:
        list[dict]: Buffered items with ``seq``, ``t`` and ``line`` keys.
    """
    with self._lock:
        return list(self._history)

def on_process_exit(self, return_code: int) -> None:
    """Emit an end-of-stream marker when the server process exits.

    Args:
        return_code (int): Exit code of the BlueSky server process.
    """
    self.feed_line(f"--- bluesky server exited (return code {return_code}) ---")

def start(self) -> dict:
    """Spawn the headless server (in its own process group) and a reader.

    The process is started in a new session with merged, line-buffered
    stdout/stderr so the reader receives one ordered stream for the
    server and all node children. A no-op if the server is already
    running. If a concurrent :meth:`stop` is mid-shutdown, waits for it
    to finish and then starts a fresh server instead of reporting the
    doomed process as "already running".

    Returns:
        dict: Result with ``success``, ``status``, ``pid`` and
            ``message`` (plus ``error`` on failure).
    """
    with self._lock:
        stopping_proc = self._proc if self._state == "stopping" else None
    if stopping_proc is not None:
        # A concurrent stop() owns this process's shutdown; wait it out
        # (worst case escalate_after + the 5s SIGKILL wait) rather than
        # report a pid that is about to die as "already running".
        try:
            stopping_proc.wait(timeout=15)
        except subprocess.TimeoutExpired:
            return {
                "success": False,
                "status": "error",
                "message": "BlueSky server is still stopping; retry shortly",
            }

    with self._lock:
        if self._proc is not None and self._proc.poll() is None:
            return {
                "success": True,
                "status": "running",
                "pid": self._proc.pid,
                "message": "BlueSky server already running",
            }
        # PYTHONUNBUFFERED keeps the server's (and inherited children's)
        # stdout line-buffered so log lines arrive promptly and in order.
        env = dict(os.environ, PYTHONUNBUFFERED="1")
        try:
            self._proc = subprocess.Popen(
                self._cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,  # merge -> single ordered stream
                stdin=subprocess.DEVNULL,
                bufsize=1,
                text=True,
                env=env,
                start_new_session=True,  # own process group / session
                close_fds=True,
            )
        except Exception as e:
            self._state = "stopped"
            return {
                "success": False,
                "status": "error",
                "message": f"Failed to start BlueSky: {e}",
                "error": str(e),
            }
        self._state = "running"
        proc = self._proc

    # Launch the reader outside the lock.
    self._spawn(self._read_loop, proc)
    return {
        "success": True,
        "status": "running",
        "pid": proc.pid,
        "message": "BlueSky server started",
    }

def stop(self, sig: int = signal.SIGTERM, escalate_after: float = 5.0) -> dict:
    """Signal the whole process group, escalating to SIGKILL if needed.

    Args:
        sig (int): Signal sent to the process group first.
        escalate_after (float): Seconds to wait for exit before
            force-killing the group with SIGKILL.

    Returns:
        dict: Result with ``success``, ``status`` and ``message``.
            ``success`` is False if the process survives even SIGKILL.
    """
    with self._lock:
        proc = self._proc
        if proc is None or proc.poll() is not None:
            self._state = "stopped"
            return {
                "success": True,
                "status": "stopped",
                "message": "BlueSky server is not running",
            }
        self._state = "stopping"

    try:
        return self._terminate(proc, sig, escalate_after)
    except BaseException:
        # Never leave the state stranded at "stopping": start() would wait
        # 15s and then refuse, with no control surface able to clear it.
        self._settle(proc)
        raise

def kill(self) -> dict:
    """Force-kill the whole process group immediately (no graceful wait).

    Returns:
        dict: Result with ``success``, ``status`` and ``message``.
    """
    with self._lock:
        proc = self._proc
        was_running = proc is not None and proc.poll() is None
    result = self.stop(sig=signal.SIGKILL, escalate_after=2.0)
    # Only claim a kill if there was actually a live process to kill --
    # otherwise keep stop()'s "is not running", which the UI shows verbatim.
    if was_running and result.get("success") and result.get("status") == "stopped":
        result["message"] = "BlueSky server killed"
    return result

def restart(self) -> dict:
    """Stop the current tree (if any) and start a fresh one.

    Returns:
        dict: The ``start()`` result, with the message adjusted to
            "restarted" on success. If the old tree could not be stopped,
            its failure result is returned instead of starting a new one
            (``start()`` would just report the surviving process as
            "already running").
    """
    stop_result = self.stop()
    if not stop_result.get("success"):
        return stop_result
    result = self.start()
    if result.get("success"):
        result["message"] = "BlueSky server restarted"
    return result

def status(self) -> dict:
    """Report whether the server is running, with its pid and state.

    Returns:
        dict: Result with ``success``, ``running``, ``status`` and
            ``pid`` (None when stopped).
    """
    with self._lock:
        proc = self._proc
        if proc is None or proc.poll() is not None:
            return {
                "success": True,
                "running": False,
                "status": "stopped",
                "pid": None,
            }
        return {
            "success": True,
            "running": True,
            "status": self._state,
            "pid": proc.pid,
        }

def start(self) -> dict:
    """Spawn the headless server (in its own process group) and a reader.

    The process is started in a new session with merged, line-buffered
    stdout/stderr so the reader receives one ordered stream for the
    server and all node children. A no-op if the server is already
    running. If a concurrent :meth:`stop` is mid-shutdown, waits for it
    to finish and then starts a fresh server instead of reporting the
    doomed process as "already running".

    Returns:
        dict: Result with ``success``, ``status``, ``pid`` and
            ``message`` (plus ``error`` on failure).
    """
    with self._lock:
        stopping_proc = self._proc if self._state == "stopping" else None
    if stopping_proc is not None:
        # A concurrent stop() owns this process's shutdown; wait it out
        # (worst case escalate_after + the 5s SIGKILL wait) rather than
        # report a pid that is about to die as "already running".
        try:
            stopping_proc.wait(timeout=15)
        except subprocess.TimeoutExpired:
            return {
                "success": False,
                "status": "error",
                "message": "BlueSky server is still stopping; retry shortly",
            }

    with self._lock:
        if self._proc is not None and self._proc.poll() is None:
            return {
                "success": True,
                "status": "running",
                "pid": self._proc.pid,
                "message": "BlueSky server already running",
            }
        # PYTHONUNBUFFERED keeps the server's (and inherited children's)
        # stdout line-buffered so log lines arrive promptly and in order.
        env = dict(os.environ, PYTHONUNBUFFERED="1")
        try:
            self._proc = subprocess.Popen(
                self._cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,  # merge -> single ordered stream
                stdin=subprocess.DEVNULL,
                bufsize=1,
                text=True,
                env=env,
                start_new_session=True,  # own process group / session
                close_fds=True,
            )
        except Exception as e:
            self._state = "stopped"
            return {
                "success": False,
                "status": "error",
                "message": f"Failed to start BlueSky: {e}",
                "error": str(e),
            }
        self._state = "running"
        proc = self._proc

    # Launch the reader outside the lock.
    self._spawn(self._read_loop, proc)
    return {
        "success": True,
        "status": "running",
        "pid": proc.pid,
        "message": "BlueSky server started",
    }

def stop(self, sig: int = signal.SIGTERM, escalate_after: float = 5.0) -> dict:
    """Signal the whole process group, escalating to SIGKILL if needed.

    Args:
        sig (int): Signal sent to the process group first.
        escalate_after (float): Seconds to wait for exit before
            force-killing the group with SIGKILL.

    Returns:
        dict: Result with ``success``, ``status`` and ``message``.
            ``success`` is False if the process survives even SIGKILL.
    """
    with self._lock:
        proc = self._proc
        if proc is None or proc.poll() is not None:
            self._state = "stopped"
            return {
                "success": True,
                "status": "stopped",
                "message": "BlueSky server is not running",
            }
        self._state = "stopping"

    try:
        return self._terminate(proc, sig, escalate_after)
    except BaseException:
        # Never leave the state stranded at "stopping": start() would wait
        # 15s and then refuse, with no control surface able to clear it.
        self._settle(proc)
        raise

def kill(self) -> dict:
    """Force-kill the whole process group immediately (no graceful wait).

    Returns:
        dict: Result with ``success``, ``status`` and ``message``.
    """
    with self._lock:
        proc = self._proc
        was_running = proc is not None and proc.poll() is None
    result = self.stop(sig=signal.SIGKILL, escalate_after=2.0)
    # Only claim a kill if there was actually a live process to kill --
    # otherwise keep stop()'s "is not running", which the UI shows verbatim.
    if was_running and result.get("success") and result.get("status") == "stopped":
        result["message"] = "BlueSky server killed"
    return result

def restart(self) -> dict:
    """Stop the current tree (if any) and start a fresh one.

    Returns:
        dict: The ``start()`` result, with the message adjusted to
            "restarted" on success. If the old tree could not be stopped,
            its failure result is returned instead of starting a new one
            (``start()`` would just report the surviving process as
            "already running").
    """
    stop_result = self.stop()
    if not stop_result.get("success"):
        return stop_result
    result = self.start()
    if result.get("success"):
        result["message"] = "BlueSky server restarted"
    return result

def status(self) -> dict:
    """Report whether the server is running, with its pid and state.

    Returns:
        dict: Result with ``success``, ``running``, ``status`` and
            ``pid`` (None when stopped).
    """
    with self._lock:
        proc = self._proc
        if proc is None or proc.poll() is not None:
            return {
                "success": True,
                "running": False,
                "status": "stopped",
                "pid": None,
            }
        return {
            "success": True,
            "running": True,
            "status": self._state,
            "pid": proc.pid,
        }

def feed_line(self, line: str) -> None:
    """Ingest one output line, assign its order, and schedule a flush.

    Args:
        line (str): The process output line to broadcast.
    """
    with self._lock:
        self._seq += 1
        item = {"seq": self._seq, "t": time.time(), "line": line}
        self._history.append(item)
        self._pending.append(item)
        if not self._flush_scheduled:
            self._flush_scheduled = True
            try:
                self._sio.start_background_task(self._flush_after_delay)
            except Exception:
                # Un-wedge the scheduler: the line stays pending and the
                # next feed_line retries; a stuck True flag would silence
                # the stream forever.
                self._flush_scheduled = False
                raise

def history(self) -> list[dict]:
    """Return a snapshot of buffered lines for late-joining clients.

    Returns:
        list[dict]: Buffered items with ``seq``, ``t`` and ``line`` keys.
    """
    with self._lock:
        return list(self._history)

def on_process_exit(self, return_code: int) -> None:
    """Emit an end-of-stream marker when the server process exits.

    Args:
        return_code (int): Exit code of the BlueSky server process.
    """
    self.feed_line(f"--- bluesky server exited (return code {return_code}) ---")