Claude Agent SDK in Production, Part 9: Durable Streams: Survive the Refresh

I did the homework Part 8 assigned. Started a real analysis, watched two badges light up, and hit refresh at the nine-second mark. Blank page. New empty desk, cheerful sample chips, as if nothing had ever happened. The server logs tell the other half: the agent kept working, the audit log kept recording, the answer arrived complete at second twenty-six, and the server wrote it to a socket nobody held anymore. The work was perfect. The delivery address had ceased to exist. Root cause, named plainly: the agent's lifetime is chained to one HTTP response, and it has been since Part 2, because POST /chat returns the stream, so whoever holds that response IS the audience, and when their tab dies the pipe dies with it. Today we break that chain. By the end of this page a refresh costs you 130 milliseconds of blank screen and nothing else: not the transcript, not the live tail, not even a pending approval card.

That screenshot looks like Part 7. That's the point. Nothing about the product changed; what changed is that the conversation you're looking at was torn down and rebuilt from disk halfway through, and neither the agent nor the approval machinery noticed. This is the deepest architecture lesson in the series, it's how the production reference app streams, and it pays three debts at once: Part 3's Stop button that only hung up the phone, Part 5's apology that a turn in flight still dies with the tab, and Part 7's confession that approval events lived only on the wire. All three, on camera, today.

The autopsy: one pipe, one audience

Look at what /chat has returned since Part 2 designed the envelope: a StreamingResponse. The agent runs inside that response's generator. Close the response and FastAPI cancels the generator, the cleanup tears down the client, and everything downstream of your question stops mattering. A refresh is nothing but the rudest way to close a response.

The fix is a separation you've seen in every restaurant: the kitchen keeps cooking even if the waiter changes shifts. Running the agent becomes a background job that talks to nobody in particular; watching the agent becomes a cheap, disposable subscription anyone can open, drop, and reopen. Between the two sits the part that makes it durable: a log. Every event the agent produces gets written down before anyone gets to see it, so "catching up" and "watching live" become the same operation with different starting points. The flight recorder doesn't care who's in the cockpit.

The claim ticket

Here's the whole decoupling, visible in one endpoint. POST /chat stops streaming:

@app.post("/chat")
async def chat(request: ChatRequest) -> dict:
    """Start a run. Note what this does NOT return: a stream. The agent's
    lifetime now belongs to the worker; this response is a claim ticket."""
    workspace_id = request.workspace_id or create_workspace()
    workspace = workspace_path(workspace_id)
    request_id = uuid.uuid4().hex
    start_request(request_id, workspace, workspace_id, request.session_id, request.message)
    return {
        "request_id": request_id,
        "workspace_id": workspace_id,
        "stream_url": f"/stream/{request_id}",
    }

It returns in milliseconds, and it returns a claim ticket: a request_id and the address where this run's events will be available, forever, to anyone who asks. The session_id flow from Part 5 is untouched; the ticket is about one turn, not the conversation. Hold onto the shape of this response, because the whole frontend rewrite later amounts to "keep the ticket somewhere a refresh can't reach."

The worker owns the run

start_request lives in a new file, app/runner.py, and the first thing in it is a dictionary with a secret second job:

@dataclass
class RunningRequest:
    """One live run: the task doing the work, the client that can
    interrupt it, and the bridge holding its unanswered cards."""
    task: asyncio.Task
    client: ClaudeSDKClient
    bridge: ApprovalBridge


# Every run currently in flight. This dict is ALSO doing invisible,
# load-bearing work: asyncio holds only weak references to tasks, so a
# create_task() result nobody stores can be garbage-collected mid-run.
# Keeping the task here is what keeps it alive.
RUNNING: dict[str, RunningRequest] = {}

Read that comment twice, because it's the classic asyncio.create_task trap and it fails in the cruelest way. The event loop does not keep your tasks alive; it holds weak references, and a task nobody stores can be garbage-collected mid-run, silently, sometimes, depending on memory pressure: a background agent that works in every test and occasionally evaporates in production. We need RUNNING anyway (the cancel endpoint has to find the client), so the reference-keeping comes free, but if you ever fire-and-forget a task elsewhere, store it on purpose.

Now the spawn itself. Two nested functions: a sink and a worker:

def start_request(
    request_id: str,
    workspace: Path,
    workspace_id: str,
    session_id: str | None,
    message: str,
) -> None:
    """Spawn the worker for one run and return immediately."""
    seq = 0

    async def emit(event: dict) -> None:
        """The single sink for this run. Log first (the truth), then
        broadcast (the courtesy). Every event gets the next seq."""
        nonlocal seq
        seq += 1
        await eventlog.append(request_id, seq, event)
        hub.publish(request_id, seq, event)

    bridge = ApprovalBridge(workspace_id, emit)
    client = ClaudeSDKClient(
        options=build_options(workspace, session_id, bridge.gate)
    )

emit is the only door out of this run, and it does two things in a fixed order: write the event to the log with the next sequence number, then hand a copy to whoever is currently listening. Log first. The log is the truth; the broadcast is a courtesy. Notice who else got handed emit: the approval bridge from Part 7, which used to drop its cards into a per-response queue. From this line on, approval events are logged like everything else, and that one change is what closes Part 7's honesty note about cards living only on the wire. We'll collect that payoff properly in a few sections.

The worker is the same pipeline you've been growing since Part 2, wearing a new coat:

    async def run() -> None:
        try:
            await client.connect()
            await client.query(message)
            stream = with_artifacts(translate(client.receive_response()), workspace)
            async for event in stream:
                if event["type"] == "session_start":
                    event = {**event, "workspace_id": workspace_id}
                await emit(event)
        except Exception as exc:  # noqa: BLE001 - failures become logged events
            await emit({"type": "error", "message": str(exc)})
        finally:
            bridge.deny_all()  # a card nobody can answer anymore is a deny
            try:
                await client.disconnect()
            finally:
                hub.close_request(request_id)
                RUNNING.pop(request_id, None)

    hub.open_request(request_id)
    RUNNING[request_id] = RunningRequest(
        task=asyncio.create_task(run()), client=client, bridge=bridge
    )

connect, query, receive_response, the translator, the artifact watcher: all unchanged since Parts 2 through 7. What changed is the audience. Nobody holds a pipe to this function. It cooks whether anyone watches or not, and one behavior quietly flipped because of that: in Part 8, a browser hanging up denied every pending approval card, because the response's teardown was the only cleanup we had. Now a dead browser means nothing. The run continues, the cards keep waiting (Part 7's 120-second timeout is still the backstop), and deny_all fires only when the run itself ends or gets cancelled. Disconnection stopped being an opinion about your intentions.

Checkpoint: POST /chat answers instantly with a ticket, and a worker runs the agent with no audience required. Right now every event vanishes into emit. Time to build the two places it goes.

The flight recorder

SQLite finally enters the series, and it enters on a leash: one table, one hand-written migration, no ORM.

# The one migration, by hand. seq restarts at 1 for every request, so
# (request_id, seq) is the natural primary key, and "replay from where I
# left off" is a WHERE clause.
SCHEMA = """
CREATE TABLE IF NOT EXISTS events (
    request_id TEXT NOT NULL,
    seq        INTEGER NOT NULL,
    type       TEXT NOT NULL,
    payload    TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    PRIMARY KEY (request_id, seq)
);
"""

The payload column stores the exact JSON dict the translator produced, envelope and all. We're not modeling the events, we're recording them; the type column exists only so you can grep the log without parsing JSON. The whole API is two functions:

async def append(request_id: str, seq: int, event: dict) -> None:
    """One event, one row. Committed before anyone gets to see it: the
    log is the source of truth, the broadcast is a courtesy copy."""
    await _db.execute(
        "INSERT INTO events (request_id, seq, type, payload) VALUES (?, ?, ?, ?)",
        (request_id, seq, event["type"], json.dumps(event)),
    )
    await _db.commit()


async def replay(request_id: str, after_seq: int = 0) -> list[tuple[int, dict]]:
    """Everything this run has said after a given point, in order."""
    cursor = await _db.execute(
        "SELECT seq, payload FROM events WHERE request_id = ? AND seq > ? ORDER BY seq",
        (request_id, after_seq),
    )
    rows = await cursor.fetchall()
    return [(seq, json.loads(payload)) for seq, payload in rows]

aiosqlite keeps the writes off the event loop's back, a lifespan hook in main.py opens the connection at boot, and that's the entire database story. WHERE seq > after_seq is the sentence that makes streams resumable; everything else in this part is delivery.

The hub carries the present

The log remembers the past. For the present, a subscriber needs the events that arrive while they're watching, and that's a broadcast hub: the second new file, app/hub.py, and it's barely a file at all. A dict of queues per request_id, publish fans an event out to every queue, subscribe adds a queue, close_request wakes everyone with a None so they know to hang up. It holds nothing durable and remembers nothing; if the hub ever loses an event, the log already has it.

def publish(request_id: str, seq: int, event: dict) -> None:
    """Hand one event to every current subscriber. Fire and forget: the
    log has already committed it, so a subscriber who misses it (or shows
    up later) recovers by replaying."""
    for queue in _subscribers.get(request_id, set()):
        queue.put_nowait((seq, event))


def close_request(request_id: str) -> None:
    """The worker is done: wake every subscriber with the hang-up signal
    and forget the request. The log is the only thing that outlives this."""
    _live.discard(request_id)
    for queue in _subscribers.pop(request_id, set()):
        queue.put_nowait(None)


def subscribe(request_id: str) -> asyncio.Queue:
    queue: asyncio.Queue = asyncio.Queue()
    _subscribers.setdefault(request_id, set()).add(queue)
    return queue

One design question hides in here, the kind that pages you at 2 a.m.: a new subscriber needs the past (from the log) and the present (from the hub). In which order? Replay first and subscribe second, and an event that lands between the two is gone for that subscriber, invisibly, forever. So the rule is subscribe first, replay second: open your queue before you read the log, and accept that an event might now arrive twice, once from the replay and once from the queue. Duplicates are solvable, because every event carries its seq; you drop what you've already seen. A lost event is not solvable. At-least-once plus deduplication beats at-most-once every time someone's watching.

The dumb pipe

Now the endpoint that replaces streaming-from-/chat, and the reason this part uses the word "dumb" with affection. GET /stream/{request_id} knows nothing about agents, models, or approvals. It reads a log, drains a queue, and never finds out what any of it meant:

@app.get("/stream/{request_id}")
async def stream(request_id: str, request: Request) -> StreamingResponse:
    """The dumb pipe: replay-then-follow. It knows nothing about agents;
    it reads a log and drains a queue. Open it twice, close it, reopen
    it mid-run: the worker never notices."""
    # A reconnecting browser sends the last SSE id it saw; replay resumes
    # from there instead of repeating the whole run.
    last_id = request.headers.get("last-event-id", "")
    after = int(last_id.rsplit(":", 1)[-1]) if ":" in last_id else 0

    async def frames():
        queue = hub.subscribe(request_id)  # subscribe FIRST, replay second:
        try:                               # no gap for an event to fall through
            last = after
            replayed = await eventlog.replay(request_id, after)
            for seq, event in replayed:
                last = seq
                yield sse(event, event_id=f"{request_id}:{seq}")

Then the follow phase, which is Part 7's keepalive loop pointed at a queue instead of a translator:

            while True:
                try:
                    item = await asyncio.wait_for(queue.get(), timeout=10)
                except asyncio.TimeoutError:
                    yield ": keepalive\n\n"  # Part 7's trick, same job:
                    continue                 # fail fast on a dead socket
                if item is None:
                    return  # the worker hung up the hub; the log has the rest
                seq, event = item
                if seq <= last:
                    continue  # already replayed; at-least-once, deduped by seq
                last = seq
                yield sse(event, event_id=f"{request_id}:{seq}")
        finally:
            hub.unsubscribe(request_id, queue)

Two details carry the architecture. First, seq <= last: continue is the dedup guard the hub section promised, four characters of insurance on the subscribe-then-replay overlap. Second, every frame now goes out with an SSE id: field, and this is where an eight-part-old decision pays out. SSE always had a second field we never used: id: rides outside the JSON payload, at the protocol level, and browsers remember the last one they saw. The Part 2 vocabulary needed zero changes to become durable. No event type was added, none was modified; the sequence number lives on the envelope's envelope. Collect the payoff explicitly: we've extended the vocabulary six times in eight parts (artifacts, approvals, and today an id: line), and the Part 3 parser, the badges, the cards, the panel: none of them ever changed. That's what "design the envelope before you need it" buys.

One tiny change in events.py makes it possible, and it's the whole diff there:

def sse(event: dict, event_id: str | None = None) -> str:
    """Frame one event dict as a server-sent event. The optional id is
    SSE's own bookmark field: browsers remember the last one they saw
    and send it back as Last-Event-ID when they reconnect."""
    head = f"id: {event_id}\n" if event_id else ""
    return head + f"data: {json.dumps(event)}\n\n"

And because the pipe is dumb, it's also honest about the one thing it can't survive: a server that died mid-run. If the log holds a story with no ending and no live worker, the pipe replays what exists and appends an apology error event, so a rejoining client renders everything up to the crash and knows the rest is gone. The log is on disk; even kill -9 only costs you the events that were never written.

The whole machine, on one page:

And here's the part I find quietly beautiful: the flight recorder sits right there on disk, inspectable with tools you already have. A run's whole story, then the same story over the wire, bookmarks and all:

EventSource, six parts late

Since Part 3 the frontend has parsed SSE by hand with a fetch-reader, and the honest reason was transport: the stream came from a POST with a JSON body, and the browser's built-in SSE client only speaks GET. That constraint just evaporated. GET /stream/{request_id} is exactly the URL shape EventSource was born for, so readSse.ts, the first file the frontend ever owned, retires with honors. Its replacement brings a superpower we'd otherwise have to build: automatic reconnection. When the connection drops, the browser retries on its own, sending the last id: it saw as a Last-Event-ID header, which the stream endpoint already parses into after. The reconnect path was ten lines of server code and zero lines of client code.

The client's half of the bargain is remembering its claim ticket somewhere a refresh can't destroy. That place is sessionStorage: it survives reloads (and rides into a duplicated tab), but dies with the browser session, which is exactly the lifetime "a run I'm currently watching" wants; conversations that finished are already the sidebar's job. send() stores the ticket the moment POST /chat answers, and attaching to the stream becomes its own small function, because two different callers need it:

  // Attach to a run's stream: replay first, then the live tail. Called by
  // send() right after the ticket arrives, and by the resume effect after
  // a refresh. Neither caller can tell the difference; that's the point.
  function followStream(requestId: string) {
    esRef.current?.close();
    requestIdRef.current = requestId;
    const es = new EventSource(`${API_BASE}/stream/${requestId}`);
    esRef.current = es;
    let lastSeq = 0;
    es.onmessage = (e) => {
      // The dedup guard. Replay-then-follow delivers at least once; the
      // seq in the SSE id makes it exactly once where it matters.
      const seq = Number(e.lastEventId.split(":").pop());
      if (seq) {
        if (seq <= lastSeq) return;
        lastSeq = seq;
      }
      handleEvent(JSON.parse(e.data) as AgentEvent, es);
    };
    // No onerror theatrics: on a dropped connection EventSource retries
    // by itself with Last-Event-ID, the server replays what we missed,
    // and the guard above swallows any overlap.
  }

handleEvent is the same switch that has lived inside the send loop since Part 3, extracted whole; types.ts didn't change by a character. The second caller is the new star, a mount effect that checks for an orphaned ticket:

  // The resume: if this tab was watching a run, pick it back up. Prior
  // turns come from Part 5's history endpoint; the in-flight turn is
  // rebuilt, event by event, from the log's replay.
  useEffect(() => {
    const pending = readPending();
    if (!pending) return;
    (async () => {
      let past: ChatMessage[] = [];
      if (pending.workspaceId) {
        setWorkspaceId(pending.workspaceId);
        if (pending.sessionId) {
          setSessionId(pending.sessionId);
          try {
            const { messages: hist, files: desk } = await fetchConversation(
              pending.workspaceId,
              pending.sessionId,
            );
            past = hist as ChatMessage[];
            // The diary already holds pieces of the turn in flight; drop
            // them and let the replay rebuild that turn instead.

Earlier turns come back through Part 5's replay endpoint, exactly as if you'd clicked the conversation in the sidebar. The turn in flight is deliberately not taken from there: the session diary already contains pieces of it, but the event log holds the authoritative, sequenced version, so the effect trims the diary at the in-flight question and lets the replay rebuild the rest, deltas, badges, cards and all, through the same applyEvent reducer that built them live. One code path, two tenses.

Try the break that opened this part. Start the month-by-month walk, let a few badges land, refresh:

The working timer deserves its sentence: it reads 9 seconds, not 0, because the ticket stores sentAt and the resume passes it straight to the timer. Small lie removed, and the kind of detail that makes refresh-survival feel like continuity instead of recovery.

A Stop button with nothing to apologize for

Part 3 shipped a Stop button with an asterisk I've been dragging along for six parts: you hung up the phone; you didn't stop the worker. AbortController killed the fetch, the SDK subprocess kept working 10 to 15 more seconds until a write hit the dead pipe, and the stopped turn's cost vanished unrecorded, which Part 3's cost table explicitly lamented. Today the asterisk dies. The worker holds a live ClaudeSDKClient, and that client has the method this series has owed you since the day the button appeared:

async def cancel_request(request_id: str) -> bool:
    """The real Stop button. interrupt() tells the SDK to abandon the
    turn; the stream then ends with a normal ResultMessage, so even a
    stopped turn gets a receipt in the log. False = nothing to stop."""
    running = RUNNING.get(request_id)
    if running is None:
        return False
    running.bridge.deny_all()  # unpark the agent if it's waiting on a card
    await running.client.interrupt()
    return True

POST /chat/{request_id}/cancel calls it, and the frontend's Stop button swaps abort() for that POST while keeping the stream open, because the receipt is still coming. Note the deny_all first: if the agent is parked on an approval card, the interrupt can't land until the callback returns, so cancelling a run answers its open questions with "no" on the way out. Measured, on this app, repeatedly: the complete event hit the log 10 to 50 milliseconds after the cancel POST. Against Part 3's 10-to-15-second afterlife, that's three orders of magnitude, and this time the button's label is telling the truth.

One honest wrinkle, measured because it surprised me. The interrupted turn's ResultMessage reports whatever accounting the SDK had finalized when the turn died: stop between API rounds and the receipt carries real numbers ($0.0159 in one run, $0.0117 in the screenshot); stop in the middle of a round and usage can come back all zeros, $0.0000 on a turn that visibly ran five tool calls. The tokens were still consumed; only the receipt is short. You get a receipt event either way, which is the architectural point, but don't build billing on interrupted turns' receipts. On 0.2.110 the tell is usage.iterations: empty list means the accounting died with the round.

Approvals meet durability

Now the composition test, the moment where you find out whether two features you built separately are actually one system. Part 7 parks the agent on an asyncio.Future until a human clicks. Part 9 claims the watching tab is disposable. So: ask for a chart, wait for the Bash card, and refresh while the agent is parked.

Walk the machinery, because all of Act II shows up for this one frame. The approval_request event is in the log now (the bridge emits through the worker's sink), so the replay re-renders the card. The asyncio.Future never lived in the HTTP response's world at all; it's parked in the server's PENDING dict, still attached to the paused can_use_tool callback. Your click POSTs to the same decision endpoint as ever, which resolves the Future, un-parks the agent, and emits approval_resolved through the sink for the log to record and the hub to broadcast to the new tab. Approve, and the chart arrives in a page load that didn't exist when the question was asked.

This formally closes Part 7's honesty note: approval events were wire-only, replayed conversations showed denied results without the cards that produced them, and I promised you a durable home. Every parcel has one now. The note comes off the wall the same way Part 1's scissors warning did in Part 7.

Two tabs, one run

The dessert is the show-off version of everything above. Start a run, then open the same analysis in a second tab (duplicate the tab, or copy the ticket; the demo below does it live). Both tabs subscribe to the same hub, both render the same events, and when the receipt comes, it comes to both:

The transcripts are byte-identical because there's exactly one source of truth and both tabs are reading it. And the second subscriber doesn't have to be a browser: curl -N localhost:8000/stream/{request_id} follows the same run from a terminal, because a dumb pipe doesn't check who's drinking from it.

The cost ritual

All real runs from building this part:

Read the middle rows together: durability's marginal cost is a SQLite file. Replays are free, extra subscribers are free, and the architecture's only new spend is the disk it takes to remember what you already paid the model to say once.

Commit it, from the project root:

git add backend frontend
git commit -m "part 9: durable streams - event log, broadcast hub, and a real stop button"

And that's the curtain on Act II. Look at what the last four parts actually assembled: a real database behind a read-only custom tool (Part 6), a human between the agent and anything risky (Part 7), deterministic law and an audit trail underneath the human (Part 8), and now runs that outlive any browser tab, with every event on disk and a Stop button that stops (Part 9). The analyst from Act I worked; this one can be trusted, audited, interrupted, and rejoined. That's the difference between a demo and a product, and it's also, not coincidentally, the shape of the production reference app this series shadows. Act III makes the analyst better at its actual job: planning before it acts, asking you structured questions, showing its thinking, delegating to a reviewer that catches its mistakes, reaching the outside world safely, proving its quality with evals, and finally shipping to a real server. First stop, Part 10: the agent proposes a plan before touching anything expensive, and asks you real questions instead of guessing.

The complete, tested code for this part lives in part-09-durable-streams in the companion repo. Code blocks with a GitHub icon link straight to the exact file; "View full file" shows the whole file in place with this section's changes highlighted.