Stream Protocol

This document describes the internal streaming protocol used by Helix Agents for real-time event communication.

The canonical schema lives at packages/core/src/types/stream.ts. This document is the authoritative internals reference for the on-the-wire shape — when other internals docs (e.g. subagent-execution.md) reference chunk fields, they should match the schemas defined here.

Overview

The stream protocol enables real-time communication of:

• Text generation — Token-by-token LLM output
• Reasoning/thinking — Internal reasoning traces
• Tool execution — Start / end / approval / argument-streaming events
• Sub-agent activity — Delegation to child agents
• State changes — RFC 6902 patches
• Custom events — Application-specific data
• Errors and output — Final results
• Run lifecycle — Interrupt, resume, pause, supersede, checkpoint, step commit/discard, stream resync, suspension marker

All chunk schemas extend baseChunkSchema() which carries agentId, agentType, timestamp, and step fields, plus three optional fields. The discriminator field is type. Validation is via Zod (streamChunkSchema(), streamMessageSchema()).

interface BaseChunk {
  agentId: string; // the session that wrote it ('' = synthesized by the framework)
  agentType: string;
  timestamp: number;
  step: number;
  remoteSource?: RemoteSource; // set on chunks proxied from a remote agent
  runId?: string; // the run whose executor wrote it (stamped by withRun)
  attemptId?: string; // the LLM attempt that streamed it (stamped by withAttempt)
}

• agentId (tree client calls T0). The session that wrote the chunk, never a run id; '' is a stream-level chunk the framework synthesized (a pause's run_paused, a failed reader's error chunk). It is required: no stream manager fills a missing one (agentType and timestamp are still filled from the writer). The rule for a write: a manager whose READ path parses stored chunks with streamChunkSchema() checks each chunk it writes the same way, on the enriched chunk and before anything is stored (core assertStorableStreamChunk): an invalid one is refused as a typed, non-retryable validation_error, so one bad caller fails its own write instead of making the stream unreadable. Redis and the DO-SQLite manager throw it; the CF stream Durable Object's /write (JSON from a binding) answers it as a typed 400, which the binding rethrows typed. Each stores the chunk it built, never the parse. Memory's reader does not parse, so it relies on the type. The contract suite pins it (stream.agent-id-is-kept, stream.enriches-agent-type-and-timestamp, stream.synthesized-chunk-is-stream-level, and stream.missing-agent-id-is-refused where untypedWriteBoundary is set).
• runId. Every chunk a run's executor writes carries the run's id. Runtimes wrap the run's writer in core withRun(writer, runId) (packages/core/src/stream/with-run.ts), which sets it when the chunk has none. Run attribution and the chunk fence use it; run_started remains the ordering marker.
• attemptId. Every chunk an LLM attempt streams carries that attempt's nonce: withAttempt(writer, attemptId?) (packages/core/src/stream/with-attempt.ts) stamps it, and newAttemptId() mints one (attempt-<ms>-<8 hex>). A step that is retried (an activity / step retry, a recoverable error step) streams again under a new attemptId, and the step's step_committed.committedAttemptId names the attempt that won. The UI transformer never forwards attemptId on content; see Attempts.

Fixed: DBOS stamped chunks one step low (RM-58)

The step field is the step the chunk belongs to, and resume cleanup treats chunks with a step greater than a checkpoint's stepCount as orphans. Every runtime, DBOS included, now stamps the step's committed number (1 for the first step), the same stepCount its step checkpoint records (checkpoint.step-count-matches-chunks). Before this release DBOS stamped every chunk one lower. See RM-58 in the conformance findings.

Chunk Types

The full discriminated union has 30+ variants. They group into the following categories:

LLM-emitted chunks

TextDeltaChunk

Token-by-token text from the LLM.

interface TextDeltaChunk {
  type: 'text_delta';
  delta: string; // Incremental text
  agentId: string;
  agentType: string;
  timestamp: number;
  step: number;
}

ThinkingChunk

Reasoning/thinking content from models that support it.

interface ThinkingChunk {
  type: 'thinking';
  content: string;
  isComplete: boolean; // false = streaming; true = complete block
  agentId: string;
  agentType: string;
  timestamp: number;
  step: number;
}

Tool argument streaming

For models that emit tool arguments incrementally.

interface ToolArgStreamStartChunk {
  type: 'tool_arg_stream_start';
  toolCallId: string;
  toolName: string;
  // + Base fields
}

interface ToolArgStreamDeltaChunk {
  type: 'tool_arg_stream_delta';
  toolCallId: string;
  delta: string; // Incremental JSON text
  // + Base fields
}

interface ToolArgStreamEndChunk {
  type: 'tool_arg_stream_end';
  toolCallId: string;
  // + Base fields
}

Tool execution chunks

ToolStartChunk

Emitted when a tool invocation begins.

interface ToolStartChunk {
  type: 'tool_start';
  toolCallId: string;
  toolName: string;
  arguments: Record<string, unknown>;
  // + Base fields
}

ToolEndChunk

Emitted when a tool invocation completes.

interface ToolEndChunk {
  type: 'tool_end';
  toolCallId: string;
  toolName: string;
  result: unknown;
  /** Human-readable error message if the tool failed. */
  error?: string;
  /**
   * Machine-readable error code when the tool failed. For client-executed
   * tools the runtime emits one of `CLIENT_TOOL_ERROR_CODES` (e.g.
   * `client_tool_timeout`, `aborted`); operator tooling can grep on this
   * rather than the humanized `error` text.
   */
  errorCode?: string;
  /**
   * True when the tool was executed by the framework (regular function
   * tools, sub-agents, remote sub-agents, finishWith). False when the
   * tool was executed client-side (`execute: 'client'`).
   *
   * Optional for backward compatibility — chunks emitted by older
   * runtimes don't carry this flag and consumers should treat absence
   * as "client-executed" (the conservative default that matches the
   * Vercel AI SDK's `!providerExecuted` semantics). Used by
   * `extractToolOutputs` in `HelixChatTransport` to filter out
   * provider-executed tool parts from the auto-flush to
   * `/submit-tool-result` — these are already complete on the server.
   */
  providerExecuted?: boolean;
  // + Base fields
}

errorCode: 'aborted' from a checkpoint rewind. resume({ mode: 'from_checkpoint' }) discards pending client-tool / approval calls made after the target checkpoint (core rewindToCheckpoint). For each one the runtime writes a tool_end built by core discardedToolEndChunk: errorCode: 'aborted', the humanized "aborted" text in error, result: undefined, providerExecuted: false, and step set to the target checkpoint's stepCount. A UI waiting on that call should settle it; a later submitToolResult for it returns { status: 'unknown_tool_call' }. Emitted on every runtime (on DBOS best-effort: a stream write failure is logged, not thrown).

ToolApprovalRequestChunk (v7)

Emitted when an approval-gated tool requires human (or automatic) approval before execution. The UI should render the proposed action for the user to approve or deny.

interface ToolApprovalRequestChunk {
  type: 'tool_approval_request';
  toolCallId: string;
  toolName: string;
  /** Server-generated `${runId}::${toolCallId}` (matches Mastra's APPROVAL_ID_SEPARATOR). */
  approvalId: string;
  /** The validated tool input the LLM produced; sent so the UI can render the proposed action. */
  input: unknown;
  /** True when approval was decided automatically (e.g., a deterministic `requireApproval` evaluator). */
  isAutomatic?: boolean;
  // + Base fields
}

ToolApprovalResponseChunk (v7)

Mirrors the request chunk so consumers can correlate request/response pairs via approvalId.

interface ToolApprovalResponseChunk {
  type: 'tool_approval_response';
  toolCallId: string;
  toolName: string;
  approvalId: string;
  approved: boolean;
  reason?: string;
  isAutomatic?: boolean;
  // + Base fields
}

AI SDK UI mapping: tool_approval_request → tool-approval-request { approvalId, toolCallId } (toolName, input, isAutomatic are not forwarded). tool_approval_response emits no UI chunk when approved (ai v6 has no tool-approval-response chunk type); when denied it emits a tool-output-available with the deny string. See the ai-sdk UI-chunk wire contract.

Both approval-gated and client-executed tool flows share the same RunOutcome.kind = 'suspended_client_tool' AgentResult.status (see ./concepts.md). Routing happens off the kind field of the submitToolResult payload ('client-tool-result' vs 'approval-response'), not off the stream-chunk type. UIs typically dispatch off the chunk type for rendering and off the kind for submission.

Tool input/output errors

Distinct from the regular error chunk — these carry toolCallId/toolName and are emitted at specific failure boundaries.

interface ToolInputErrorChunk {
  type: 'tool_input_error';
  toolCallId: string;
  toolName: string;
  error: string;
  partialInput?: unknown;
  // + Base fields
}

interface ToolOutputErrorChunk {
  type: 'tool_output_error';
  toolCallId: string;
  toolName: string;
  error: string;
  // + Base fields
}

Sub-agent chunks

SubAgentStartChunk

Emitted when a sub-agent begins execution.

interface SubAgentStartChunk {
  type: 'subagent_start';
  subAgentType: string; // Sub-agent type (e.g. 'researcher')
  subSessionId: string; // Sub-agent's session ID; correlate sub-agent chunks via this
  callId: string; // Tool call ID that spawned this sub-agent
  // + Base fields (agentId is the PARENT's session ID)
}

SubAgentEndChunk

Emitted when a sub-agent completes.

interface SubAgentEndChunk {
  type: 'subagent_end';
  subAgentType: string;
  subSessionId: string;
  callId: string;
  result: unknown; // Sub-agent output (failures surface via the result payload itself or via emitted error chunks)
  // + Base fields
}

This is the authoritative schema. Other internals docs that reference these chunks should match.

Custom + state chunks

CustomEventChunk

Application-specific events from tools (created via ToolContext.emit()).

interface CustomEventChunk {
  type: 'custom';
  eventName: string;
  data: unknown;
  // + Base fields
}

StatePatchChunk

RFC 6902 JSON Patches for state updates.

interface StatePatchChunk {
  type: 'state_patch';
  patches: JSONPatchOperation[]; // RFC 6902 'add'|'remove'|'replace'|'move'|'copy'|'test'
  // + Base fields
}

Error + output chunks

ErrorChunk

interface ErrorChunk {
  type: 'error';
  error: string;
  code?: string;
  errorDetail?: {
    message: string;
    code?: string;
    category?: string;
    cause?: string;
    retryable?: boolean;
  };
  recoverable: boolean;
  // + Base fields
}

When the error originates from a classified HelixError, the code field contains the error code (e.g., provider_overloaded, provider_rate_limited) and recoverable reflects whether the operation can be retried. See Error Handling Guide.

OutputChunk

Final structured output (when the agent has an outputSchema).

interface OutputChunk {
  type: 'output';
  output: unknown;
  // + Base fields
}

Source / citation chunks (RAG)

interface SourceUrlChunk {
  type: 'source_url';
  sourceId: string;
  url: string;
  title?: string;
  // + Base fields
}

interface SourceDocumentChunk {
  type: 'source_document';
  sourceId: string;
  mediaType: string;
  title: string;
  filename?: string;
  // + Base fields
}

Step chunks (multi-step workflow tracking)

interface StepStartChunk {
  type: 'step_start';
  stepId?: string;
  // + Base fields
}

interface StepEndChunk {
  type: 'step_end';
  stepId?: string;
  usage?: {
    inputTokens: number;
    outputTokens: number;
    cachedTokens?: number;
    cacheWriteTokens?: number;
  };
  finishReason?: 'stop' | 'length' | 'tool-calls' | 'content-filter' | 'error' | 'other';
  // + Base fields
}

Producers:

• step_start: every runtime writes it itself as each LLM attempt's FIRST chunk, stamped with the attempt's attemptId, before the model is called (runtime-js run-loop.ts, Temporal runLLMStep, CF Workflows executeAgentStep, DBOS callLLMStep, which writes no stepId). Its stream position orders a step's retried attempts. The adapter's onStepStart callback is not wired.
• step_end: written only from the adapter's onStepEnd callback, on every runtime (DBOS included, RM-49). @helix-agents/llm-vercel never calls onStepEnd (its fullStream loop has no finish-step case, RM-48), so with the stock Vercel adapter no runtime writes step_end today. RM-49's DBOS half is fixed; the record stays open until RM-48 is.

On the AI SDK UI stream (includeStepEvents: true) they become bare start-step / finish-step; stepId / usage / finishReason stay on the Helix stream only (ai's strict schema forbids them on the UI chunks, DI-20).

File chunk (generated files / attachments)

interface FileChunk {
  type: 'file';
  url: string; // Data URL or hosted URL
  mediaType: string;
  filename?: string;
  // + Base fields
}

Run lifecycle chunks (v7)

These are emitted by the runtime (not the LLM) and signal run-level state transitions.

RunStartedChunk

The run boundary marker. Written once per run, by the run-start WINNER only, after its run-start commit, its stream claim and making the stream live, and before any of the run's own content (see Run-start order). A losing entry writes none. The run of a chunk is the runId of the latest run_started below it, so a reader can split a multi-run stream by run. Core emitRunStarted (packages/core/src/stream/run-markers.ts) builds it. Temporal, DBOS and CF Workflows check for the run's own marker before writing it on an activity / step retry, and Temporal fixes its timestamp outside the retried body, so a re-emission is byte-identical (chunk.run-started-unique). RunMetadata.startSequence is the head at the commit, so run_started.seq > startSequence (ordering.run-started-after-commit). The ai-sdk transformer drops it (no UI event).

interface RunStartedChunk {
  type: 'run_started';
  runId: string;
  // + Base fields
}

RunTerminatedChunk

Cloudflare Workflows only. It names a run whose instance was killed, so a client sees a typed end rather than text stopping mid-word:

• 'owner_unknown_timeout': the current run's instance stayed unknown past cfwUnknownTerminateAfterMs; the next entry terminates it and writes this chunk before its own run_started;
• 'admin_terminate': the operator ran executor.terminateRun(sessionId) (for an operator-paused instance); it is written right after the instance is terminated, on a stream that is still active or paused and owned by the session.

It is exempt from the chunk fence (it names the killed run while another writer writes it). UI: data-run-terminated with { runId, reason } inside data (the only shape valid across ai 6.x).

interface RunTerminatedChunk {
  type: 'run_terminated';
  runId: string; // the killed run
  reason: 'owner_unknown_timeout' | 'admin_terminate';
  // + Base fields
}

RunInterruptedChunk

Emitted when the agent is interrupted. The agent can be resumed via executor.resume().

interface RunInterruptedChunk {
  type: 'run_interrupted';
  runId: string;
  checkpointId: string | null; // null if checkpointing failed
  reason?: string;
  // + Base fields
}

RunResumedChunk

Emitted when an interrupted/paused agent is resumed.

interface RunResumedChunk {
  type: 'run_resumed';
  runId: string;
  fromCheckpointId: string | null;
  fromStepCount: number;
  mode: string; // 'continue' | 'with_message' | 'with_confirmation' | 'from_checkpoint'
  // + Base fields
}

RunPausedChunk

Emitted when the agent has paused (typically waiting for user confirmation). A stream manager synthesizes it on pauseStream: it is stream-level (agentId: '') and its runId is the pausing run (the pause fence's run, else the stream's owner), never the run the stream's first writer was created with (RM-109; ai-sdk forwards it as data-run-paused.runId).

interface RunPausedChunk {
  type: 'run_paused';
  runId: string;
  reason: string;
  pendingToolName?: string;
  pendingToolCallId?: string;
  // + Base fields
}

CheckpointCreatedChunk

Defined in the schema, but no runtime writes it today. To find the latest checkpoint, call stateStore.getLatestCheckpoint(sessionId) or listCheckpoints(sessionId).

interface CheckpointCreatedChunk {
  type: 'checkpoint_created';
  runId: string;
  checkpointId: string;
  stepCount: number;
  // + Base fields
}

StepCommittedChunk / StepDiscardedChunk

Signal whether staged changes were committed or discarded at a step boundary. Core emitStepCommitted / emitStepDiscarded build them.

step_committed is written after each step commit (on Temporal, DBOS and CF Workflows in the same activity / step as the commit, and re-written by an idempotent re-invocation that finds the commit landed but the marker missing). Its checkpointId is the step's kind = 'step' checkpoint, and committedAttemptId names the LLM attempt whose chunks are the step's final attempt. Ground truth for whether a step committed is the checkpoint, not this marker: (run r, step s) is committed iff a step checkpoint with that runId and stepCount exists, and discarded iff none exists and the run has ended in the stream.

interface StepCommittedChunk {
  type: 'step_committed';
  runId: string;
  stepId: string;
  checkpointId: string;
  /** The attempt that won ('' when the runtime has none to name; such a commit changes no attempt). */
  committedAttemptId: string;
  // + Base fields
}

interface StepDiscardedChunk {
  type: 'step_discarded';
  runId: string;
  stepId: string;
  reason: string;
  // + Base fields
}

StreamResyncChunk

Emitted after the server has discarded history or stream chunks, so clients should reload state (e.g. from a /snapshot endpoint). Every runtime writes it the same way (ruling R34): only after the entry WON its run-start and wrote its entry state, so it always follows the run's run_started and precedes the run's content; a losing entry writes none. It carries truncateCheckpointId, the truncate checkpoint the run-start commit wrote, when the rewind removed rows; checkpointId / messageCount / fromSequence then describe that truncate (else the restored target). Exactly one per rewinding run:

• retry() (reason: 'retry', followed by a run_resumed) on every runtime.
• resume({ mode: 'from_checkpoint' }) (reason: 'rollback', RM-60) on runtime-js, Temporal, Cloudflare Workflows and DBOS, every time. The stream still holds the rewound-away steps' chunks: no runtime removes chunks of an earlier turn (the cleanup below is floored at the resumed run's startSequence, and Temporal cleans nothing). The Cloudflare DO exposes no from_checkpoint (/resume takes only continue); its rewind is the truncating /retry.
• any other resume() (reason: 'crash_recovery') on runtime-js, the Cloudflare DO, Cloudflare Workflows and DBOS, only when it removed uncommitted rows past the latest checkpoint or the resumed run's orphan chunks.
• a companion re-spawn (reason: 'respawn', CP-142) on the Cloudflare DO: a failed / terminated persistent companion re-spawned by name, or a first spawn that finds a session an earlier incarnation of its parent left (CP-205), starts on a fresh session its DO wiped (the old generation's chunks are gone, the sequence stays monotonic); basis genesis (checkpointId: '', counts 0). Other runtimes delete the old session and start a new one, with no resync.
• The branch reason is defined but unused.

Readers. The chat host never serves the stale chunks to a resuming client: every read is floored at the current run's startSequence (handleChatStream), and the snapshot is checkpoint-anchored, so a page refreshed during or after the rewind renders the rewound log. A client that was already showing the pre-rewind conversation re-anchors on the resync (useHelixChat({ resync }) / useAutoResync: stop, latest snapshot, resume from its cursor). The snapshot names the session's latest truncate (FrontendSnapshot.latestTruncateCheckpointId); a page seeded from a snapshot read after the rewind's run-start commit replays the resync from that snapshot's cursor, and resyncCoveredBySnapshot makes the hooks skip it (pass the seed snapshot as useHelixChat({ resync: { snapshot } }) / useAutoResync({ snapshot })).

This chunk is not emitted on the happy path. In particular, it is not part of the client-executed-tool continuation flow: on the Cloudflare DO, Cloudflare Workflows (its __submit__ continuation instance) and DBOS runtimes the run auto-continues after a submit and streams the continuation live (the submit answers new_run / deferred, and the AI SDK attaches to the running stream without a resync). See Client-Executed Tools for the per-runtime continuation model.

Run-scoped orphan cleanup (cleanupToStep minSequence). Orphan stream-chunk cleanup on resume is bounded to the resumed run. Step numbers reset per turn (each turn restarts step counting at 1), but stream sequence numbers are monotonic across the whole per-session stream. A naive cleanupToStep(checkpoint.stepCount) would therefore delete prior turns' committed chunks — they carry per-turn step numbers greater than the current turn's low stepCount — spuriously tripping cleanedUpOrphans and emitting a bogus crash_recovery resync on every multi-turn client-tool resume. To prevent this, runtimes pass the resumed run's startSequence as the cleanupToStep minSequence floor; only chunks with sequence > minSequence are eligible for deletion, so earlier turns are protected while genuine within-run orphans (a crashed step's chunks) are still removed. All four stream managers (in-memory, Cloudflare DO SQLite, Cloudflare StreamDurableObject, Redis) honor the floor; the cross-store contract test pins the behavior. Who cleans: a resume() on runtime-js (and the Cloudflare DO), Cloudflare Workflows and DBOS, floored at the previous run's startSequence, after its run-start won; an interrupt on runtime-js (and the Cloudflare DO) and Cloudflare Workflows, floored at the interrupted run's startSequence. So the stopped step's chunks a live reader received are gone from the stream once the interrupt ends it: a replay read afterwards (a reconnect, a late reader from 0) holds the run's committed steps only, and an attached reader is told with rewound. No runtime cleans stream chunks on retry(), and Temporal cleans none at all.

fromSequence accuracy. fromSequence is the streamSequence of the checkpoint the resync describes (the run-start truncate checkpoint when the rewind removed rows, else the restored target). Every commit records the live stream head (core resolveCommitSequence; -1 with no stream; a failed head read is the typed state_checkpoint_sequence_unavailable, never a silent 0), on every runtime and store, so fromSequence is a real cursor. Earlier releases recorded 0 on saveState checkpoints (RM-37), on every DBOS checkpoint (RM-39) and on a failed head read (DI-21). A client still re-anchors from the snapshot, not from fromSequence.

A suspension commit covers its frames. When a step suspends on client tools or approval gates, its pending calls' frames (tool_start / tool_approval_request) are written BEFORE the step's suspension commit. That commit's streamSequence is therefore at or above every frame of the calls its assistant row holds (spec P1). A client seeded from that snapshot never receives those calls again past the cursor; before this fix it would run a re-delivered call, have the part reset by the re-sent tool-input-available, and never submit. JS and Temporal write the frames during the step's tool dispatch, before their suspension commit. DBOS and CF Workflows write the run's suspended_* status first, then the frames, then the suspension commit, then pause the stream. On DBOS each waiting call's whole prologue (beforeTool, the pending registration, the frame) runs before the commit, and onAgentSuspended fires after it. The e2e checkpoint-truth matrix checks it on every backend as checkpoint.suspension-covers-frames. A finishWith call deferred by a client call is not covered: it announces itself in the continuation run.

interface StreamResyncChunk {
  type: 'stream_resync';
  checkpointId: string;
  stepCount: number;
  messageCount: number;
  fromSequence: number;
  /** The rewind's run-start `truncate` checkpoint (R34); absent when nothing was truncated. */
  truncateCheckpointId?: string;
  reason: 'crash_recovery' | 'rollback' | 'branch' | 'retry' | 'respawn';
  // + Base fields (runId: the rewinding run, stamped by its writer)
}

The chunk has no runId field of its own: it carries the base runId of the run that wrote it, which the UI data-stream-resync event forwards when present (see UI events). With no target at all (a genesis retry), checkpointId is '' and messageCount / fromSequence are 0.

SuspensionMarkerChunk (Helix-internal)

Emitted by Temporal only, whose executor handle observes a suspension by watching the stream (the workflow itself returns at the suspension). Temporal writes it after the suspension commit and the run's suspended_* status, and pauses the stream after it. Lets the executor's handle observe via stream-watching that the run is suspended and project the right 'suspended_*' status from handle.result() without re-deriving from persisted state.

interface SuspensionMarkerChunk {
  type: 'suspension_marker';
  /** Suspension kind, matching `AgentResult.status`. */
  kind: 'suspended_client_tool' | 'suspended_awaiting_children' | 'suspended_step_partial';
  /**
   * Suspension payload, matching `AgentResult.suspended` shape.
   * - `suspended_client_tool`: { toolCallIds }
   * - `suspended_step_partial`: { toolCallIds, stepId }
   * - `suspended_awaiting_children`: { children }
   */
  payload:
    | { toolCallIds: string[]; stepId: string }
    | { toolCallIds: string[] }
    | { children: SuspendedChildWaitPayload[] };
  // + Base fields
}

@internal Helix coordination chunk — NOT part of the public stream surface. The binding-layer transformer filters this out of v6 streams; AI SDK consumers see suspension via existing tool_start, tool_approval_request, and subagent_start chunks.

Control message: stream_end

The only non-data variant. Lives on streamMessageSchema() (the union of data chunks + control messages).

interface StreamEnd {
  type: 'stream_end';
  streamId: string;
  timestamp: number;
  finalOutput?: unknown;
}

Per-runtime chunk-emission parity

Chunk type	runtime-js	runtime-cloudflare (DO)	runtime-cloudflare (Workflow)	runtime-temporal	runtime-dbos
text_delta / thinking / tool_arg_stream_*	yes	yes	yes	yes	yes (RM-55)
tool_start / tool_end	yes	yes	yes	yes	yes
tool_end.providerExecuted field	yes (v7)	yes (v7)	yes (v7)	yes (v7)	yes (v7)
tool_approval_request / tool_approval_response	yes (v7)	yes (v7)	yes (v7)	yes (v7)	yes (DBOS-native)
subagent_start / subagent_end	yes	yes	yes	yes	partial
state_patch / custom / error / output	yes	yes	yes	yes	yes
step_start (an attempt's first chunk)	yes	yes	yes	yes	yes
step_end	adapter onStepEnd only (none with llm-vercel, RM-48)	same	same	same	same (RM-49)
run_started	yes	yes	yes	yes	yes
run_terminated	n/a	n/a	yes (unknown timeout, admin terminate)	n/a	n/a
run_interrupted	yes	yes	yes	yes	no (CP-17)
run_resumed	resume, retry, recovery	resume, retry (via JS)	resume (not recovery), retry	retry only	retry only
run_paused (via pauseStream)	yes	yes	yes	yes	yes
checkpoint_created	no	no	no	no	no
step_committed	yes	yes	yes	yes	yes
step_discarded	no	no	yes	no (RM-59)	yes (interrupt, dead-owner settle; RM-57)
stream_resync	retry, rollback, crash_recovery	retry, crash_recovery, respawn	retry, rollback, crash_recovery	retry, rollback	retry, rollback, crash_recovery
suspension_marker	n/a	n/a	no	yes	n/a

suspension_marker is emitted by Temporal, whose executor handle observes a suspension by watching the stream (its workflow returns at the suspension and a later resume() starts a new one). JS, CF DO, CF Workflows and DBOS surface suspension via the AgentResult.status projection instead. Every runtime now pauses the stream at a suspension (pauseStream, which writes run_paused), after the run's suspended_* status (ordering.run-status-before-paused); see Suspension order. The RM-55 / RM-56 records (DBOS text and tool-input deltas; no stream pause on DBOS, Temporal and CF Workflows) describe the pre-fix behaviour and are closed by the conformance owner.

Temporal writes no step_discarded and no crash_recovery resync (RM-59)

An abort() mid-step on runtime-temporal leaves the uncommitted step's chunks on the stream with no step_discarded marker, and a Temporal resume() never cleans orphan chunks. Readers that use the step ground truth (a step checkpoint per committed (run, step)) are unaffected; a reader that relies on the marker is not. DBOS writes step_discarded on its interrupt path and when it settles a run whose owner died or was cancelled (an interrupt or abort(), RM-57). See RM-57 and RM-59 in the conformance findings.

Stream Lifecycle

Creation

const writer = await streamManager.createWriter(streamId, agentId, agentType);

Writing

await writer.write({
  type: 'text_delta',
  delta: 'Hello',
  agentId: sessionId, // the writing session
  agentType: 'assistant',
  timestamp: Date.now(),
  step: 0,
});

await writer.close();

A write resolves with what it did (StreamWriteResult): { written: true }, or { written: false, reason } when the manager deliberately dropped the chunk:

• fenced: a superseded run's own chunk (see The chunk fence);
• guest_inactive: a child's chunk to an ended, failed or paused parent stream (CP-129);
• stopped_run: CP-193, a run's own chunk to a stream its recorded stop has ended (the CF DO's stop deadline ended the run under a tool that ignores its abort signal and keeps emitting). The tool is not interrupted by a throw; it runs on to its end and every later write is dropped, logged once per run at warn (core.stream.stopped_run_write_dropped). Only the CF DO manager reads stops (DOStreamManagerOptions.isRunStopped); every other manager refuses that write, as before;
• duplicate_remote_chunk: a chunk proxied from a remote child that the stream already holds (see Remote child chunks are written once).

A deliberate drop is never a throw. Every first-party manager (memory, Redis, the Cloudflare DO managers) reports the result on every write; a writer that resolves with nothing (a test fake, a third-party writer) reports nothing, and wrappers forward the inner write's result.

Size (RM-90). A manager refuses a chunk whose stored JSON is over its declared limits.maxChunkBytes with a typed state_size_exceeded (operation: 'stream_chunk', non-retryable). It does so BEFORE allocating a sequence, so a refused chunk leaves no gap. An end is never refused: a part of its terminal payload over the limit (the final output, state or usage) is not stored, and the end names it in StreamMetadata.endOmitted and the end frame's omitted. Defaults: the Cloudflare managers 1.8 MB (under SQLite's 2 MB value cap); memory and Redis Infinity. A model's oversized delta (text_delta, a streaming thinking delta, tool_arg_stream_delta) is split at the producer by withChunkSplit into pieces that fit, before stamping or proxying. A chunk carrying remoteSource is never split.

The CF DO writer yields (CP-193). A Durable Object's SQLite and its WebSocket / SSE fan-out are synchronous, so a DOStreamWriter write does no I/O, and await writer.write() would resolve as a microtask. A run streaming output with no I/O of its own (a tool emitting a large result through ctx.emit) would then be one unbroken microtask chain: the DO dispatches no other request (/status, a stop, a submit), and delivers nothing to its readers, until the run ends. So every DO_STREAM_YIELD_EVERY_WRITES (2048) writes, or DO_STREAM_YIELD_EVERY_BYTES (1 MiB) of serialized events, whichever comes first, a write yields to the event loop (setTimeout(0)) after its chunk is stored and broadcast (a dropped or refused write counts too): queued requests and alarms are dispatched there, and the slice's implicit SQLite transaction commits. Each yield costs a few milliseconds, so the slice is sized to keep that to a few percent of a flood's throughput; a request queued behind a flood waits about one slice (tens of milliseconds). While an SSE client is attached, the byte slice is at most half the connection's unread budget (SSEConnectionManager.unreadBudgetBytes, 256 KiB by default): a client reads only when the DO yields, and a slice larger than the budget would get every live reader evicted (CP-26) within one slice. While a reader is attached, a write whose frame would take the slice past that limit yields first, so a reader that reads at every yield has read the slice's earlier frames before a large one is sent. The slice is counted, not timed: in production Workers Date.now() is frozen while code runs and advances only across I/O, so a timed slice would never fire there (local workerd advances it, so it would pass locally). A write's sequence is taken synchronously, before any yield, so sequence order and the chunk fence are unchanged; what is new is that a request (a stop, a status read) can now run between two writes of one run. The other managers do real I/O per write (Redis, the CF Workflows binding to its stream DO) except the in-memory manager, which still never yields (FU-CP193-INMEMORY-STREAM-YIELD).

Reading

const reader = await streamManager.createReader(streamId);

for await (const chunk of reader) {
  // Process chunk
}

Ending

// Completion, with the run's terminal status (required, CP-87; recorded as metadata.endStatus)
await streamManager.endStream(streamId, undefined, { status: 'completed' });
await streamManager.endStream(streamId, output, { status: 'completed' }); // With final output
await streamManager.endStream(streamId, undefined, { status: 'interrupted' }); // A stopped run

// With error
await streamManager.failStream(streamId, 'Something went wrong');

// Runtimes fence every transition on the run (CP-85): a no-op once a newer run owns the stream
await streamManager.endStream(streamId, output, { status: 'completed' }, { expectRunId: runId });
await streamManager.failStream(streamId, message, errorDetail, { expectRunId: runId });
await streamManager.pauseStream(streamId, { expectRunId: runId }); // writes run_paused

Stream End Sentinel

A special marker indicates stream completion:

const STREAM_END_SENTINEL = Symbol.for('helix.stream.end');

interface StreamEndSentinel {
  __streamEnd: true;
  status: 'ended' | 'failed';
  error?: string;
}

Readers yield chunks until they encounter the sentinel.

Resumability

Sequence Numbers

Each chunk can have a sequence number for resumability:

interface StoredChunk {
  chunk: StreamChunk;
  sequence: number; // Monotonically increasing
}

Resumable Reader

const reader = await streamManager.createResumableReader(streamId, {
  fromSequence: 100, // Resume from sequence 100
});

for await (const { chunk, sequence } of reader) {
  // sequence can be used as Last-Event-ID
}

Stream Status

type ResumableStreamStatus = 'active' | 'paused' | 'ended' | 'failed';

interface ResumableStreamReader extends StreamReader {
  readonly currentSequence: number; // last chunk yielded
  readonly totalChunks: number;
  readonly latestSequence: number; // head at reader creation
  readonly catchUpHead: number; // the backlog boundary
  readonly caughtUp: Promise<void>; // resolves once the backlog was yielded
}

A reader treats paused like a terminal status: it drains what is past its cursor and ends. "Suspension committed" is signalled by the stream's paused status, the last write of the suspension sequence (a run's suspended_* status alone does not imply it, because DBOS and CF Workflows write it before the commit).

Cursor semantics

A snapshot's streamSequence is the streamSequence of the ONE checkpoint it was read from (cp.streamSequence), and its messages are exactly [0, cp.messageCount). Every commit records the live head after the chunks that describe its rows were written (see ./session-model.md §Commits and checkpoints), so a client that seeds from the snapshot and resumes from its cursor sees each committed content chunk exactly once:

• Resume tails strictly after the cursor. createResumableReader(streamId, { fromSequence }) yields chunks with seq > fromSequence on every manager.
• catchUpHead / caughtUp. Every resumable reader captures, atomically with its initial history read, the highest sequence that read covers (catchUpHead), and caughtUp resolves once every retained backlog chunk was yielded. The chat handler uses it as a deterministic backlog / live boundary (the superseded-attempt filter drops a superseded attempt's backlog as a whole) instead of a quiet period. Push readers (SSE) learn it from the server's history replay. Contract cases in streamManagerContractTests.
• waitForLive. Between a run's run-start commit and its stream claim, the snapshot already shows the run as live (the run decides the status) while the stream still shows the previous turn's ended / failed, or does not exist yet. A reader created with waitForLive: true parks instead of ending until the stream goes live (active at attach, a transition to active, a chunk past the cursor, or the stream being created), then applies the normal terminal rules. handleChatStream resumes with it and bounds the wait with liveWaitTimeoutMs (default 30 000 ms; past it, a 503 { code: 'transport_stream_drop', cause: 'run_not_live', retryable: true }). Contract cases WL.*.
• Legacy sessions (latest checkpoint kind: null) serve the complete live log with the stream head as the cursor until their first new commit.

Stream ownership and claims

StreamMetadata.currentRunId records which run owns a stream. StreamManager.claimStream?(streamId, runId, expect?) is a compare-and-set (CP-85): with expect: { owner } it sets the owner only while the current owner (null = none) is expect.owner, and resolves true when runId owns the stream afterwards (contract case RF.claim.cas). Runtimes call it only through core claimStreamForCommittedRun({ …, mode }) (packages/core/src/stream/run-markers.ts), and only for a run that committed its run start:

• entry mode, right after the run-start commit: a current-run read that does not show the run while its own record is live is a stale read and throws the retryable StreamClaimStaleReadError (never a silent skip); a run that already ended claims nothing and never revives its stream; the stream is made live if it is not active.
• terminal mode, before a committed run's fenced end / fail / pause: a run that is no longer current claims nothing, and a stream is never taken over from an owner whose run is live or newer.
• A child that writes into its parent's stream never claims it (R41).

endStream / failStream / pauseStream take an optional fence { expectRunId }: once a newer run owns the stream, a fenced transition by an older run is a no-op.

The chunk fence

Once a newer run claimed a stream, every stream manager (memory, Redis, the Cloudflare DO managers) drops an own-session chunk stamped with another runId (core chunkFenceRunId(streamId, chunk)), including a superseded run's late output / error. Not fenced: a child's chunk in its parent's stream (agentId !== streamId), a chunk with no runId (legacy writers), and run_terminated. On runtime-js the window before the new run's claim lands is covered by the lease gate (gateWriterOnLease, see ./session-model.md §The JS run lease).

Remote child chunks are written once

A parent proxies a remote child's chunks into its own stream, each stamped with remoteSource (childSessionId, childSequence). A forwarder that re-attaches (a crash, a Durable Object eviction, an activity or step retry) re-reads the child's stream from the progress it persisted, which can lag what it wrote, and a zombie forwarder can still be running beside its replacement. So the stream decides, inside the write's own atomic read (core decideStreamWrite, RM-86):

• it keeps a high-water per remote child run, keyed on (remoteSource.childSessionId, remoteSource.childRunId) (core remoteChunkDedup / remoteHighWaterKey). childRunId is the runId the child stamped, which the forwarder records before the parent's own writer can fill a missing runId with the parent's run. The key is the child's RUN, not its session: a child's sequences only increase within one of its runs, while a continued companion or a respawned child is a new run, so its chunks are never mistaken for duplicates;
• a proxied chunk at or below its high-water is dropped (duplicate_remote_chunk); an appended one raises it, only once the chunk is stored;
• the high-water is cleared when a new run claims the stream (its clients read from its run-start snapshot, past every earlier copy, so a chunk re-forwarded after it must land again) and by resetStream. A re-claim by the run that already owns the stream (a recovery keeps its run) keeps it;
• a proxied chunk with no childRunId (its producer stamped no run: a broken contract, since every executor stamps its run) is appended without the dedup, with one core.stream.remote_chunk_unstamped warn per stream. A lost chunk is worse than a duplicate.

This is exact because every forwarder (core executeRemoteSubAgentDispatch and Temporal through it, runtime-js and the CF DO, CF Workflows, DBOS) records a chunk as forwarded only after its write resolved (RM-89): a re-attach never seeds past a chunk the stream lacks, so per child run the stream holds a gap-free prefix and "at or below the high-water" means "already written". Contract cases: RD.* in core testing/stream-manager-contract.ts.

Attempts and the final attempt

Attempts are scoped per (agentId, runId, step). Before the step commits, the current attempt is the one that STARTED last, ordered by the stream position of each attempt's step_start (attempts with no step_start, from legacy writers, fall back to first-seen order). Once the step's step_committed names an attempt that streamed in that step, that attempt is final for good; a commit that names no attempt (''), or one never seen in the step, changes nothing. Only what an LLM attempt streams is attempt-scoped (text_delta, thinking, tool_start, tool_arg_stream_*, source_url, source_document, file, step_start, step_end); tool results and every other chunk pass through. The ai-sdk SupersededAttemptFilter drops a superseded attempt's content from a stream the server replays, and the transformer emits data-attempt-superseded when the current attempt switches live (packages/ai-sdk/src/transformer/attempts.ts).

UI events for run lifecycle chunks

The ai-sdk transformer (packages/ai-sdk/src/transformer/stream-transformer.ts) maps them to data-* parts (every payload is valid against ai's strict UI chunk schemas, DI-20):

Helix chunk / event	UI part
run_started	dropped (no UI event)
step_committed	data-step-committed { runId, stepId, checkpointId } (committedAttemptId is not forwarded)
step_discarded	data-step-discarded { runId, stepId, reason }
stream_resync	data-stream-resync { checkpointId, stepCount, messageCount, fromSequence, reason, truncateCheckpointId?, runId? }
run_terminated	data-run-terminated { runId, reason } (inside data)
an attempt switch	data-attempt-superseded { reason: 'attempt_superseded', runId, agentId, step, attemptId, supersededAttemptId }
a replayed prefix (C-1)	transient data-helix-replay { phase: 'begin' | 'end', messageId }
a result, earlier call	data-tool-output-elsewhere { reason: 'tool_output_elsewhere', toolCallId, toolName, runId? } (see below)

attemptId is never forwarded on content events; it appears only inside the data-attempt-superseded payload. On a switch the transformer first closes open text / reasoning blocks. The chat handler also announces superseded attempts the client already received, once after start.

A tool_end, tool_output_error or denied tool_approval_response becomes a tool-output-* part only for a call the response message owns: its input was emitted in this response (live, replayed, or in the C-1 prefix), or it is in StreamTransformerOptions.continuedToolCallIds (the client-held message a continuation is appended to). ai throws No tool invocation found on a result for any other call and fails the whole response, so the transformer emits data-tool-output-elsewhere instead, once per call; the client re-anchors on the snapshot. The runtime stream is unchanged: every observer of tool_end still sees it.

The replay-prefix resume (C-1)

ai 6.0.281 starts a resumed response from an empty message that replaces the last one, so seeding the whole snapshot and resuming only the tail would drop the in-flight message's committed steps. snapshotResumeOptions(snapshot) (@helix-agents/ai-sdk) returns { resume, resumeFromSequence?, existingMessageId?, replayPrefix?, messages }: it resumes only on active / paused, and with the replay prefix it seeds the conversation WITHOUT the resumeMessageId message. The reconnect sends X-Existing-Message-Id and X-Helix-Replay-Prefix: 1; the server reads ONE snapshot (HandleChatStreamDeps.snapshotOptions), replays that message's committed parts between data-helix-replay begin / end markers, then tails after THAT snapshot's cursor. Only a client-supplied id is replayed; an id the snapshot does not hold (and is not its resumeMessageId) gets 409 { code: 'state_not_resumable', cause: 'resume_message_unknown' }. Clients that cannot suppress replayed client tools pass replayPrefix: false. See the upgrade guide.

Wire Format

SSE carries these chunks in the stream protocol's named frames (chunk, end, fail, truncated, rewound, caught-up; wire contract v2, encoded and decoded by core): see Stream wire, which also defines the terminal state machine (a body with no terminal frame is a StreamDropError).

WebSocket clients (the Cloudflare Durable Object's WS fan-out) receive the core StreamEvent shape:

interface StreamEvent {
  type: 'chunk' | 'end' | 'fail' | 'status' | 'truncated';
  // ... payload based on type
}

interface StreamChunkEvent {
  type: 'chunk';
  chunk: StreamChunk;
  sequence?: number;
}

interface StreamEndEvent {
  type: 'end';
  finalOutput?: unknown;
}

interface StreamFailEvent {
  type: 'fail';
  error: string;
}

interface StreamStatusEvent {
  type: 'status';
  // Lifecycle transition. Used as a wakeup ping for blocked readers when
  // the stream pauses/resumes — readers re-check metadata and do NOT emit a
  // chunk to the consumer.
  status: 'paused' | 'active';
}

interface StreamTruncatedEvent {
  type: 'truncated';
  // The `stepCount` boundary passed to `cleanupToStep`; chunks with
  // `step > truncatedAtStep` (above `minSequence` when present) were removed.
  truncatedAtStep: number;
  // The stream's sequence high-water mark at the moment of cleanup. Readers
  // gate the throw on `atSequence > lastYieldedSequence`. Optional for
  // wire compatibility.
  atSequence?: number;
  // The cleanup's run-scoping floor (`cleanupToStep`'s `minSequence`): only
  // chunks above it were eligible. A reader prunes its buffer by the same
  // rule. Absent means no floor.
  minSequence?: number;
}

G4 — truncation surfacing

The 'truncated' event implements StreamManager concurrency guarantee #4 (G4): when a cleanupToStep(N) removes chunks past an attached reader's cursor, PUSH transports broadcast a control event so the attached reader surfaces a StreamTruncatedError on its next iteration — the WebSocket truncated StreamEvent above, and on SSE the rewound{atStep, atSequence, minSequence?} frame (the CF DO and the StreamServer DO; see Stream wire). This is best-effort — a reader that attaches after cleanup (a fresh replay) won't see the event and instead resumes from the surviving chunks.

MARKER/POLL transports do not need the wire event: they detect the same condition via a stored marker consulted on each next() — the Redis hash field __truncated_at_step, the in-process Durable Object SQLite truncated_at_step meta, and an equivalent O(1) in-process field in the in-memory manager.

A cleanup never costs an attached reader a chunk the store kept (contract cases G4.rebuild-keeps-live-chunks and G4.floor-kept-backlog-survives):

• Push readers prune by the cleanup's own rule. An SSE reader (the binding-side reader, the CF DO client) drops a buffered chunk only when its step > atStep AND its sequence is above the frame's minSequence. The resume-time cleanup of runtime-js (and the CF DO, which runs it), DBOS and CF Workflows is floored at the previous run's startSequence, and their interrupt-time cleanup at the interrupted run's (when its record reads), so a step-only prune dropped earlier turns' chunks the DO kept. Temporal never calls cleanupToStep.
• The Redis rebuild reads its snapshot and the writer's counter in one MULTI. While the marker is set, a drained reader rebuilds its buffer from LRANGE on the main connection, while pub/sub delivers new chunks on the subscriber connection. A pending pub/sub chunk is dropped as cleaned up only if the snapshot could hold it (sequence ≤ the counter in the same transaction) and does not. Before T1-A it was dropped whenever it was missing from the snapshot, so a chunk written just after the LRANGE was lost. A resume's cleanup sets the marker even when it removes nothing, so every continuation run on Redis (runtime-js, DBOS) streamed under that rebuild.

SSE Encoding

id: 1
event: chunk
data: {"sequence":1,"chunk":{"type":"text_delta","delta":"Hello", ...}}

id: 2
event: chunk
data: {"sequence":2,"chunk":{"type":"text_delta","delta":" world", ...}}

event: end
data: {"status":"completed","output":{"answer":42}}

Every connection ends with exactly one terminal frame (end, fail or truncated); see Stream wire.

Resume Headers (v7)

Reconnecting clients negotiate resume position via these headers (see prepare-helix-chat-request.ts):

• X-Resume-From-Sequence — Client-supplied resume position; server honors via extractResumePosition(headers).
• Last-Event-ID — Standard SSE; mapped to the same resume semantics as X-Resume-From-Sequence if both are absent (server falls back to Last-Event-ID).
• X-Existing-Message-Id — Client-supplied existing message ID so the reconnected stream's start event reuses it (avoids creating a duplicate assistant message client-side on reconnect).
• X-Helix-Replay-Prefix: 1 — With X-Existing-Message-Id: replay that message's committed prefix from one snapshot, then tail after that snapshot's cursor (see The replay-prefix resume).

Filtering Streams

Utility functions for stream manipulation:

import {
  filterByAgentId,
  filterByAgentType,
  filterByType,
  excludeTypes,
  filterWith,
  combineStreams,
  take,
  skip,
  collectText,
  collectAll,
} from '@helix-agents/core';

// Filter by agent
const agentChunks = filterByAgentId(stream, 'run-123');

// Only text chunks
const textChunks = filterByType(stream, ['text_delta']);

// Exclude thinking
const noThinking = excludeTypes(stream, ['thinking']);

// Custom filter
const important = filterWith(stream, (chunk) => chunk.type === 'error' || chunk.type === 'output');

// Combine multiple streams
const combined = combineStreams([stream1, stream2]);

// Collect all text
const fullText = await collectText(stream);

Type Guards

Runtime type checking for chunks (full set — one per chunk variant):

import {
  isTextDeltaChunk,
  isThinkingChunk,
  isToolStartChunk,
  isToolEndChunk,
  isSubAgentStartChunk,
  isSubAgentEndChunk,
  isCustomEventChunk,
  isStatePatchChunk,
  isErrorChunk,
  isOutputChunk,
  isToolArgStreamStartChunk,
  isToolArgStreamDeltaChunk,
  isToolArgStreamEndChunk,
  isToolInputErrorChunk,
  isToolOutputErrorChunk,
  isSourceUrlChunk,
  isSourceDocumentChunk,
  isStepStartChunk,
  isStepEndChunk,
  isFileChunk,
  isRunInterruptedChunk,
  isRunResumedChunk,
  isRunPausedChunk,
  isCheckpointCreatedChunk,
  isStepCommittedChunk,
  isStepDiscardedChunk,
  isStreamResyncChunk,
  isSuspensionMarkerChunk,
  isStreamEnd,
} from '@helix-agents/core';

for await (const chunk of stream) {
  if (isTextDeltaChunk(chunk)) {
    process.stdout.write(chunk.delta);
  } else if (isToolStartChunk(chunk)) {
    console.log(`Tool: ${chunk.toolName}`);
  }
}

Validation

Zod schemas for runtime validation:

import { streamChunkSchema, streamMessageSchema } from '@helix-agents/core';

// Validate a chunk
const result = streamChunkSchema().safeParse(data);
if (result.success) {
  const chunk: StreamChunk = result.data;
}

Implementation Notes

Memory Streams

In-memory streams use arrays and async iterators:

class InMemoryStreamManager {
  private streams = new Map<
    string,
    {
      chunks: StreamChunk[];
      readers: Set<() => void>;
      status: 'active' | 'ended' | 'failed';
    }
  >();
}

Redis Streams

Redis implementation uses Redis Streams (XADD/XREAD):

// Writing
await redis.xadd(
  `stream:${streamId}`,
  '*',
  'chunk',
  JSON.stringify(chunk),
  'sequence',
  sequence.toString()
);

// Reading with blocking
const entries = await redis.xread('BLOCK', 5000, 'STREAMS', `stream:${streamId}`, lastId);

Durable Object Streams

Cloudflare Durable Objects provide coordination:

class StreamServer {
  private chunks: StreamChunk[] = [];
  private waiters: Set<(chunk: StreamChunk) => void> = new Set();

  async write(chunk: StreamChunk) {
    this.chunks.push(chunk);
    for (const waiter of this.waiters) {
      waiter(chunk);
    }
  }
}

See Also

• Streaming Guide
• State Tracking
• Frontend Integration
• Sub-Agent Execution — subagent_start / subagent_end field schema must match this file.