Lesson 01 — Chat Stream Lifecycle

1. Why this lesson exists

#1575 is not a bug fix you can localise to a single function. It is a parity question: do live observers and reconnecting observers see the same terminal outcome? To answer that, you need a mental model of what a chat turn is as it moves through three layers in parallel:

• the wire (WebSocket frames a client receives),
• the durable store (cf_ai_chat_stream_metadata and cf_ai_chat_stream_chunks),
• and the programmatic surfaces (SaveMessagesResult, onChatResponse, agent-tool child run rows).

#1567 already made the programmatic surfaces honest about errors. #1575 is about making the wire and the durable store agree.

2. Cast of characters

Before any diagrams, pin these down. They reappear constantly.

3. A happy turn, end to end

Walk through a single successful chat turn before introducing any failure mode. Everything else is a deviation from this picture.

3.1 The pipeline

flowchart LR
  client[Browser useChat]
  ws[(WebSocket)]
  agent[Think or AIChatAgent]
  stream["_streamResult / _streamSSEReply"]
  rs[ResumableStream]
  meta[(cf_ai_chat_stream_metadata)]
  chunks[(cf_ai_chat_stream_chunks)]
  hook[onChatResponse]

  client -- cf_agent_use_chat_request --> ws --> agent
  agent --> stream
  stream -- toUIMessageStream --> stream
  stream -- broadcast cf_agent_use_chat_response --> ws --> client
  stream --> rs
  rs -- start --> meta
  rs -- storeChunk batched --> chunks
  stream -- complete --> rs
  rs -- status = 'completed' --> meta
  stream -- fire --> hook

3.2 Lifecycle, one frame at a time

sequenceDiagram
  autonumber
  participant C as Client
  participant A as Agent (Think)
  participant S as _streamResult
  participant R as ResumableStream
  participant DB as SQLite

  C->>A: cf_agent_use_chat_request (POST body)
  A->>S: invoke with requestId
  S->>R: start(requestId)
  R->>DB: INSERT cf_ai_chat_stream_metadata status='streaming'
  loop for each chunk from toUIMessageStream()
    S->>R: storeChunk(streamId, body)
    R->>DB: buffered INSERT cf_ai_chat_stream_chunks
    S->>C: cf_agent_use_chat_response { body, done:false }
  end
  S->>R: complete(streamId)
  R->>DB: UPDATE metadata status='completed'
  S->>C: cf_agent_use_chat_response { body:"", done:true }
  S->>A: _fireResponseHook(status:"completed")

4. Four terminal statuses

SaveMessagesResult.status is the single source of truth. Memorise the matrix below, because #1575 is largely about making the wire and the durable store reach the same status as this enum.

5. The resumable stream tables

ResumableStream exists so that a client which drops a WebSocket connection can reconnect and recover whatever it missed without rerunning the LLM. There are two tables.

cf_ai_chat_stream_metadata
  id           text primary key
  request_id   text
  status       text   -- 'streaming' | 'completed' | 'error'
  created_at   integer
  completed_at integer

cf_ai_chat_stream_chunks
  id           text primary key
  stream_id    text
  body         text   -- the JSON-encoded UIMessage chunk
  chunk_index  integer
  created_at   integer

5.1 Three lifecycle states

stateDiagram-v2
  [*] --> streaming: start(requestId)
  streaming --> completed: complete(streamId)
  streaming --> error: markError(streamId)
  completed --> [*]
  error --> [*]

5.2 The four-frame reconnect handshake

Connecting clients always negotiate. The server proactively says "there is a stream you can resume"; the client decides whether to opt in.

sequenceDiagram
  autonumber
  participant C as New WS client
  participant A as Agent
  participant R as ResumableStream

  Note over C,A: Case A — stream is still live
  A->>C: cf_agent_stream_resuming { id: requestId }
  C->>A: cf_agent_stream_resume_ack { id: requestId }
  A->>R: replayChunks(connection, requestId)
  R-->>C: cf_agent_use_chat_response { body, replay:true } (xN)
  R-->>C: cf_agent_use_chat_response { body:"", replay:true, replayComplete:true }
  Note over A,R: live stream continues from here

  Note over C,A: Case B — stream completed before client reconnected
  C->>A: cf_agent_stream_resume_request
  A->>R: replayCompletedChunksByRequestId(connection, id)
  R-->>C: cf_agent_use_chat_response { body, replay:true } (xN)
  R-->>C: cf_agent_use_chat_response { body:"", done:true, replay:true }

  Note over C,A: Case C — nothing to resume
  C->>A: cf_agent_stream_resume_request
  A-->>C: cf_agent_stream_resume_none

6. Where errors enter the picture

Three places a turn can end with status:"error":

flowchart TD
  start([_streamResult begins])
  loop{for await chunk}
  inband{chunk.type === error?}
  thrown{thrown error?}
  late{captureOutput &&
result.output rejects?}
  ok[status: completed]
  err[status: error]

  start --> loop
  loop -- next chunk --> inband
  inband -- yes --> err
  inband -- no --> loop
  loop -- iterator throws --> thrown
  thrown -- yes --> err
  loop -- iterator done --> late
  late -- yes --> err
  late -- no --> ok

6.1 In-band error chunk (the common case)

// think.ts:5720 (paraphrased)
if (action?.type === "error") {
  streamError = action.error;
  this._emit("message:error", { error: streamError });
  this._broadcastChat({
    type: MSG_CHAT_RESPONSE,
    id: requestId,
    body: action.error,
    done: false,
    error: true,
  });
  break;
}
// ...later
if (streamError) {
  this._resumableStream.markError(streamId);
} else {
  this._resumableStream.complete(streamId);
}
this._broadcastChat({ type: MSG_CHAT_RESPONSE, id: requestId, body: "", done: true });

Two wire frames go out: an error:true, done:false frame carrying the error text, then a normal done:true frame. The metadata moves to error. onChatResponse fires with status:"error". Partial parts before the break are persisted.

6.2 Thrown stream error

The catch branch in _streamResult sets streamError from error.message, calls markError, and (if no done was sent) emits a done:true, error:true frame in a single shot.

6.3 Structured output rejection

When captureOutput is true, awaiting result.output after the stream loop can reject. Same terminal handling, no extra wire frame because the stream is already done.

7. Live vs replay parity

This is the heart of #1575. Walk both observers through the same failing turn and compare what they see.

sequenceDiagram
  autonumber
  participant L as Live client
  participant A as Agent
  participant R as ResumableStream
  participant N as Reconnecting client

  L->>A: cf_agent_use_chat_request
  A->>R: start
  A->>R: storeChunk "Hello"
  A->>L: { body:"Hello", done:false }
  A->>R: storeChunk "Wor"
  A->>L: { body:"Wor", done:false }
  Note over A: provider emits { type:"error", errorText:"boom" }
  A->>L: { body:"boom", done:false, error:true }  ←  L's stream errors
  A->>R: markError
  A->>L: { body:"", done:true }
  Note over L,A: ...time passes, L's tab refreshes as N...
  N->>A: connect, cf_agent_stream_resume_request
  A-->>N: cf_agent_stream_resume_none  ←  bug: no error replay
  Note over N: N sees an empty turn, no partial content, no error

What #1575 wants instead:

8. Agent-tool runs ride on top

An agent-tool run is a sub-agent invocation packaged so the parent can see its progress and final status. It is built on top of saveMessages, plus a child-run row and a tailer forwarder pipeline for live chunks.

flowchart LR
  parent[Parent agent]
  child[Child agent]
  save[saveMessages]
  childrow[(cf_agent_tool_child_runs)]
  fwd[broadcast forwarder]
  tailer["tailAgentToolRun (ReadableStream)"]
  forwarders["_agentToolForwarders Map>"]
  errors["_agentToolLastErrors Map"]

  parent -- runAgentTool(input, runId) --> child
  child -- startAgentToolRun(input, {runId}) --> childrow
  child -- saveMessages --> save
  save -- broadcast cf_agent_use_chat_response --> fwd
  fwd -- if forwarders exist --> forwarders
  fwd -- if error frame --> errors
  parent -- tailAgentToolRun(runId) --> tailer
  tailer -. registers forward fn .-> forwarders

8.1 How the run gets its terminal status

startAgentToolRun awaits saveMessages, then looks at three signals to classify the run:

const streamError = result.error ?? this._agentToolLastErrors.get(options.runId);
const status =
  result.status === "error" || skipped || streamError
    ? "error"
    : result.status === "aborted"
      ? "aborted"
      : "completed";

Post-#1567, result.error is populated whenever the child turn errored, so the run can be classified without ever observing a forwarded error frame. The _agentToolLastErrors map is mostly belt-and-braces.

8.2 Where the two packages diverge

flowchart TB
  subgraph think["@cloudflare/think"]
    tbroad["broadcast override"]
    tloop["for each forwarder
set _agentToolLastErrors[runId]"]
    tbroad --> tloop
  end
  subgraph ai["@cloudflare/ai-chat"]
    abroad["broadcast override"]
    aactive{_agentToolActiveRunId
set?}
    aactive_yes["scope error/chunk to that one run"]
    aactive_no["fan out to all forwarders
(legacy behavior)"]
    abroad --> aactive
    aactive -- yes --> aactive_yes
    aactive -- no --> aactive_no
  end

9. Mapping back to #1575

10. Self check

Answer in your head, then expand to verify. If any answer surprises you, that is where to dig next.