Adds assistants.

castle-api/pyproject.toml

@@ -5,7 +5,7 @@ description = "Castle API"
 requires-python = ">=3.13"
 dependencies = [
     "fastapi>=0.115.0",
-    "uvicorn>=0.34.0",
+    "uvicorn[standard]>=0.34.0",
     "pydantic-settings>=2.0.0",
     "httpx>=0.27.0",
     "castle-core",

castle-api/src/castle_api/agent_registry.py

@@ -0,0 +1,234 @@
+"""Resolve launchable agents for the dashboard terminal UX.
+
+Config (a `castle.yaml` `agents:` block) declares agents; this layer adds the
+runtime view: is the binary present, and what's the absolute cwd. Castle stays
+agnostic — it only ever runs `command args` in a pty.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import shutil
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from castle_core.config import USER_TOOL_PATH_DIRS
+
+from castle_api.config import get_castle_root, get_config
+
+# Zero-config fallback so the feature works out of the box. A castle.yaml
+# `agents:` block overrides this entirely.
+_DEFAULT_AGENTS: dict[str, dict] = {
+    "claude": {
+        "command": "claude",
+        "description": "Anthropic Claude Code",
+        # Bare --resume pops claude's interactive session-history picker (whereas
+        # --continue would silently reopen only the most recent conversation).
+        "resume_args": ["--resume"],
+    },
+    "opencode": {
+        "command": "opencode",
+        "description": "opencode",
+        # opencode has no CLI flag to pop a picker, but it lists sessions as JSON
+        # and resumes by id — so the dashboard renders the picker itself.
+        "sessions": {
+            "list_command": ["opencode", "session", "list", "--format", "json"],
+            "resume": ["--session", "{id}"],
+            "id_field": "id",
+            "title_field": "title",
+            "time_field": "updated",
+        },
+    },
+    "amplifier": {
+        "command": "amplifier",
+        "description": "Amplifier",
+        # `amplifier resume` is a subcommand that interactively selects a past
+        # session (like claude --resume). (`amplifier continue` = most recent.)
+        "resume_args": ["resume"],
+    },
+}
+
+# Extra dirs to look in beyond the process PATH — where user CLIs actually live.
+# opencode installs to ~/.opencode/bin, which the systemd service PATH omits.
+_EXTRA_PATH_DIRS = [
+    *USER_TOOL_PATH_DIRS,
+    Path.home() / ".opencode" / "bin",
+]
+
+
+def _augmented_path() -> str:
+    extra = [str(d) for d in _EXTRA_PATH_DIRS if d.exists()]
+    current = os.environ.get("PATH", "")
+    return os.pathsep.join([*extra, current]) if extra else current
+
+
+@dataclass
+class ResolvedAgent:
+    name: str
+    command: str
+    args: list[str] = field(default_factory=list)
+    cwd: str = ""
+    env: dict[str, str] = field(default_factory=dict)
+    description: str | None = None
+    resume_args: list[str] = field(default_factory=list)
+    sessions: dict | None = None  # SessionsSpec-shaped, or None
+    resolved_command: str | None = None  # absolute path if found on PATH
+
+    @property
+    def available(self) -> bool:
+        return self.resolved_command is not None
+
+    @property
+    def can_continue(self) -> bool:
+        return bool(self.resume_args)
+
+    @property
+    def can_list_sessions(self) -> bool:
+        return bool(self.sessions and self.sessions.get("list_command"))
+
+    def launch_env(self) -> dict[str, str]:
+        """Env overrides for the pty: augmented PATH so the agent's own tool
+        lookups (node, etc.) resolve, plus any configured env."""
+        return {"PATH": _augmented_path(), **self.env}
+
+    def info(self) -> dict:
+        return {
+            "name": self.name,
+            "command": self.command,
+            "available": self.available,
+            "cwd": self.cwd,
+            "description": self.description,
+            "can_continue": self.can_continue,
+            "can_list_sessions": self.can_list_sessions,
+        }
+
+
+def _agent_specs() -> dict[str, dict]:
+    """Configured agents (as plain dicts), or the built-in defaults."""
+    try:
+        config = get_config()
+    except FileNotFoundError:
+        return _DEFAULT_AGENTS
+    if config.agents:
+        return {n: s.model_dump(exclude_none=True) for n, s in config.agents.items()}
+    return _DEFAULT_AGENTS
+
+
+def default_cwd() -> str:
+    """Where agents launch by default: the castle git repo (its CLAUDE.md /
+    AGENTS.md and `castle` sources), falling back to the config root, then home."""
+    try:
+        repo = get_config().repo
+        if repo:
+            return str(repo)
+    except FileNotFoundError:
+        pass
+    root = get_castle_root()
+    return str(root) if root else str(Path.home())
+
+
+def _resolve(name: str, spec: dict) -> ResolvedAgent:
+    command = spec["command"]
+    cwd = spec.get("cwd") or default_cwd()
+    resolved = shutil.which(command, path=_augmented_path())
+    return ResolvedAgent(
+        name=name,
+        command=command,
+        args=list(spec.get("args") or []),
+        cwd=cwd,
+        env=dict(spec.get("env") or {}),
+        description=spec.get("description"),
+        resume_args=list(spec.get("resume_args") or []),
+        sessions=spec.get("sessions"),
+        resolved_command=resolved,
+    )
+
+
+def list_agents() -> list[ResolvedAgent]:
+    return [_resolve(name, spec) for name, spec in _agent_specs().items()]
+
+
+def resolve_agent(name: str) -> ResolvedAgent | None:
+    specs = _agent_specs()
+    spec = specs.get(name)
+    if spec is None:
+        return None
+    return _resolve(name, spec)
+
+
+def _dig(obj: object, path: str) -> object:
+    """Read a (possibly dotted) field path off a nested dict, else None."""
+    cur = obj
+    for part in path.split("."):
+        if isinstance(cur, dict):
+            cur = cur.get(part)
+        else:
+            return None
+    return cur
+
+
+def resume_argv(agent: ResolvedAgent, session_id: str) -> list[str] | None:
+    """Build the argv that resumes a specific past session by id, or None."""
+    if not agent.sessions:
+        return None
+    template = agent.sessions.get("resume") or []
+    if not template:
+        return None
+    base = agent.resolved_command or agent.command
+    return [base, *(part.replace("{id}", session_id) for part in template)]
+
+
+async def list_agent_history(agent: ResolvedAgent, limit: int = 40) -> list[dict]:
+    """Run the agent's declared list_command and normalize it to
+    [{id, title, time}] — newest first. Best-effort: any failure yields []."""
+    if not agent.can_list_sessions:
+        return []
+    assert agent.sessions is not None
+    argv = agent.sessions["list_command"]
+    env = {**os.environ, "PATH": _augmented_path()}
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            *argv,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.DEVNULL,
+            cwd=agent.cwd or None,
+            env=env,
+        )
+        out, _ = await asyncio.wait_for(proc.communicate(), timeout=20)
+    except (OSError, asyncio.TimeoutError):
+        return []
+    try:
+        data = json.loads(out.decode() or "[]")
+    except json.JSONDecodeError:
+        return []
+    items = (
+        data
+        if isinstance(data, list)
+        else (data.get("sessions") or data.get("data") or [])
+    )
+    id_f = agent.sessions.get("id_field", "id")
+    title_f = agent.sessions.get("title_field", "title")
+    time_f = agent.sessions.get("time_field", "updated")
+    out_rows: list[dict] = []
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        sid = _dig(item, id_f)
+        if not sid:
+            continue
+        out_rows.append(
+            {
+                "agent": agent.name,
+                "id": str(sid),
+                "title": str(_dig(item, title_f) or ""),
+                "time": _dig(item, time_f),
+            }
+        )
+    # Newest first when the time field is sortable; keep input order otherwise.
+    try:
+        out_rows.sort(key=lambda r: r["time"] or 0, reverse=True)
+    except TypeError:
+        pass
+    return out_rows[:limit]

castle-api/src/castle_api/agent_sessions.py

@@ -0,0 +1,169 @@
+"""Live agent/terminal session registry.
+
+A session is a running PTY (an agent CLI or a shell) whose lifetime is decoupled
+from any WebSocket connection: closing the browser tab detaches, it does not
+kill the process. Sessions can be listed, re-attached (resumed) by id, and
+explicitly terminated. Each session keeps a bounded scrollback buffer so a
+re-attaching client can repaint the terminal.
+
+This is deliberately in-memory and single-node: sessions do not survive a
+castle-api restart. That's the right scope for a personal dashboard — for true
+cross-restart persistence you'd launch the agent under tmux, which breaks the
+"castle just runs a command" agnosticism.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+import uuid
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from castle_api.pty_session import PtySession
+
+# Cap the per-session replay buffer. Raw terminal bytes; enough to repaint a
+# full-screen TUI and some history without unbounded growth.
+_SCROLLBACK_BYTES = 256 * 1024
+# Reap sessions that have exited and sat idle longer than this (seconds).
+_EXITED_TTL = 3600
+
+
+@dataclass
+class AgentSession:
+    """One live PTY plus its scrollback and attached subscribers."""
+
+    id: str
+    agent: str  # agent name, or "terminal" for the plain shell
+    command: str
+    cwd: str
+    created_at: float
+    pty: PtySession | None = None
+    scrollback: bytearray = field(default_factory=bytearray)
+    subscribers: set[asyncio.Queue[bytes | None]] = field(default_factory=set)
+    exited: bool = False
+    exit_code: int | None = None
+    exited_at: float | None = None
+
+    def _on_output(self, data: bytes) -> None:
+        self.scrollback.extend(data)
+        if len(self.scrollback) > _SCROLLBACK_BYTES:
+            del self.scrollback[: len(self.scrollback) - _SCROLLBACK_BYTES]
+        for q in list(self.subscribers):
+            try:
+                q.put_nowait(data)
+            except asyncio.QueueFull:
+                # Slow client: drop it; the WS side will notice on its next send.
+                self.subscribers.discard(q)
+
+    def _on_exit(self) -> None:
+        self.exited = True
+        self.exit_code = self.pty.proc.returncode if self.pty else None
+        self.exited_at = time.time()
+        for q in list(self.subscribers):
+            q.put_nowait(None)  # EOF sentinel
+
+    def subscribe(self) -> asyncio.Queue[bytes | None]:
+        q: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=256)
+        self.subscribers.add(q)
+        return q
+
+    def unsubscribe(self, q: asyncio.Queue[bytes | None]) -> None:
+        self.subscribers.discard(q)
+
+    @property
+    def running(self) -> bool:
+        return not self.exited and self.pty is not None and self.pty.running
+
+    def info(self) -> dict:
+        return {
+            "id": self.id,
+            "agent": self.agent,
+            "command": self.command,
+            "cwd": self.cwd,
+            "created_at": self.created_at,
+            "running": self.running,
+            "exited": self.exited,
+            "exit_code": self.exit_code,
+            "cols": self.pty.cols if self.pty else None,
+            "rows": self.pty.rows if self.pty else None,
+            "clients": len(self.subscribers),
+        }
+
+
+class SessionManager:
+    """In-memory registry of live agent sessions."""
+
+    def __init__(self) -> None:
+        self._sessions: dict[str, AgentSession] = {}
+
+    def _reap_exited(self) -> None:
+        now = time.time()
+        stale = [
+            sid
+            for sid, s in self._sessions.items()
+            if s.exited and s.exited_at and (now - s.exited_at) > _EXITED_TTL
+        ]
+        for sid in stale:
+            self._sessions.pop(sid, None)
+
+    async def create(
+        self,
+        agent: str,
+        command: str,
+        args: list[str] | None = None,
+        cwd: str | Path | None = None,
+        env: dict[str, str] | None = None,
+        cols: int = 80,
+        rows: int = 24,
+    ) -> AgentSession:
+        self._reap_exited()
+        session = AgentSession(
+            id=uuid.uuid4().hex[:12],
+            agent=agent,
+            command=command,
+            cwd=str(cwd) if cwd else "",
+            created_at=time.time(),
+        )
+        session.pty = await PtySession.start(
+            command,
+            args=args,
+            cwd=cwd,
+            env=env,
+            cols=cols,
+            rows=rows,
+            on_output=session._on_output,
+            on_exit=session._on_exit,
+        )
+        self._sessions[session.id] = session
+        return session
+
+    def get(self, session_id: str) -> AgentSession | None:
+        return self._sessions.get(session_id)
+
+    def list(self) -> list[dict]:
+        self._reap_exited()
+        return [
+            s.info()
+            for s in sorted(
+                self._sessions.values(), key=lambda s: s.created_at, reverse=True
+            )
+        ]
+
+    async def close(self, session_id: str) -> bool:
+        session = self._sessions.pop(session_id, None)
+        if session is None:
+            return False
+        if session.pty is not None:
+            await session.pty.close()
+        for q in list(session.subscribers):
+            q.put_nowait(None)
+        return True
+
+    async def close_all(self) -> None:
+        for sid in list(self._sessions):
+            await self.close(sid)
+
+
+# Module-level singleton (mirrors mesh_state / stream subscribers).
+manager = SessionManager()

castle-api/src/castle_api/agents.py

@@ -0,0 +1,260 @@
+"""Agent terminal UX — list agents, run/resume them in a pty over WebSocket.
+
+Castle is assistant-agnostic: it launches a configured command (or a shell) in a
+real tty and streams raw bytes to an xterm.js terminal in the browser. Sessions
+outlive their WebSocket connection, so they can be listed, resumed, and killed.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import shlex
+from urllib.parse import urlparse
+
+from fastapi import APIRouter, HTTPException, WebSocket, WebSocketDisconnect, status
+from starlette.websockets import WebSocketState
+
+from castle_api.agent_registry import (
+    default_cwd,
+    list_agent_history,
+    list_agents,
+    resolve_agent,
+    resume_argv,
+)
+from castle_api.agent_sessions import AgentSession, manager
+from castle_api.config import get_config
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/agents", tags=["agents"])
+
+# The reserved agent name that launches the user's login shell (Option A: plain
+# terminal). Not a configured agent — Castle knows nothing about what you run in it.
+TERMINAL_AGENT = "terminal"
+
+
+# --- Origin allowlist (this endpoint is an interactive shell) ---------------
+
+
+def _origin_allowed(origin: str | None) -> bool:
+    """Allow same-node/local and configured-domain origins for the WS handshake.
+
+    CORS (`*`) governs XHR, not WebSocket upgrades, so we gate the handshake
+    here. Non-browser clients (no Origin) and localhost are allowed; browser
+    origins must match the gateway domain (or `CASTLE_API_TERMINAL_ORIGINS`).
+    Set `CASTLE_API_ALLOW_ALL_ORIGINS=1` to disable the check entirely.
+    """
+    if os.environ.get("CASTLE_API_ALLOW_ALL_ORIGINS"):
+        return True
+    if not origin:
+        return True  # curl/websocat and other non-browser clients
+    host = (urlparse(origin).hostname or "").lower()
+    if host in ("localhost", "127.0.0.1", "::1"):
+        return True
+    try:
+        domain = get_config().gateway.domain
+    except Exception:
+        domain = None
+    if domain and (host == domain or host.endswith("." + domain)):
+        return True
+    extra = os.environ.get("CASTLE_API_TERMINAL_ORIGINS", "")
+    allowed = {h.strip().lower() for h in extra.split(",") if h.strip()}
+    return host in allowed
+
+
+# --- HTTP: agents + sessions ------------------------------------------------
+
+
+@router.get("")
+def get_agents() -> list[dict]:
+    """List launchable agents with availability (binary present) + default cwd."""
+    return [a.info() for a in list_agents()]
+
+
+@router.get("/sessions")
+def get_sessions() -> list[dict]:
+    """List live terminal sessions (running or recently exited)."""
+    return manager.list()
+
+
+@router.get("/history")
+async def get_history() -> list[dict]:
+    """Unified list of agents' own past sessions (from each agent's declared
+    `sessions.list_command`), newest first. Agents without the capability
+    (or with none stored) simply contribute nothing."""
+    agents = [a for a in list_agents() if a.available and a.can_list_sessions]
+    results = await asyncio.gather(*(list_agent_history(a) for a in agents))
+    rows = [row for agent_rows in results for row in agent_rows]
+    try:
+        rows.sort(key=lambda r: r["time"] or 0, reverse=True)
+    except TypeError:
+        pass
+    return rows
+
+
+@router.delete("/sessions/{session_id}")
+async def delete_session(session_id: str) -> dict:
+    """Kill a session's process group and drop it from the registry."""
+    ok = await manager.close(session_id)
+    if not ok:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="session not found"
+        )
+    return {"status": "closed", "id": session_id}
+
+
+# --- WebSocket: interactive session ----------------------------------------
+
+
+def _login_shell() -> str:
+    return os.environ.get("SHELL") or "/bin/bash"
+
+
+async def _new_session(
+    name: str, cont: bool = False, resume_session: str | None = None
+) -> AgentSession:
+    """Create a session for a named agent, or the reserved login shell.
+
+    Everything runs under the user's *interactive login* shell (`-lic`) so it
+    inherits exactly the environment a real terminal would — the systemd service
+    env is stripped down and would otherwise miss vars set in shell rc files
+    (API keys, etc.). Agents are `exec`'d so the pty's foreground process is the
+    agent itself (clean signals + exit).
+
+    ``cont`` appends the agent's ``resume_args`` (open its own picker / continue).
+    ``resume_session`` builds the agent's resume-by-id argv for a specific past
+    session. Either way castle only passes declared flags through — it never
+    reads the agent's session storage.
+    """
+    shell = _login_shell()
+    if name == TERMINAL_AGENT:
+        return await manager.create(
+            TERMINAL_AGENT, shell, args=["-l"], cwd=default_cwd()
+        )
+    agent = resolve_agent(name)
+    if agent is None:
+        raise LookupError(f"unknown agent: {name}")
+    if not agent.available:
+        raise LookupError(f"agent not installed: {name}")
+    if resume_session:
+        argv = resume_argv(agent, resume_session)
+        if argv is None:
+            raise LookupError(f"agent cannot resume by id: {name}")
+    else:
+        argv = [agent.resolved_command or agent.command, *agent.args]
+        if cont and agent.resume_args:
+            argv += agent.resume_args
+    launch = shlex.join(argv)
+    return await manager.create(
+        name,
+        shell,
+        args=["-lic", f"exec {launch}"],
+        cwd=agent.cwd,
+        env=agent.env,  # explicit per-agent overrides; the login shell sets PATH
+    )
+
+
+@router.websocket("/{name}/session")
+async def agent_session(ws: WebSocket, name: str) -> None:
+    """Run (or resume) an agent in a pty; stream it to the browser terminal.
+
+    Query param `session=<id>` resumes an existing live session (replaying its
+    scrollback); otherwise a new session is created for `name` (or the reserved
+    `terminal` shell). Disconnecting detaches — it does NOT kill the session.
+    """
+    if not _origin_allowed(ws.headers.get("origin")):
+        await ws.close(code=1008)  # policy violation
+        return
+
+    resume_id = ws.query_params.get("session")
+
+    # Resolve the target session (before accept only for hard rejects).
+    session: AgentSession | None = None
+    if resume_id:
+        session = manager.get(resume_id)
+        if session is None or not session.running:
+            await ws.accept()
+            await ws.send_json({"type": "error", "error": "session not resumable"})
+            await ws.close()
+            return
+    else:
+        cont = ws.query_params.get("mode") == "continue"
+        resume_session = ws.query_params.get("resume_session")
+        try:
+            session = await _new_session(name, cont=cont, resume_session=resume_session)
+        except LookupError as e:
+            await ws.accept()
+            await ws.send_json({"type": "error", "error": str(e)})
+            await ws.close()
+            return
+        except Exception:
+            logger.exception("failed to launch agent %s", name)
+            await ws.accept()
+            await ws.send_json({"type": "error", "error": "failed to launch"})
+            await ws.close()
+            return
+
+    assert session is not None  # narrowed by the branches above
+    await ws.accept()
+
+    # Subscribe + snapshot scrollback atomically (no await between → the output
+    # callback, which runs in this same event loop, cannot interleave), so a
+    # resumed client gets the buffer once with no duplication or gap.
+    q = session.subscribe()
+    snapshot = bytes(session.scrollback)
+
+    await ws.send_json(
+        {
+            "type": "session",
+            "id": session.id,
+            "agent": session.agent,
+            "resumed": bool(resume_id),
+        }
+    )
+    if snapshot:
+        await ws.send_bytes(snapshot)
+
+    async def pump() -> None:
+        try:
+            while True:
+                chunk = await q.get()
+                if chunk is None:  # child exited
+                    if ws.application_state == WebSocketState.CONNECTED:
+                        await ws.send_json({"type": "exit", "code": session.exit_code})
+                    break
+                if ws.application_state != WebSocketState.CONNECTED:
+                    break
+                await ws.send_bytes(chunk)
+        except Exception:
+            pass
+
+    pump_task = asyncio.create_task(pump())
+    try:
+        while True:
+            msg = await ws.receive()
+            if msg["type"] == "websocket.disconnect":
+                break
+            data = msg.get("bytes")
+            text = msg.get("text")
+            if data is not None and session.pty is not None:
+                session.pty.write(data)
+            elif text is not None and session.pty is not None:
+                try:
+                    ctrl = json.loads(text)
+                except json.JSONDecodeError:
+                    continue
+                if ctrl.get("type") == "resize":
+                    session.pty.resize(int(ctrl["cols"]), int(ctrl["rows"]))
+                elif ctrl.get("type") == "input":
+                    session.pty.write(str(ctrl.get("data", "")).encode())
+    except WebSocketDisconnect:
+        pass
+    finally:
+        pump_task.cancel()
+        session.unsubscribe(q)
+        # Intentionally do NOT close the session here — it lives on for resume.
+        # Sessions are ended via DELETE /agents/sessions/{id}, when the child
+        # exits, or by idle reaping.

castle-api/src/castle_api/main.py

@@ -12,6 +12,8 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from starlette.responses import StreamingResponse
 
+from castle_api.agent_sessions import manager as agent_session_manager
+from castle_api.agents import router as agents_router
 from castle_api.config import get_registry, settings
 from castle_api.config_editor import router as config_router
 from castle_api.deploy_routes import router as deploy_router
@@ -98,6 +100,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     if mdns_service:
         mdns_service.stop()
 
+    await agent_session_manager.close_all()
     close_all_subscribers()
 
 
@@ -123,6 +126,7 @@ app.include_router(secrets_router)
 app.include_router(services_router)
 app.include_router(programs_router)
 app.include_router(deploy_router)
+app.include_router(agents_router)
 
 
 @app.get("/health")

castle-api/src/castle_api/pty_session.py

@@ -0,0 +1,156 @@
+"""Run a command inside a pseudo-terminal and stream it over asyncio.
+
+Castle stays assistant-agnostic: it just launches a command in a real TTY and
+shunts raw bytes both ways. The command's own TUI (an agent CLI, a shell, …)
+renders exactly as it would in a terminal — Castle never parses its output.
+
+The PTY master is a raw OS fd, not an asyncio stream, so we watch it with
+``loop.add_reader`` (no extra thread, no busy-poll). Output is delivered to an
+``on_output`` callback (a fan-out layer above can broadcast it to any number of
+attached clients and buffer scrollback), so a session's lifetime is decoupled
+from any one WebSocket connection. Keystroke writes are tiny, so a plain
+``os.write`` is fine.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import fcntl
+import os
+import pty
+import signal
+import struct
+import termios
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+def _set_winsize(fd: int, rows: int, cols: int) -> None:
+    """Apply a terminal window size via TIOCSWINSZ (struct winsize)."""
+    winsize = struct.pack("HHHH", rows, cols, 0, 0)
+    fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
+
+
+@dataclass
+class PtySession:
+    """A child process attached to a pty master.
+
+    Output is pushed to ``on_output(bytes)`` as it arrives and ``on_exit()`` is
+    fired once when the child closes the tty. Both run in the event-loop thread.
+    """
+
+    proc: asyncio.subprocess.Process
+    master_fd: int
+    on_output: Callable[[bytes], None] | None = None
+    on_exit: Callable[[], None] | None = None
+    cols: int = 80
+    rows: int = 24
+    _closed: bool = field(default=False)
+
+    @classmethod
+    async def start(
+        cls,
+        command: str,
+        args: list[str] | None = None,
+        cwd: str | Path | None = None,
+        env: dict[str, str] | None = None,
+        cols: int = 80,
+        rows: int = 24,
+        on_output: Callable[[bytes], None] | None = None,
+        on_exit: Callable[[], None] | None = None,
+    ) -> PtySession:
+        """Spawn ``command args`` on a fresh pty, sized ``cols`` x ``rows``."""
+        master_fd, slave_fd = pty.openpty()
+        # Size the tty before exec so the first TUI render lays out correctly.
+        _set_winsize(master_fd, rows, cols)
+
+        proc_env = {**os.environ, "TERM": "xterm-256color"}
+        if env:
+            proc_env.update(env)
+
+        proc = await asyncio.create_subprocess_exec(
+            command,
+            *(args or []),
+            stdin=slave_fd,
+            stdout=slave_fd,
+            stderr=slave_fd,
+            cwd=str(cwd) if cwd else None,
+            env=proc_env,
+            start_new_session=True,  # own session/group → controlling tty + killpg
+            close_fds=True,
+        )
+        os.close(slave_fd)  # parent keeps only the master
+        os.set_blocking(master_fd, False)
+
+        session = cls(proc=proc, master_fd=master_fd, cols=cols, rows=rows)
+        if on_output is not None:
+            session.on_output = on_output
+        if on_exit is not None:
+            session.on_exit = on_exit
+        asyncio.get_running_loop().add_reader(master_fd, session._drain_master)
+        return session
+
+    def _drain_master(self) -> None:
+        """Reader callback: pull available bytes off the master and fan them out."""
+        try:
+            data = os.read(self.master_fd, 65536)
+        except (BlockingIOError, InterruptedError):
+            return
+        except OSError:
+            data = b""  # slave closed → child gone
+        if not data:
+            try:
+                asyncio.get_running_loop().remove_reader(self.master_fd)
+            except (ValueError, OSError):
+                pass
+            if self.on_exit is not None:
+                self.on_exit()
+            return
+        if self.on_output is not None:
+            self.on_output(data)
+
+    def write(self, data: bytes) -> None:
+        """Forward keystroke bytes to the child's tty."""
+        try:
+            os.write(self.master_fd, data)
+        except OSError:
+            pass
+
+    def resize(self, cols: int, rows: int) -> None:
+        """Resize the tty (browser terminal changed dimensions)."""
+        self.cols, self.rows = cols, rows
+        try:
+            _set_winsize(self.master_fd, rows, cols)
+        except OSError:
+            pass
+
+    @property
+    def running(self) -> bool:
+        return self.proc.returncode is None
+
+    async def close(self) -> None:
+        """Kill the whole process group and release the master fd (idempotent)."""
+        if self._closed:
+            return
+        self._closed = True
+        try:
+            asyncio.get_running_loop().remove_reader(self.master_fd)
+        except (ValueError, OSError):
+            pass
+        if self.proc.returncode is None:
+            try:
+                os.killpg(os.getpgid(self.proc.pid), signal.SIGTERM)
+            except ProcessLookupError:
+                pass
+            try:
+                await asyncio.wait_for(self.proc.wait(), timeout=3)
+            except asyncio.TimeoutError:
+                try:
+                    os.killpg(os.getpgid(self.proc.pid), signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+        try:
+            os.close(self.master_fd)
+        except OSError:
+            pass