feat(chat): keep-alive + reattach for dashboard terminal sessions

hermes_cli/pty_session.py

@@ -0,0 +1,185 @@
+"""Keep-alive PTY sessions for dashboard terminals.
+
+A PTY process outlives the WebSocket that created it: a single drain task
+always reads the PTY into a bounded RingBuffer and forwards to the attached
+socket when present. Reconnecting with the same opaque token replays the
+buffer and resumes live. See
+docs/superpowers/specs/2026-06-20-pty-keepalive-reattach-design.md.
+"""
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import Optional
+
+WS_CLOSE_PROCESS_EXITED = 4410
+WS_CLOSE_SUPERSEDED = 4409
+
+
+class RingBuffer:
+    """Keeps only the most recent ``capacity`` bytes appended to it."""
+
+    def __init__(self, capacity: int) -> None:
+        self._cap = capacity
+        self._buf = bytearray()
+        self._truncated = False
+
+    def append(self, data: bytes) -> None:
+        self._buf.extend(data)
+        overflow = len(self._buf) - self._cap
+        if overflow > 0:
+            del self._buf[:overflow]
+            self._truncated = True
+
+    def snapshot(self) -> bytes:
+        return bytes(self._buf)
+
+    @property
+    def truncated(self) -> bool:
+        return self._truncated
+
+
+class PtySession:
+    def __init__(self, key: str, bridge, *, buffer_cap: int, read_timeout: float) -> None:
+        self.key = key
+        self.bridge = bridge
+        self.buffer = RingBuffer(buffer_cap)
+        self.alive = True
+        self.attached = False
+        self.last_detached_at: Optional[float] = None
+        self._read_timeout = read_timeout
+        self._ws = None
+        self._drain_task: Optional[asyncio.Task] = None
+
+    async def start(self) -> None:
+        self._drain_task = asyncio.create_task(self._drain())
+
+    async def _drain(self) -> None:
+        loop = asyncio.get_running_loop()
+        while True:
+            chunk = await loop.run_in_executor(None, self.bridge.read, self._read_timeout)
+            if chunk is None:                       # EOF — the agent process exited
+                self.alive = False
+                ws = self._ws
+                if ws is not None:
+                    try:
+                        await ws.close(code=WS_CLOSE_PROCESS_EXITED)
+                    except Exception:
+                        pass
+                return
+            if not chunk:                            # idle tick
+                await asyncio.sleep(0)
+                continue
+            self.buffer.append(chunk)
+            ws = self._ws
+            if ws is not None:
+                try:
+                    await ws.send_bytes(chunk)
+                except Exception:
+                    pass                             # detached mid-send; keep buffering
+
+    async def attach(self, ws) -> None:
+        old = self._ws
+        if old is not None and old is not ws:
+            try:
+                await old.close(code=WS_CLOSE_SUPERSEDED)
+            except Exception:
+                pass
+        self._ws = ws
+        self.attached = True
+        self.last_detached_at = None
+        snap = self.buffer.snapshot()
+        if snap:
+            await ws.send_bytes(snap)
+
+    def detach(self, ws) -> None:
+        if self._ws is ws:
+            self._ws = None
+        self.attached = False
+        self.last_detached_at = time.monotonic()
+
+    async def close(self) -> None:
+        if self._drain_task is not None:
+            self._drain_task.cancel()
+            try:
+                await self._drain_task
+            except (asyncio.CancelledError, Exception):
+                pass
+        try:
+            self.bridge.close()
+        except Exception:
+            pass
+
+
+from typing import Callable, Dict, Tuple
+
+
+class RegistryFull(Exception):
+    pass
+
+
+async def run_reaper(registry: "PtySessionRegistry", *, interval: float = 60.0) -> None:
+    """Periodically reap idle/dead keep-alive sessions. Cancelled on shutdown."""
+    while True:
+        await asyncio.sleep(interval)
+        try:
+            await registry.reap_idle()
+        except Exception:
+            pass
+
+
+class PtySessionRegistry:
+    def __init__(self, *, ttl: float, max_sessions: int,
+                 buffer_cap: int, read_timeout: float) -> None:
+        self._ttl = ttl
+        self._max = max_sessions
+        self._buffer_cap = buffer_cap
+        self._read_timeout = read_timeout
+        self._sessions: Dict[str, PtySession] = {}
+
+    async def attach_or_spawn(self, key: str, *, spawn: Callable[[], object]
+                              ) -> Tuple[PtySession, bool]:
+        await self.reap_idle()
+        existing = self._sessions.get(key)
+        if existing is not None and existing.alive:
+            return existing, False
+        if existing is not None:                       # dead remnant
+            await existing.close()
+            self._sessions.pop(key, None)
+        if len(self._sessions) >= self._max:
+            self._reap_one_idle_or_raise()
+        bridge = spawn()
+        session = PtySession(key, bridge, buffer_cap=self._buffer_cap,
+                             read_timeout=self._read_timeout)
+        await session.start()
+        self._sessions[key] = session
+        return session, True
+
+    def detach(self, key: str, ws) -> None:
+        s = self._sessions.get(key)
+        if s is not None:
+            s.detach(ws)
+
+    async def reap_idle(self, now: Optional[float] = None) -> None:
+        now = time.monotonic() if now is None else now
+        doomed = [
+            key for key, s in self._sessions.items()
+            if (not s.alive)
+            or (not s.attached and s.last_detached_at is not None
+                and (now - s.last_detached_at) > self._ttl)
+        ]
+        for key in doomed:
+            await self._sessions.pop(key).close()
+
+    def _reap_one_idle_or_raise(self) -> None:
+        idle = [s for s in self._sessions.values()
+                if not s.attached and s.last_detached_at is not None]
+        if not idle:
+            raise RegistryFull()
+        oldest = min(idle, key=lambda s: s.last_detached_at or 0.0)
+        self._sessions.pop(oldest.key, None)
+        asyncio.create_task(oldest.close())
+
+    async def close_all(self) -> None:
+        for key in list(self._sessions):
+            await self._sessions.pop(key).close()

hermes_cli/web_server.py

@@ -166,9 +166,14 @@ async def _lifespan(app: "FastAPI"):
         )
         cron_thread.start()
 
+    # Reap idle/dead keep-alive PTY sessions in the background (30-min TTL).
+    pty_reaper_task = asyncio.create_task(run_reaper(PTY_REGISTRY))
+
     try:
         yield
     finally:
+        pty_reaper_task.cancel()
+        await PTY_REGISTRY.close_all()
         if cron_stop is not None:
             cron_stop.set()
 
@@ -10925,6 +10930,65 @@ class PtyUnavailableError(RuntimeError):  # type: ignore[no-redef]
 
 _RESIZE_RE = re.compile(rb"\x1b\[RESIZE:(\d+);(\d+)\]")
 _PTY_READ_CHUNK_TIMEOUT = 0.2
+
+# Keep-alive PTY sessions: a terminal connecting with ``?attach=<token>`` is
+# bound to a process that survives disconnect/refresh and is reattachable.
+from hermes_cli.pty_session import PtySessionRegistry, RegistryFull, run_reaper  # noqa: E402
+
+PTY_REGISTRY = PtySessionRegistry(
+    ttl=30 * 60,
+    max_sessions=16,
+    buffer_cap=1 * 1024 * 1024,
+    read_timeout=_PTY_READ_CHUNK_TIMEOUT,
+)
+
+
+async def _legacy_pump(ws: "WebSocket", bridge) -> None:
+    """Original 1:1 socket<->PTY pump: stream until disconnect, then close the
+    bridge. Used when no ``?attach=`` token is supplied (keep-alive opt-in)."""
+    loop = asyncio.get_running_loop()
+
+    async def pump_pty_to_ws() -> None:
+        while True:
+            chunk = await loop.run_in_executor(None, bridge.read, _PTY_READ_CHUNK_TIMEOUT)
+            if chunk is None:
+                return
+            if not chunk:
+                await asyncio.sleep(0)
+                continue
+            try:
+                await ws.send_bytes(chunk)
+            except Exception:
+                return
+
+    reader_task = asyncio.create_task(pump_pty_to_ws())
+    try:
+        while True:
+            msg = await ws.receive()
+            if msg.get("type") == "websocket.disconnect":
+                break
+            raw = msg.get("bytes")
+            if raw is None:
+                text = msg.get("text")
+                raw = text.encode("utf-8") if isinstance(text, str) else b""
+            if not raw:
+                continue
+            match = _RESIZE_RE.match(raw)
+            if match and match.end() == len(raw):
+                bridge.resize(cols=int(match.group(1)), rows=int(match.group(2)))
+                continue
+            bridge.write(raw)
+    except WebSocketDisconnect:
+        pass
+    finally:
+        reader_task.cancel()
+        try:
+            await reader_task
+        except (asyncio.CancelledError, Exception):
+            pass
+        bridge.close()
+
+
 _VALID_CHANNEL_RE = re.compile(r"^[A-Za-z0-9._-]{1,128}$")
 # Starlette's TestClient reports the peer as "testclient"; treat it as
 # loopback so tests don't need to rewrite request scope.
@@ -11439,43 +11503,43 @@ async def pty_ws(ws: WebSocket) -> None:
         return
 
 
+    attach_token = ws.query_params.get("attach") or None
+
+    def _spawn():
+        return PtyBridge.spawn(argv, cwd=cwd, env=env)
+
+    if attach_token is None:
+        # Legacy path: 1:1 socket<->PTY, killed on disconnect (unchanged).
+        try:
+            bridge = _spawn()
+        except PtyUnavailableError as exc:
+            await ws.send_text(f"\r\n\x1b[31mChat unavailable: {exc}\x1b[0m\r\n")
+            await ws.close(code=1011)
+            return
+        except (FileNotFoundError, OSError) as exc:
+            await ws.send_text(f"\r\n\x1b[31mChat failed to start: {exc}\x1b[0m\r\n")
+            await ws.close(code=1011)
+            return
+        await _legacy_pump(ws, bridge)
+        return
+
+    # Keep-alive path: the PTY outlives this socket; reattach by token.
     try:
-        bridge = PtyBridge.spawn(argv, cwd=cwd, env=env)
+        session, _created = await PTY_REGISTRY.attach_or_spawn(attach_token, spawn=_spawn)
     except PtyUnavailableError as exc:
         await ws.send_text(f"\r\n\x1b[31mChat unavailable: {exc}\x1b[0m\r\n")
         await ws.close(code=1011)
         return
-    except (FileNotFoundError, OSError) as exc:
-        await ws.send_text(f"\r\n\x1b[31mChat failed to start: {exc}\x1b[0m\r\n")
+    except (FileNotFoundError, OSError, RegistryFull) as exc:
+        await ws.send_text(f"\r\n\x1b[31mChat unavailable: {exc}\x1b[0m\r\n")
         await ws.close(code=1011)
         return
 
-    loop = asyncio.get_running_loop()
-
-    # --- reader task: PTY master → WebSocket ----------------------------
-    async def pump_pty_to_ws() -> None:
-        while True:
-            chunk = await loop.run_in_executor(
-                None, bridge.read, _PTY_READ_CHUNK_TIMEOUT
-            )
-            if chunk is None:  # EOF
-                return
-            if not chunk:  # no data this tick; yield control and retry
-                await asyncio.sleep(0)
-                continue
-            try:
-                await ws.send_bytes(chunk)
-            except Exception:
-                return
-
-    reader_task = asyncio.create_task(pump_pty_to_ws())
-
-    # --- writer loop: WebSocket → PTY master ----------------------------
+    await session.attach(ws)
     try:
         while True:
             msg = await ws.receive()
-            msg_type = msg.get("type")
-            if msg_type == "websocket.disconnect":
+            if msg.get("type") == "websocket.disconnect":
                 break
             raw = msg.get("bytes")
             if raw is None:
@@ -11487,21 +11551,14 @@ async def pump_pty_to_ws() -> None:
             # Resize escape is consumed locally, never written to the PTY.
             match = _RESIZE_RE.match(raw)
             if match and match.end() == len(raw):
-                cols = int(match.group(1))
-                rows = int(match.group(2))
-                bridge.resize(cols=cols, rows=rows)
+                session.bridge.resize(cols=int(match.group(1)), rows=int(match.group(2)))
                 continue
 
-            bridge.write(raw)
+            session.bridge.write(raw)
     except WebSocketDisconnect:
         pass
     finally:
-        reader_task.cancel()
-        try:
-            await reader_task
-        except (asyncio.CancelledError, Exception):
-            pass
-        bridge.close()
+        PTY_REGISTRY.detach(attach_token, ws)
 
 
 # ---------------------------------------------------------------------------