breakglass UI v2: attachable sessions (tmux model) + mobile-first redesign

Full audit-driven rework. Keeps the proven SSE-translation + verb logic; everything else upgraded for phone-primary use.

Backend — server owns the session, clients attach (Viktor's tmux idea):
- session.py: SessionManager + Session with an event log, subscriber pub/sub, and turns that run DETACHED (keep going if the client disconnects).
- GET /api/session/{id}/stream = attach (SSE): replays the transcript then tails live; per-event id: lines so an EventSource auto-reconnect resumes from Last-Event-ID (free re-attach). POST /{id}/prompt starts a detached turn; POST /{id}/cancel = Stop. Replaces the old one-shot /api/chat.
- agent_session trimmed to the argv + translate_event helpers; 21 new/updated tests (replay, Last-Event-ID resume, broadcast, detached turn, resume, cancel, routes) — 53 green.

Frontend — mobile-first via the frontend-design skill (emergency-console aesthetic):
- EventSource attach (native auto-reconnect, zero client reconnect logic); transcript.js folds events->messages with id-dedupe so replays never double-render (30 unit assertions).
- Installable PWA: manifest + icons (wrench/break-glass mark) + apple-mobile-web-app meta + theme-color; viewport-fit=cover + safe-area; 100dvh; 16px composer (no iOS zoom).
- One-tap diagnosis presets (Triage / Memory-OOM / Disk / Services / QEMU-wedged) mapped to the devvm's real failure modes; Stop button while a turn runs.
- Foldable VM-control sheet, cycle the dominant recovery action w/ confirm, output capped 46vh.
- a11y: fixed --ink-faint contrast 3.6:1 -> 6.1:1 (WCAG AA); >=44px tap targets. Deleted the obsolete fetch-reader sse.js (EventSource replaces it).

Verified: 53 backend tests + 30 transcript assertions; Playwright @390x844 (input on-screen y=721-821, presets/sheet/fold/cap); local integration smoke vs the real backend (attach->caught-up, 404, verbs, PWA served).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

app/breakglass/agent_session.py

@@ -1,26 +1,13 @@
-"""Drive the breakglass Claude agent and stream its work to the browser.
+"""Claude CLI argv + stream-json → UI-event translation for the breakglass agent.
 
-Each chat turn runs ``claude -p --output-format stream-json`` in the session's
-persistent workspace; the first turn opens the session with ``--session-id`` and
-later turns ``--resume`` it, so the conversation has memory across turns. The
-CLI's JSON events are translated to a small, stable SSE vocabulary the UI
-renders (``session`` / ``text`` / ``tool`` / ``result`` / ``error``) — we do not
-leak the raw event firehose to the client.
-
-Subprocesses use ``asyncio.create_subprocess_exec`` (list argv, no shell): the
-prompt and ids are argv elements, never interpreted by a shell.
+The session lifecycle (running turns, attaching clients) lives in ``session.py``;
+this module is just the two helpers it builds on:
+  * ``_turn_argv`` — the no-shell list argv for one ``claude -p`` turn.
+  * ``translate_event`` — map a raw stream-json event to the small UI vocabulary
+    (session / text / tool / result), dropping the hook/thinking-token noise.
 """
-import asyncio
-import json
-import os
-from subprocess import PIPE
-from typing import AsyncIterator
-
 from . import config
 
-# Sessions we've already opened (so the next turn resumes instead of re-creating).
-_started: set[str] = set()
-
 
 def _turn_argv(session_id: str, prompt: str, resume: bool, model: str) -> list[str]:
     argv = [
@@ -66,7 +53,7 @@ def translate_event(obj: dict) -> dict | None:
                 })
         if not events:
             return None
-        # The server flattens a "batch" into individual SSE frames.
+        # The session log flattens a "batch" into individual events.
         return events[0] if len(events) == 1 else {"kind": "batch", "events": events}
 
     if etype == "result":
@@ -78,68 +65,3 @@ def translate_event(obj: dict) -> dict | None:
         }
 
     return None
-
-
-async def run_turn(
-    session_id: str, prompt: str, model: str | None = None
-) -> AsyncIterator[dict]:
-    """Run one chat turn, yielding translated UI events as they arrive."""
-    resume = session_id in _started
-    model = model or config.DEFAULT_MODEL
-    workspace = os.path.join(config.SESSIONS_DIR, session_id)
-    os.makedirs(workspace, exist_ok=True)
-
-    argv = _turn_argv(session_id, prompt, resume, model)
-    proc = await asyncio.create_subprocess_exec(
-        *argv, cwd=workspace, stdout=PIPE, stderr=PIPE,
-    )
-    _started.add(session_id)
-    assert proc.stdout is not None and proc.stderr is not None
-
-    try:
-        async def _pump() -> AsyncIterator[dict]:
-            async for raw in proc.stdout:
-                line = raw.decode(errors="replace").strip()
-                if not line:
-                    continue
-                try:
-                    obj = json.loads(line)
-                except json.JSONDecodeError:
-                    continue
-                ev = translate_event(obj)
-                if ev is None:
-                    continue
-                if ev.get("kind") == "batch":
-                    for sub in ev["events"]:
-                        yield sub
-                else:
-                    yield ev
-
-        async for ev in _with_timeout(_pump(), config.TURN_TIMEOUT_SECONDS):
-            yield ev
-    except asyncio.TimeoutError:
-        proc.kill()
-        await proc.wait()
-        yield {"kind": "error", "error": f"turn timed out after {config.TURN_TIMEOUT_SECONDS}s"}
-        return
-
-    await proc.wait()
-    if proc.returncode not in (0, None):
-        err = (await proc.stderr.read()).decode(errors="replace")
-        yield {"kind": "error", "error": err.strip()[:500] or f"exit {proc.returncode}"}
-
-
-async def _with_timeout(agen: AsyncIterator[dict], timeout: float) -> AsyncIterator[dict]:
-    """Yield from an async generator but raise TimeoutError if the WHOLE turn
-    exceeds ``timeout`` seconds (a wedged agent shouldn't stream forever)."""
-    loop = asyncio.get_event_loop()
-    deadline = loop.time() + timeout
-    it = agen.__aiter__()
-    while True:
-        remaining = deadline - loop.time()
-        if remaining <= 0:
-            raise asyncio.TimeoutError
-        try:
-            yield await asyncio.wait_for(it.__anext__(), timeout=remaining)
-        except StopAsyncIteration:
-            return