fix(tui): close slash_worker sessions on WebSocket disconnect

tui_gateway/server.py

@@ -321,15 +321,73 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
             pass
 
 
+# ── Session teardown helpers ─────────────────────────────────────────
+
+
+def _coerce_close_on_disconnect(value: object) -> bool:
+    """Normalise ``close_on_disconnect`` from JSON-RPC params to a bool."""
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.strip().lower() in {"1", "true", "yes", "on"}
+    if value is None:
+        return False
+    return bool(value)
+
+
+def _close_session_by_id(sid: str, *, end_reason: str = "tui_close") -> bool:
+    """Pop and fully tear down a session: finalize, notify, close agent + worker.
+
+    Shared by ``session.close`` RPC, WebSocket-disconnect cleanup, and
+    server shutdown — guarantees all teardown paths are identical.
+    """
+    session = _sessions.pop(sid, None)
+    if not session:
+        return False
+    _finalize_session(session, end_reason=end_reason)
+    try:
+        from tools.approval import unregister_gateway_notify
+
+        unregister_gateway_notify(session["session_key"])
+    except Exception:
+        pass
+    try:
+        agent = session.get("agent")
+        if agent and hasattr(agent, "close"):
+            agent.close()
+    except Exception:
+        pass
+    try:
+        worker = session.get("slash_worker")
+        if worker:
+            worker.close()
+    except Exception:
+        pass
+    return True
+
+
+def _close_sessions_for_transport(
+    transport: object, *, end_reason: str = "ws_disconnect"
+) -> None:
+    """Tear down (or migrate) sessions owned by a particular transport.
+
+    * Sessions marked ``close_on_disconnect=True`` (sidecar / short-lived)
+      are eagerly finalised and their ``slash_worker`` is killed.
+    * Normal sessions fall back to the stdio transport so the session can
+      be reconnected later — historical ``handle_ws`` behaviour.
+    """
+    for sid, session in list(_sessions.items()):
+        if session.get("transport") is not transport:
+            continue
+        if session.get("close_on_disconnect"):
+            _close_session_by_id(sid, end_reason=end_reason)
+        else:
+            session["transport"] = _stdio_transport
+
+
 def _shutdown_sessions() -> None:
-    for session in list(_sessions.values()):
-        _finalize_session(session, end_reason="tui_shutdown")
-        try:
-            worker = session.get("slash_worker")
-            if worker:
-                worker.close()
-        except Exception:
-            pass
+    for sid in list(_sessions.keys()):
+        _close_session_by_id(sid, end_reason="tui_shutdown")
 
 
 atexit.register(_shutdown_sessions)
@@ -2069,6 +2127,7 @@ def _init_session(sid: str, key: str, agent, history: list, cols: int = 80):
         "tool_progress_mode": _load_tool_progress_mode(),
         "edit_snapshots": {},
         "tool_started_at": {},
+        "close_on_disconnect": False,
         # Pin async event emissions to whichever transport created the
         # session (stdio for Ink, JSON-RPC WS for the dashboard sidebar).
         "transport": current_transport() or _stdio_transport,
@@ -2239,6 +2298,9 @@ def _(rid, params: dict) -> dict:
     sid = uuid.uuid4().hex[:8]
     key = _new_session_key()
     cols = int(params.get("cols", 80))
+    close_on_disconnect = _coerce_close_on_disconnect(
+        params.get("close_on_disconnect")
+    )
     _enable_gateway_prompts()
 
     ready = threading.Event()
@@ -2249,6 +2311,7 @@ def _(rid, params: dict) -> dict:
         "agent_ready": ready,
         "attached_images": [],
         "cols": cols,
+        "close_on_disconnect": close_on_disconnect,
         "edit_snapshots": {},
         "history": [],
         "history_lock": threading.Lock(),
@@ -2780,29 +2843,7 @@ def _(rid, params: dict) -> dict:
 @method("session.close")
 def _(rid, params: dict) -> dict:
     sid = params.get("session_id", "")
-    session = _sessions.pop(sid, None)
-    if not session:
-        return _ok(rid, {"closed": False})
-    _finalize_session(session)
-    try:
-        from tools.approval import unregister_gateway_notify
-
-        unregister_gateway_notify(session["session_key"])
-    except Exception:
-        pass
-    try:
-        agent = session.get("agent")
-        if agent and hasattr(agent, "close"):
-            agent.close()
-    except Exception:
-        pass
-    try:
-        worker = session.get("slash_worker")
-        if worker:
-            worker.close()
-    except Exception:
-        pass
-    return _ok(rid, {"closed": True})
+    return _ok(rid, {"closed": _close_session_by_id(sid)})
 
 
 @method("session.branch")

tui_gateway/slash_worker.py

@@ -1,19 +1,85 @@
 """Persistent slash-command worker — one HermesCLI per TUI session.
 
 Protocol: reads JSON lines from stdin {id, command}, writes {id, ok, output|error} to stdout.
+
+Self-protection (defence-in-depth against orphaned workers):
+  1. **Parent watchdog** (best-effort) — if ``psutil`` is available, a daemon
+     thread monitors the parent's PID + create_time fingerprint every 10 s and
+     exits if the parent disappears. This handles crashes, SIGKILL, and PID
+     reuse (critical on Windows).
+  2. **Parent-PID poll** (fallback) — the main loop checks ``os.getppid()``
+     on each stdin timeout; works without ``psutil`` but cannot detect PID reuse.
+  3. **Idle timeout** — exits after ``_IDLE_TIMEOUT_S`` (30 min) without a
+     command, bounding worst-case resource usage even if both parent-orphan
+     guards fail.
 """
 
 import argparse
 import contextlib
 import io
 import json
 import os
+import select
 import sys
+import threading
+import time
 
 import cli as cli_mod
 from cli import HermesCLI
 from rich.console import Console
 
+# ── Optional psutil for parent fingerprinting ────────────────────────
+try:
+    import psutil
+except ImportError:
+    psutil = None
+
+# Max seconds of inactivity before the worker self-exits.
+_IDLE_TIMEOUT_S = 1800  # 30 minutes
+
+# How often the stdin poll returns to re-check conditions when idle.
+_POLL_INTERVAL_S = 60
+
+
+def _start_parent_watchdog() -> None:
+    """Start a daemon thread that exits this process if the parent disappears.
+
+    Uses ``psutil.Process(ppid).create_time()`` as a fingerprint to reliably
+    detect parent replacement (PID reuse).  Falls back to no-op when psutil
+    is not installed — the main loop's ``getppid()`` check provides a less
+    robust but dependency-free alternative.
+    """
+    if not psutil:
+        return
+
+    ppid = os.getppid()
+    try:
+        parent = psutil.Process(ppid)
+        parent_started = parent.create_time()
+    except (psutil.NoSuchProcess, psutil.AccessDenied):
+        os._exit(0)  # parent already gone
+
+    def _watch() -> None:
+        time.sleep(5)  # let the main process stabilise
+        while True:
+            try:
+                if not psutil.pid_exists(ppid):
+                    os._exit(0)
+                # Fingerprint check — catches PID reuse on Windows / fast-restart
+                if psutil.Process(ppid).create_time() != parent_started:
+                    os._exit(0)
+                # POSIX orphan check: adopted by init
+                if os.name != "nt" and os.getppid() == 1:
+                    os._exit(0)
+            except (psutil.NoSuchProcess, psutil.AccessDenied):
+                os._exit(0)
+            except Exception:
+                pass  # transient psutil error — retry
+            time.sleep(10)
+
+    t = threading.Thread(target=_watch, daemon=True, name="ParentWatchdog")
+    t.start()
+
 
 def _run(cli: HermesCLI, command: str) -> str:
     cmd = (command or "").strip()
@@ -44,6 +110,8 @@ def _run(cli: HermesCLI, command: str) -> str:
 
 
 def main():
+    _start_parent_watchdog()
+
     p = argparse.ArgumentParser(add_help=False)
     p.add_argument("--session-key", required=True)
     p.add_argument("--model", default="")
@@ -55,7 +123,38 @@ def main():
     with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
         cli = HermesCLI(model=args.model or None, compact=True, resume=args.session_key, verbose=False)
 
-    for raw in sys.stdin:
+    parent_pid = os.getppid()
+    last_command_time = time.monotonic()
+
+    while True:
+        # Guard 1: parent-PID check (works without psutil, covers basic
+        # orphan scenarios but not PID reuse).
+        if os.getppid() != parent_pid:
+            break
+
+        # Guard 2: idle timeout — bounds resource use when the upstream
+        # cleanup path (close_on_disconnect + server-side teardown) misses
+        # a code path.
+        idle = time.monotonic() - last_command_time
+        if idle >= _IDLE_TIMEOUT_S:
+            break
+
+        # Poll stdin with a timeout so we can periodically re-check the
+        # conditions above. Without this, a blocking ``for raw in sys.stdin``
+        # would hang forever if the pipe is left open.
+        poll_timeout = min(_POLL_INTERVAL_S, _IDLE_TIMEOUT_S - idle)
+        try:
+            r, _, _ = select.select([sys.stdin], [], [], poll_timeout)
+        except (ValueError, OSError):
+            break  # stdin closed or invalid
+
+        if not r:
+            continue  # Timeout — loop back to check parent/idle
+
+        raw = sys.stdin.readline()
+        if not raw:
+            break  # EOF — stdin closed
+
         line = raw.strip()
         if not line:
             continue
@@ -67,6 +166,7 @@ def main():
             out = _run(cli, req.get("command", ""))
             sys.stdout.write(json.dumps({"id": rid, "ok": True, "output": out}) + "\n")
             sys.stdout.flush()
+            last_command_time = time.monotonic()
         except Exception as e:
             sys.stdout.write(json.dumps({"id": rid, "ok": False, "error": str(e)}) + "\n")
             sys.stdout.flush()

tui_gateway/ws.py

@@ -166,11 +166,10 @@ async def handle_ws(ws: Any) -> None:
     finally:
         transport.close()
 
-        # Detach the transport from any sessions it owned so later emits
-        # fall back to stdio instead of crashing into a closed socket.
-        for _, sess in list(server._sessions.items()):
-            if sess.get("transport") is transport:
-                sess["transport"] = server._stdio_transport
+        # Preserve the historical "session survives reconnect" behaviour for
+        # normal TUI sessions, but eagerly close explicit sidecar sessions
+        # whose slash_worker should not outlive this websocket.
+        server._close_sessions_for_transport(transport, end_reason="ws_disconnect")
 
         try:
             await ws.close()