Streaming Architecture

Data part protocol, transport-agnostic StreamWriter, and resumable SSE that fuses a Redis Stream (durable history) with pub/sub (live push) — monotonic ids stitch history to live with no dedup table

The engine only writes to one stream; delivery is someone else’s job

Once the agent is running, output is continuous — text tokens, tool results, state transitions, notifications, metadata. All of it has to reach the client reliably and behave identically on web (SSE) and desktop (IPC) — yet the engine itself shouldn’t care how the network wobbles or which transport carries it. It writes to one transport-agnostic stream, and everything else is split across four layers, each owning one span:

  1. StreamWriter — the transport-agnostic event-write interface;
  2. Data Part Protocol — agent-specific events riding alongside the AI SDK content parts;
  3. Task Transport — SSE (default) or WebSocket, both carrying the same frame format;
  4. Resumable SSE — Redis-backed reconnect-and-resume, SSE path only.

StreamWriter Interface

The backend agent writes events through a transport-agnostic StreamWriter interface. The business layer injects the concrete implementation:

export interface StreamWriter {
  write(event: { type: `data-${string}`; data: unknown; transient?: boolean }): void;
}
PlatformImplementation
Web — SSE (default)Wraps UIMessageStreamWriter from createUIMessageStream({ execute })
Web — WebSocketSame UIMessageStreamWriter; frames are piped to wsHub.send(connectionId, ...)
Chat broadcastRoom-wide fan-out via wsHub.broadcast() (text-only, chat service only)
Desktop (IPC)Wraps Electron IPC send

Events are classified as persistent (stored in message history via context.write()) or transient (displayed but not persisted via context.writeTransient()). State transitions and notifications are typically transient; tool results and text content are persistent.

Data Part Protocol

The Vercel AI SDK defines a standard streaming protocol for text and tool-call chunks. Zapvol extends this with custom data part events that carry agent-specific metadata alongside the content stream:

EventPayloadPurpose
agent-initAgent config, sandbox metadata, model infoBootstrap client-side agent state before first token
agent-stateAgentStateKey enum valueDrive the UI progress indicator through state transitions
notificationMessage string + severity levelSurface agent-generated notifications as toast messages
title-updatedAuto-generated title stringUpdate the task title without a separate API call
tool-streamstdout/stderr line chunksStream tool execution output in real-time (e.g., build logs)

All events are multiplexed into the same SSE stream as content, ensuring a single connection carries all agent-to-client communication.

State Transition Events

The toUIMessageStream() function maps AI SDK stream parts to state events: start → generating, start-step → executing, finish → completed, error → error, abort → aborted. Each transition is broadcast as a transient agent-state event.

Message Metadata

The messageMetadata callback attaches timing and usage data to stream messages:

FieldSet whenPurpose
streamCreatedAtStream startLatency measurement baseline
firstTokenAtFirst text/reasoning/tool-callTime to first token metric
completedAtStream finishTotal execution duration
totalUsageStream finishToken counts (input + output)
finishReasonStream finishWhy the agent stopped

Task Transport — SSE (default) vs WebSocket (opt-in)

Task Stream Transport — SSE vs WebSocket One stream core (buildUiStream), two transport wrappers, same UIMessageChunk frames SSE (default) WebSocket (VITE_TASK_WS_TRANSPORT=1) ① send ② transport ③ server entry ④ server core ⑤ wire frame ⑥ client parse useChat + DefaultChatTransport (default fetch) fetch("POST /api/tasks/:id/messages") routes/tasks.ts → taskOrchestrator.execute buildUiStream → createUIMessageStreamResponse data: <UIMessageChunk JSON>\n\n Browser SSE parser (inside DefaultChatTransport) useChat + DefaultChatTransport (createWsFetch) wsClient.send("task:stream:start", body) ws-service → taskOrchestrator.executeWs buildUiStream → reader.read() loop {type:"task:stream:frame", taskId, frame: ...} createWsFetch re-encodes → same SSE parser ⚠ only real diff Convergence — useChat sees identical UIMessageChunks Same state updates, same rendering, same hooks — transport layer is invisible above this line Resume: SSE only (Redis-backed). Abort: both transports call taskOrchestrator.abort (HTTP /abort shared).

The core builds one stream (startTaskUiStream in task-runner.ts); two thin wrappers on task-orchestrator.ts frame it:

WrapperEntryClient reaches it viaWire format
executePOST /api/tasks/:id/messagesDefaultChatTransport with standard fetchSSE frames data: <UIMessageChunk JSON>\n\n
executeWstask:stream:start on /wsDefaultChatTransport with createWsFetch() synthetic bodyWS text frames { type: "task:stream:frame", taskId, frame: UIMessageChunk }

Key property — format equivalence: toUIMessageStream() produces the same UIMessageChunk JSON objects in both paths. SSE adds a data: prefix; WS wraps in an envelope. The agent engine, writer, and useChat consumer are completely transport-agnostic.

When to pick which

FactorSSEWebSocket
DefaultOn unless VITE_TASK_WS_TRANSPORT=1Opt-in
Resume after dropYes — the run produces into a Redis stream buffer a reconnect reads back (see next section)None — reconnect refetches DB history; in-flight frames lost
Connection modelOne HTTP request per task runOne long-lived socket shared with chat + any future features
Desktopn/a — desktop uses IPCn/a — customFetch short-circuits WS
Multi-process infraRedis Streams for resumeRedis Pub/Sub already in wsHub for chat broadcast; point-to-point via wsHub.send(connectionId) bypasses it

SSE remains the recommended default because of resume. WebSocket is useful for eventual bidirectional features (cheaper abort round-trip, live cursor hints, shared socket with chat/presence) and for environments where SSE is blocked.

Abort flow (identical on both transports)

The client’s Stop button always goes through POST /api/tasks/:id/abort. That handler calls taskOrchestrator.abort, which runs two steps in order:

  1. taskService.abort(userId, taskId) — ownership check + isActive=false. Throws on unauthorized access.
  2. abortManager.abort(taskId) — fires the AbortSignal passed into runAgentLoop, interrupting the LLM at its next await point.

The WS task:stream:abort message routes to the same orchestrator method, so the two transports have byte-identical Stop semantics. (Client-side, useChat’s internal signal aborts only detach the local consumer — they do not fire an abort to the server, matching SSE’s behavior when the HTTP connection drops.)

Resumable SSE (Redis Stream + pub/sub)

The run never streams straight to the client — it produces into a Redis-backed StreamBuffer (redis-stream-buffer.ts), and every reader (the live execute response and a mid-run reconnect) reads that one buffer. That is what lets a minutes-long run survive a dropped connection with no re-execution.

Resumable SSE — Redis Streams recovery Monotonic entry IDs + cursor continuity guarantee strict FIFO ordering — zero gaps, zero duplicates Writer Task Orchestrator Redis one Stream per run (XADD / XREAD) Reader → Client XREAD from cursor → SSE PHASE 1 — Normal streaming XADD chunk1 → id 1-0 XREAD → chunk1 (cursor=1-0) XADD chunk2 → id 2-0 XREAD → chunk2 (cursor=2-0) XADD chunk3 ... XREAD → chunk3 ... PHASE 2 — Network interruption XADD chunk4 (buffered in stream) XADD chunk5 (buffered in stream) disconnected PHASE 3 — Client reconnects from its last cursor reconnect · XREAD from cursor 3-0 replay 3-0..5-0 (history), then block for live XADD chunk6 (live) XREAD → chunk6 (cursor=6-0) XADD end-entry (terminal) XREAD → end-entry → DONE Ordering guarantee Monotonic stream IDs + a per-reader cursor — history and live entries drain from one XREAD loop. Client perception Resumed stream = uninterrupted stream. Zero gaps, zero duplicates.

Durable history + live push

The buffer combines two Redis mechanisms per chunk:

  • XADD appends the chunk to a per-turn Redis Stream — the durable history a (re)connecting reader backfills from.
  • PUBLISH pushes the same chunk on a channel the instant it is XADD’d — live delivery, event-driven (no polling, no blocking XREAD), so it keeps the model’s natural token cadence.

One shared subscriber connection multiplexes every channel via a dispatch map — so there is no per-stream dedicated connection, which is precisely why live delivery is pub/sub + XRANGE rather than a blocking XREAD loop.

How a reader self-coordinates history ↔ live

A reader has no handshake with the writer — the producer may be a worker that starts after the reader connects — so it orders itself by Redis’s monotonic stream ids:

  1. SUBSCRIBE first (awaited), buffering any live pushes — nothing published after the coming snapshot can be missed.
  2. XRANGE-backfill the history; track the last id emitted.
  3. Flush the buffered pushes with id > lastId (dedup), then stream live.

Because stream ids are monotonic, those three phases are id-contiguous — no gap, no overlap, no dedup table. Correctness rests on two orderings: the producer XADDs before it PUBLISHes (the id comes from the XADD), and the reader SUBSCRIBEs before it XRANGEs. Client-generated monotonic ids (<baseMs>-<seq>) let the XADD and PUBLISH for a chunk share one pipeline. This replaces the old per-listener promise-chain dedup (lib/resumable-stream.ts) with consumer-side id-contiguity — but pub/sub did not go away; it is the live-push half, now paired with the Stream for durable history.

Lifecycle & memory

One stream = one assistant turn’s chunks — produce DELs the key first, so a stream never spans turns. Each write refreshes a grace TTL (600 s) so a live long-horizon turn stays alive and a crashed producer (no terminal entry) self-expires within it. The terminal entry (an end field via XADD) drops the key to a shorter retention window (300 s) for late reconnects. A reader that subscribes before the producer ever appears gives up after a 120 s initial-wait cap.

OperationRedisPurpose
Append chunkXADD + PUBLISHDurable history entry + live push, sharing one pipeline
Live deliverySUBSCRIBE on the channelEvent-driven push, no polling
BackfillXRANGE over the streamReplay history for a (re)connecting reader
Completionterminal end entry (XADD)Signal end of stream to readers
Cleanupgrace TTL (600 s) → retention (300 s)Live turn stays alive; settles to a reconnect window
Was this page helpful?
Edit on GitHub