routes: detached generator + per-session buffer + /resume endpoint

Mid-session resume support. When the phone backgrounds (OS suspends
JS, SSE socket drops) the server used to cancel the generator —
asyncio.CancelledError propagated through stream_troubleshoot and
the model run died. On foreground the chat had no real recovery
beyond "Start a new chat" + the Try-Again button shipped in
b2a7dcb.

New flow:
  - POST /troubleshoot mints (or reuses) a SessionState and starts
    the bridge in an asyncio.create_task that's detached from the
    SSE handler. The handler is now a buffer reader.
  - GET /troubleshoot/resume?session_id=X&from=N reattaches to the
    same task's output by streaming buffered events with seq > N,
    then awaits the per-session asyncio.Condition for new events.
  - Each SSE frame carries an `id: <seq>` line so the client's
    EventSource records lastEventId; the client supplies that as
    `?from=` on the next reconnect.
  - Buffer cap is 500 events per session; on overflow the oldest
    is dropped + dropped_count increments. When a resume's `from`
    is older than the oldest buffered seq, the SSE stream injects
    a synthetic `thought` event explaining the gap (no schema
    change — reusing thought for a transport concern per advisor).

Concurrency:
  - Last-wins consumer policy via SessionState.consumer_generation.
    Each new SSE handler bumps the token; older consumer loops
    check the token and exit cleanly. Auto-resume on foreground
    can't end up with two SSE streams pumping duplicate frames.
  - POST /troubleshoot on a session with a still-running task
    returns 409 (callers must use /resume instead). Once the prior
    task is done, POST starts a fresh generator and resets the
    buffer.

SessionState additions:
  - event_buffer, next_seq, dropped_count, generator_done,
    generator_task, consumer_generation, cond
  - append_event(): single writer entry point (notify_all after
    append)
  - mark_done(): idempotent termination + wake-all

Test parsers across 4 files updated to handle the new
`id: N\\ndata: {...}` SSE block format. All 256 tests pass.

Client-side wiring (AsyncStorage persistence + AppState
foreground subscriber) ships in apps/box separately.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: ehsan6sha

src/routes/troubleshoot.py

@@ -7,17 +7,28 @@
 C5 adds:
   - POST /troubleshoot/user-reply (matches Phase 11 contract)
   - POST /troubleshoot/phone-context (matches Phase 11 contract)
+
+2026-05-28 resume support:
+  - POST /troubleshoot now starts the generator as a detached
+    `asyncio.create_task`; the SSE handler is just a buffer reader.
+    Disconnecting the SSE consumer no longer cancels the model run.
+  - GET /troubleshoot/resume?session_id=X&from=N reattaches to the
+    same task's buffered output. App-side persistence (lastEventSeq
+    in AsyncStorage) plus an AppState foreground subscriber makes
+    background-then-resume seamless.
 """
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
+from typing import AsyncIterator
 
-from fastapi import APIRouter, HTTPException, Request, Response
+from fastapi import APIRouter, HTTPException, Query, Request, Response
 from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel, ConfigDict, Field
 
-from src.session.manager import sanitize_for_log
+from src.session.manager import SessionState, sanitize_for_log
 from src.session.tool_call_loop import stream_troubleshoot
 
 
@@ -61,6 +72,124 @@ def _error_body(code: str, detail: str = "") -> dict:
     return out
 
 
+async def _drive_generator_into_buffer(
+    session: SessionState,
+    backend,
+    tool_executor,
+    validator,
+    prompt: str,
+) -> None:
+    """Run the bridge generator detached from any SSE consumer, writing
+    each event into the session's buffer. SSE consumers come and go via
+    `_stream_from_buffer`; this task is the SOLE writer. Survives
+    consumer disconnect — that's the whole point of the resume feature.
+
+    On exception we still try to append a synthetic error event so the
+    consumer can render the failure instead of staring at a hanging
+    chat. `mark_done` is in the finally so even hard exceptions release
+    the consumers' `cond.wait()`.
+    """
+    backend_handles_tools = getattr(backend, "consumes_tool_results", False)
+    try:
+        backend_events = backend.run_troubleshoot(
+            prompt=prompt,
+            session_id=session.session_id,
+        )
+        async for event in stream_troubleshoot(
+            backend_events, tool_executor, validator,
+            session=session,
+            backend_handles_tools=backend_handles_tools,
+        ):
+            await session.append_event(event)
+    except asyncio.CancelledError:
+        # Container shutdown or explicit cancel — propagate after
+        # marking done so consumers exit.
+        await session.mark_done()
+        raise
+    except Exception as e:  # noqa: BLE001
+        logger.exception("background generator failed session=%s", session.session_id)
+        try:
+            await session.append_event({
+                "type": "error",
+                "code": "INTERNAL_ERROR",
+                "message": str(e)[:200],
+                "recoverable": False,
+            })
+        except Exception:
+            pass  # last-ditch; buffer write itself shouldn't crash
+    finally:
+        await session.mark_done()
+
+
+async def _stream_from_buffer(
+    session: SessionState,
+    from_seq: int,
+) -> AsyncIterator[str]:
+    """Yield SSE-formatted strings from the session's event buffer
+    starting at `from_seq`, blocking on `session.cond` for new events
+    until the generator marks itself done OR a newer consumer claims
+    this session (last-wins policy).
+
+    SSE `id:` field carries the seq number so the client's
+    EventSource records it on lastEventId; the client persists this to
+    AsyncStorage and supplies it as `?from=` on the next reconnect.
+
+    Truncation marker: if `from_seq` is less than the oldest buffered
+    seq (the consumer was away long enough for events to fall off the
+    cap), inject a synthetic `thought` event with the dropped count.
+    Per advisor input we don't grow the SSE schema for a flow-control
+    concern — a `thought` event with the marker text is enough; the
+    chat surface renders it as italic gray prose."""
+    my_generation = session.consumer_generation
+    last_yielded_seq = from_seq - 1
+
+    # Truncation detection — fires when the consumer is asking for an
+    # event older than what's still in the buffer. Includes the
+    # from_seq=0 case (fresh resume after the head of the buffer
+    # already overflowed).
+    if (
+        session.event_buffer
+        and session.event_buffer[0][0] > from_seq
+    ):
+        gap = session.event_buffer[0][0] - from_seq
+        marker = {
+            "type": "thought",
+            "payload": (
+                f"[resume] {gap} earlier event(s) dropped from the on-device "
+                f"buffer (cap 500). Newer events from this session continue below."
+            ),
+        }
+        # Emit the marker with a special id=-1 (no real seq) so the
+        # client's lastEventId tracker doesn't try to use it as a
+        # resume offset later.
+        yield f"id: -1\ndata: {json.dumps(marker, separators=(',', ':'))}\n\n"
+
+    while True:
+        # Yield any buffered events strictly newer than last_yielded.
+        # Materialize so we don't hold the buffer reference across the
+        # await (the buffer is mutated by the producer task).
+        new_events = [
+            (s, e) for s, e in session.event_buffer if s > last_yielded_seq
+        ]
+        for s, e in new_events:
+            yield f"id: {s}\ndata: {json.dumps(e, separators=(',', ':'))}\n\n"
+            last_yielded_seq = s
+
+        # Check exit conditions BEFORE waiting (covers the case where
+        # mark_done already fired or a new consumer already took over).
+        if session.generator_done:
+            return
+        if session.consumer_generation != my_generation:
+            # Last-wins: another /troubleshoot or /resume call took
+            # over this session. Quietly exit so the network frame
+            # stream isn't doubled. The newer consumer reads the same
+            # buffer and picks up from from_seq.
+            return
+
+        async with session.cond:
+            await session.cond.wait()
+
+
 @router.post("/troubleshoot")
 async def troubleshoot(req: TroubleshootRequest, request: Request) -> Response:
     backend = request.app.state.backend
@@ -69,7 +198,7 @@ async def troubleshoot(req: TroubleshootRequest, request: Request) -> Response:
     session_mgr = request.app.state.session_manager
 
     # Resolve session: caller-supplied wins; else mint new.
-    session = None
+    session: SessionState | None = None
     if req.session_id:
         session = session_mgr.get(req.session_id)
         if session is None:
@@ -80,30 +209,47 @@ async def troubleshoot(req: TroubleshootRequest, request: Request) -> Response:
     else:
         session = session_mgr.create()
 
-    backend_events = backend.run_troubleshoot(
-        prompt=req.prompt,
-        session_id=session.session_id,
-    )
+    # If this session already has a running generator (caller retried
+    # POST on the same session_id while a previous task is still
+    # active), reject. Resume via GET /troubleshoot/resume instead so
+    # we don't spawn duplicate writers into one buffer.
+    if (
+        session.generator_task is not None
+        and not getattr(session.generator_task, "done", lambda: True)()
+        and not session.generator_done
+    ):
+        return JSONResponse(
+            status_code=409,
+            content=_error_body(
+                "session_already_active",
+                "use GET /troubleshoot/resume?session_id=...&from=N to reattach",
+            ),
+        )
 
-    backend_handles_tools = getattr(backend, "consumes_tool_results", False)
+    # Reset per-session buffer state for the new generator. We keep the
+    # SessionState (so phone_context, reply_queue, etc. survive) but
+    # wipe the conversation buffer so the new prompt starts at seq 0.
+    session.event_buffer = []
+    session.next_seq = 0
+    session.dropped_count = 0
+    session.generator_done = False
+    session.consumer_generation += 1
+    my_generation = session.consumer_generation
+
+    session.generator_task = asyncio.create_task(
+        _drive_generator_into_buffer(
+            session, backend, tool_executor, validator, req.prompt,
+        ),
+        name=f"blox-ai-generator-{session.session_id}",
+    )
 
     async def sse_stream():
+        # consumer_generation was bumped above; capture it for the
+        # last-wins check in _stream_from_buffer.
+        session.consumer_generation = my_generation
         try:
-            async for event in stream_troubleshoot(
-                backend_events, tool_executor, validator,
-                session=session,
-                backend_handles_tools=backend_handles_tools,
-            ):
-                yield f"data: {json.dumps(event, separators=(',', ':'))}\n\n"
-        except Exception:
-            logger.exception("unexpected bridge failure")
-            fallback = {
-                "type": "error",
-                "code": "INTERNAL_ERROR",
-                "message": "unexpected bridge failure",
-                "recoverable": False,
-            }
-            yield f"data: {json.dumps(fallback, separators=(',', ':'))}\n\n"
+            async for chunk in _stream_from_buffer(session, from_seq=0):
+                yield chunk
         finally:
             # Slide TTL on stream completion so a session that just
             # finished a turn doesn't expire instantly.
@@ -119,6 +265,54 @@ async def sse_stream():
     )
 
 
+@router.get("/troubleshoot/resume")
+async def troubleshoot_resume(
+    request: Request,
+    session_id: str = Query(min_length=1, max_length=128),
+    from_seq: int = Query(alias="from", default=0, ge=0),
+) -> Response:
+    """Reattach to a /troubleshoot session's existing generator output.
+    Returns 404 if the session was evicted (TTL, LRU, or container
+    restart); the client clears its persisted state on 404 + offers
+    Start-new-chat.
+
+    Replays buffered events newer than `from_seq`, injects a
+    truncation marker if `from_seq` is older than the oldest buffered
+    event, then blocks on the session's cond for new events until the
+    generator marks itself done or a newer consumer takes over.
+
+    Idempotent: multiple consumers can call /resume; last-wins kicks
+    the older consumer (it exits cleanly without re-emitting events).
+    """
+    session_mgr = request.app.state.session_manager
+    session = session_mgr.get(session_id)
+    if session is None:
+        return JSONResponse(
+            status_code=404,
+            content=_error_body("session_not_found"),
+        )
+    # Slide TTL — resume IS user activity.
+    session_mgr.touch(session.session_id)
+    # Last-wins: bump generation so any prior consumer's loop exits.
+    session.consumer_generation += 1
+
+    async def sse_stream():
+        try:
+            async for chunk in _stream_from_buffer(session, from_seq=from_seq):
+                yield chunk
+        finally:
+            session_mgr.touch(session.session_id)
+
+    return StreamingResponse(
+        sse_stream(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
 @router.post("/troubleshoot/user-reply")
 async def user_reply(req: UserReplyRequest, request: Request) -> Response:
     """Phase 11 contract: the app submits this when the user answers a

src/session/manager.py

@@ -27,6 +27,14 @@
 
 DEFAULT_TTL_SEC = 30 * 60  # 30 minutes
 DEFAULT_MAX_SESSIONS = 50
+# Resume support — per-session event buffer cap. ~1 KB/event × 500 cap ×
+# 50 sessions ≈ 25 MB worst-case; comfortable on a device with 7-8 GB
+# RAM. On overflow we drop the oldest event AND inject a synthetic
+# `thought` truncation marker (see SessionState.append_event) so the
+# resuming consumer sees a plain-text note about the gap. Reusing
+# `thought` avoids touching the SSE schema for a flow-control concern
+# (advisor input: don't grow the schema for transport metadata).
+DEFAULT_BUFFER_CAP = 500
 
 
 @dataclass
@@ -52,6 +60,58 @@ class SessionState:
     # for the executor; the HMAC token binds to action_id, the
     # dispatch table needs action_name + args.
     issued_recommendations: dict[str, dict] = field(default_factory=dict)
+    # ---- Resume support (added 2026-05-28) -----------------------------
+    # event_buffer holds (seq, event) tuples in arrival order. seq is
+    # monotonic per-session — even on truncation we keep counting up so
+    # gaps in seq let the consumer detect dropped events.
+    event_buffer: list[tuple[int, dict]] = field(default_factory=list)
+    next_seq: int = 0
+    # Total events dropped from the head of the buffer due to cap.
+    # Cumulative across the session lifetime. Surfaced to the consumer
+    # as the marker payload when a resume's `from` is older than the
+    # oldest buffered seq.
+    dropped_count: int = 0
+    # Set True once the generator task finishes (verdict reached, error,
+    # or session_cancelled). Consumers exit their wait loop on this.
+    generator_done: bool = False
+    # The detached generator task. Created on the first /troubleshoot
+    # POST for this session; subsequent /resume calls reattach to the
+    # same task's output (via the buffer). NEVER awaited from the SSE
+    # handler — that's the whole point of the resume feature (SSE
+    # consumer disconnect must NOT cancel the generator).
+    generator_task: object | None = None
+    # Last-wins consumer policy: each new SSE consumer increments this
+    # token; consumers loop while their captured token equals the
+    # current value, exit when a newer consumer has taken over. Avoids
+    # multiple SSE streams writing duplicate frames to the network.
+    consumer_generation: int = 0
+    # Producer/consumer wake — asyncio.Condition (not Event) per
+    # advisor input: Event has the "lost wake between clear and wait"
+    # race. Condition's notify_all + acquire pattern is race-free for
+    # multiple consumers, and the migration from single-consumer to
+    # last-wins doesn't require touching wake logic.
+    cond: asyncio.Condition = field(default_factory=asyncio.Condition)
+
+    async def append_event(self, event: dict) -> None:
+        """Append an event to the buffer + wake any waiting consumer.
+        Drops the oldest event on overflow (incrementing dropped_count
+        so the next resume can synthesize a truncation marker)."""
+        seq = self.next_seq
+        self.next_seq += 1
+        self.event_buffer.append((seq, event))
+        if len(self.event_buffer) > DEFAULT_BUFFER_CAP:
+            self.event_buffer.pop(0)
+            self.dropped_count += 1
+        async with self.cond:
+            self.cond.notify_all()
+
+    async def mark_done(self) -> None:
+        """Mark the generator as finished + wake all consumers so they
+        exit their stream loop. Idempotent — safe to call from the
+        generator wrapper's try/except/finally branches."""
+        self.generator_done = True
+        async with self.cond:
+            self.cond.notify_all()
 
 
 class SessionManager:

tests/test_c6_routes.py

@@ -23,8 +23,10 @@ def client_with_tmp_feedback_log(client, tmp_path):
 
 def _open_session(client) -> str:
     r = client.post("/troubleshoot", json={"prompt": "just diagnose"})
-    first = r.text.split("\n\n")[0]
-    return json.loads(first[len("data: "):])["session_id"]
+    first_block = r.text.split("\n\n")[0]
+    # 2026-05-28 resume support: SSE events now have `id:` then `data:`.
+    data_line = next(L for L in first_block.split("\n") if L.startswith("data: "))
+    return json.loads(data_line[len("data: "):])["session_id"]
 
 
 def test_feedback_happy_path_writes_log_line(client_with_tmp_feedback_log):