features streaming chat

Streaming chat

Active contributors: Mario Zechner, kt, Armin Ronacher

Purpose

Streaming chat is the end-to-end path that moves a user's message from the browser into the agent loop and streams the assistant's response back as discrete frames. It has two complementary channels. An NDJSON turn stream over POST /api/chat is authoritative for the in-flight turn. A separate SSE push channel at GET /api/chat/events carries out-of-turn events (interactive tool questions, status changes, queue updates) and replays frames emitted while a tab was closed. Both channels are adapted by web/server from AgentSessionEvent frames produced by the agent runtime in packages/coding-agent, and both are consumed by the same pure client reducer in web/app.

The architecture keeps the browser side-effect free. All event translation happens in web/server/src/event-mapper.ts; the client only applies ChatStreamEvent frames through a pure reducer. This makes the turn stream replayable, resumable, and testable in isolation.

How it works

A turn is streamed as NDJSON over POST /api/chat. web/app/src/routes/api/chat.ts delegates to handleChatPost (web/server/src/handlers/chat.ts), which validates the body against ChatRequestSchema, resolves or resumes the session through PrimeBridge, writes a leading start frame, and subscribes a listener that forwards every matching session frame. The done or error frame closes the ReadableStream. Frames are written one JSON object per line, so the browser reads them line by line.

Out-of-turn events take the SSE channel. handleChatEventsGet (web/server/src/handlers/chat-events.ts) opens a text/event-stream and replays retained frames from the session RingBuffer (web/server/src/ring-buffer.ts). Each frame carries a monotonic seq. A reconnecting client sends Last-Event-ID: <last seq> (or a lastEventId query param, since a freshly created EventSource does not resend the header), and the server replays every entry with a higher seq. If the client is behind the oldest retained frame, the buffer reports overflow, the handler emits a state: agent_settled frame with message resync-required, and the client falls back to GET /api/chat/session to rebuild the transcript. The buffer capacity is 500 frames per session.

sequenceDiagram
    participant B as Browser (web/app)
    participant ND as POST /api/chat (web/server)
    participant PB as PrimeBridge
    participant AG as AgentSession (coding-agent)
    participant RB as RingBuffer
    participant SSE as GET /api/chat/events
    participant R as Reducer (chat-stream-state)

    B->>ND: POST /api/chat {sessionId, message}
    ND->>PB: addEventListener + bridge.prompt(...)
    PB->>AG: session.prompt(text)
    AG-->>PB: tool_execution_start (IPython cell)
    PB->>RB: push tool frame
    RB-->>ND: forward tool frame (NDJSON)
    ND-->>B: "tool" line
    B->>R: applyChatStreamEvent (upsert tool part)
    AG-->>PB: tool_execution_end
    PB->>RB: push tool frame (output-available)
    RB-->>ND: forward
    ND-->>B: "tool" line
    B->>R: applyChatStreamEvent (mark output-available)
    AG-->>PB: ExtensionUIContext.select (out-of-turn question)
    PB->>PB: dialogs.open(...)
    PB->>RB: push tool-Question frame
    RB-->>SSE: tool-Question frame (event: message)
    SSE-->>B: tool-Question frame
    B->>R: append tool-Question part
    B->>ND: POST /api/chat/question {toolCallId, answer}
    ND->>PB: bridge.answerDialog(...)
    PB->>AG: select resolves with choice
    AG-->>PB: agent_end
    PB->>RB: push done frame (final ChatMessage)
    RB-->>ND: forward
    ND-->>B: "done" line, stream closes
    B->>R: merge final message, status -> ready

Loading

The tool question is the out-of-turn path: it does not arrive on the NDJSON stream the turn is using, it arrives on SSE while the turn stream stays open. web/app/src/lib/pi/use-pi-chat.ts ignores NDJSON-sourced frames while status is submitted/streaming and only applies out-of-turn pushes, so the two channels never double-apply a frame.

Client stream state machine

usePiChat (web/app/src/lib/pi/use-pi-chat.ts) owns messages, status, queue, activityLabel, and planLabel. status is ready | submitted | streaming | error. Sending a message routes through usePiChatMessaging.sendMessage (web/app/src/lib/pi/use-pi-chat-messaging.ts): it lazily creates a session on first send, optimistically appends the user message, sets submitted, and calls chatClient.streamMessage. readChatStream (web/app/src/lib/pi/chat-fetch.ts) reads NDJSON lines and feeds each frame to handleStreamEvent, which calls the pure reducer applyChatStreamEvent (web/app/src/lib/pi/chat-stream-state.ts).

The reducer reconciles one in-flight assistant bubble. A start frame opens a placeholder bubble and flips status to streaming. delta/thinking/tool frames append or upsert text, thinking, and tool parts, reconciling the bubble id if the mapper assigns a different messageId mid-turn. queue, plan, state, compaction, and retry frames update derived labels. done merges the final ChatMessage, promotes a thinking-only turn to visible text, resets the queue, clears the activity label, and flips status back to ready.

Queue states: steering and follow-up

During an active turn the server can publish a queue frame with two lists, steering and followUp (QueueState in web/app/src/lib/pi/chat-fetch.ts). Pressing Enter mid-stream steers the current turn; pressing Alt+Enter queues a follow-up. enqueueDuringStream POSTs the extra message with streamingBehavior: "steer" | "followUp", optimistically appends the user message, and refreshes the sessions list so the queue badge reflects the just-submitted item. The queue event lands on the main turn's NDJSON stream; because the steer POST opened its own (short) stream, the client refreshes the session list to surface the badge instead of waiting for the turn to end.

SSE cursor persistence

For every visible sessionId, usePiChat opens one EventSource. The last seen seq is written to sessionStorage under pi:sse:last-event-id:<sessionId> on every message, so a page reload resumes the cursor without a server round-trip. The server replays frames emitted while the tab was closed, and an agent_settled state frame triggers a transcript resync via client.loadSession.

Event mapping

mapAgentSessionEvent (web/server/src/event-mapper.ts) is a pure AgentSessionEvent -> ChatStreamEvent[] translation with no I/O. Per-session state lives on EventMapperState. Tool names pass through toPascalCase, so ipython becomes tool-IPython, bash becomes tool-Bash, and edit_file becomes tool-EditFile; thinking becomes tool-Thinking. Tool execution maps to three frame kinds: tool_execution_start and tool_execution_update produce input-streaming tool parts, and tool_execution_end produces an output-available or output-error part. Compaction, auto-retry, and queue events map to their own frame types; a known set of events the UI does not render (rlm_child_update, bash_output, session_info_changed, and others) is deliberately ignored.

Integration points

• Runtime: packages/coding-agent (AgentSession, AgentSessionEvent, ExtensionUIContext, SessionManager) drives the events. See coding-agent and session-runtime.
• Wire contract: web/protocol (ChatStreamEvent, ChatRequest, QueueState). See web-protocol.
• Server adapter: web/server (PrimeBridge, event-mapper, ring-buffer, handlers). See web-server.
• Client: web/app (use-pi-chat, chat-stream-state, chat-fetch). See web-app.
• Rendering: web/design renders the tool and question cards. See web-design and tool-cards.
• HTTP endpoints: web-api.

Entry points for modification

• Change how runtime events surface as frames: edit web/server/src/event-mapper.ts (pure mapping) and web/server/src/sse-replay.ts (replay filtering).
• Change replay capacity or overflow behavior: edit web/server/src/ring-buffer.ts and web/server/src/handlers/chat-events.ts.
• Change the turn stream handler: edit web/server/src/handlers/chat.ts.
• Change client turn/stream handling: edit web/app/src/lib/pi/use-pi-chat.ts, web/app/src/lib/pi/use-pi-chat-messaging.ts, and the reducer in web/app/src/lib/pi/chat-stream-state.ts.
• Change steering/follow-up: edit web/app/src/lib/pi/use-pi-chat-messaging.ts.
• Change the NDJSON line reader or state labels: edit web/app/src/lib/pi/chat-fetch.ts.

Key source files

File	Role
web/server/src/handlers/chat.ts	POST /api/chat: validates the request, streams NDJSON frames over a ReadableStream, closes on done/error.
web/server/src/handlers/chat-events.ts	GET /api/chat/events: SSE with ring-buffer replay, Last-Event-ID, heartbeat, and resync-required overflow.
web/server/src/ring-buffer.ts	Per-session frame store with monotonic seq; replaySince with overflow detection.
web/server/src/event-mapper.ts	Pure AgentSessionEvent -> ChatStreamEvent[] translation; toPascalCase tool naming; EventMapperState.
web/server/src/sse-replay.ts	Decides which ring-buffer frames to replay to a first-time SSE client.
web/server/src/prime-bridge.ts	Session coordinator; prompt/steer/followUp, ring buffers, pending dialogs, dispatch.
web/server/src/pending-dialogs.ts	Registers and resolves confirm/select/input promises; 60s auto-cancel.
web/app/src/routes/api/chat.ts	TanStack route wrapper delegating POST /api/chat to handleChatPost.
web/app/src/lib/pi/use-pi-chat.ts	Chat hook: state, session lifecycle, SSE EventSource cursor, plan/question handling.
web/app/src/lib/pi/use-pi-chat-messaging.ts	Turn send, steering, and follow-up; applies stream events through the reducer.
web/app/src/lib/pi/chat-stream-state.ts	Pure applyChatStreamEvent reducer plus session-metadata normalization.
web/app/src/lib/pi/chat-fetch.ts	NDJSON stream reader, QueueState, labelForState, sequence tracking.