Real-Time Streaming

Streaming provides real-time visibility into agent execution. The framework emits typed events for text generation, tool execution, sub-agent activity, custom events, and more.

Overview

When an agent executes, it produces a stream of chunks:

typescript

const handle = await executor.execute(agent, 'Research AI agents');
const stream = await handle.stream();

for await (const chunk of stream) {
  console.log(chunk.type, chunk);
}

Each chunk has:

type - Discriminator for the chunk kind
agentId - Which agent produced this chunk
agentType - The agent's type name
timestamp - When the chunk was created
step - The LLM step the chunk belongs to
runId - The run that wrote it. Every chunk a run's executor writes is stamped with it (withRun)
attemptId - On an LLM attempt's chunks: which attempt produced them (withAttempt). A retried step streams again under a new attemptId (see Run and step markers). It is never forwarded to the AI SDK UI stream
Type-specific fields

A run's first chunk is its run_started, written once by the run that won its run-start commit. Chunks read back from a resumable reader also carry their stream position as _sequence.

Stream Chunk Types

text_delta

Incremental text from the LLM. Append deltas to build complete text.

typescript

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

Usage:

typescript

let fullText = '';

for await (const chunk of stream) {
  if (chunk.type === 'text_delta') {
    fullText += chunk.delta;
    process.stdout.write(chunk.delta); // Real-time display
  }
}

thinking

Reasoning/thinking content from LLMs that support it (Claude with extended thinking, OpenAI o-series).

typescript

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

Usage:

typescript

let thinkingBuffer = '';

for await (const chunk of stream) {
  if (chunk.type === 'thinking') {
    if (!chunk.isComplete) {
      // Streaming - accumulate
      thinkingBuffer += chunk.content;
    } else {
      // Complete block
      console.log('Thinking:', chunk.content);
      thinkingBuffer = '';
    }
  }
}

tool_start

Emitted when a tool begins execution.

typescript

interface ToolStartChunk {
  type: 'tool_start';
  agentId: string;
  agentType: string;
  timestamp: number;
  toolCallId: string; // Unique ID for this call
  toolName: string; // Tool being called
  arguments: Record<string, unknown>; // Arguments passed
}

Usage:

typescript

for await (const chunk of stream) {
  if (chunk.type === 'tool_start') {
    console.log(`Starting: ${chunk.toolName}`);
    console.log(`Args: ${JSON.stringify(chunk.arguments)}`);
  }
}

tool_end

Emitted when a tool completes (success or failure).

typescript

interface ToolEndChunk {
  type: 'tool_end';
  agentId: string;
  agentType: string;
  timestamp: number;
  toolCallId: string; // Matches tool_start
  toolName: string;
  result: unknown; // Tool's return value
  error?: string; // Present if tool failed
}

Usage:

typescript

for await (const chunk of stream) {
  if (chunk.type === 'tool_end') {
    if (chunk.error) {
      console.log(`${chunk.toolName} failed: ${chunk.error}`);
    } else {
      console.log(`${chunk.toolName} result:`, chunk.result);
    }
  }
}

subagent_start

Emitted when a sub-agent begins execution.

typescript

interface SubAgentStartChunk {
  type: 'subagent_start';
  agentId: string; // Parent agent ID
  agentType: string; // Parent agent type
  timestamp: number;
  subAgentType: string; // Sub-agent's type name
  subSessionId: string; // Sub-agent's unique session ID
  callId: string; // Tool call ID that spawned this
}

subagent_end

Emitted when a sub-agent completes.

typescript

interface SubAgentEndChunk {
  type: 'subagent_end';
  agentId: string; // Parent agent ID
  agentType: string; // Parent agent type
  timestamp: number;
  subAgentType: string;
  subSessionId: string;
  callId: string;
  result: unknown; // Sub-agent's output
}

Usage:

typescript

for await (const chunk of stream) {
  if (chunk.type === 'subagent_start') {
    console.log(`Delegating to ${chunk.subAgentType}...`);
  }
  if (chunk.type === 'subagent_end') {
    console.log(`${chunk.subAgentType} completed:`, chunk.result);
  }
}

Sub-agent chunk ordering

Sub-agent calls (created via createSubAgentTool or createRemoteSubAgentTool) appear to the LLM as tool calls with names prefixed subagent__<name>. Each invocation therefore emits both a sub-agent lifecycle pair (subagent_start / subagent_end) and a tool call pair (tool_start / tool_end):

tool_start    { toolCallId: 'X', toolName: 'subagent__researcher' }
subagent_start { callId: 'X', subAgentType: 'researcher', subSessionId: '…' }
  …proxied chunks from inside the sub-agent (tool_start/tool_end for
   inner tools, text_delta, etc.)…
subagent_end  { callId: 'X', result: <output> }
tool_end      { toolCallId: 'X', toolName: 'subagent__researcher', result: <output> }

The two are not redundant — they serve different consumers:

subagent_start / subagent_end are lifecycle markers for an optional sub-agent activity overlay in the UI.
tool_start / tool_end close the parent's subagent__<name> tool part. AI SDK consumers that build dynamic-tool message parts incrementally from streaming events use these to transition the part from state: 'input-available' → state: 'output-available'. Without the trailing tool_end, the part stays stuck in 'input-available' with no output, and any persistence layer that snapshots the live UI inherits the broken state.

The tool_end always follows subagent_end so the activity overlay closes before the tool part transitions to its final state.

A child still running after the parent's turn ends

A non-blocking companion can still be writing when the parent's turn ends or suspends. Its chunks to the parent's ended, failed or paused stream are then dropped (logged at debug as stream.guest_chunk_dropped) instead of failing the child. They land again once the parent's next run makes the stream live. The parent's own writes to a non-active stream still throw. See Sub-Agents → Streaming Integration for the full rule.

Failure-path semantics

When a sub-agent fails, the runtime emits tool_end with an error field (and result: undefined) — this transitions the parent's dynamic-tool part out of 'input-available' to a terminal state.

DBOS exception (DI-19)

On DBOS, an ephemeral sub-agent that ends failed without throwing (for example after forced-completion exhaustion) is reported as a success with undefined output, with no error (inferred from source, not reproduced by a test). See DI-19 in the conformance findings.

There are two distinct paths consumers see, both terminal:

Live stream: the AI SDK transformer's transformToolEnd emits a tool-output-available event with output: { error: <message> } (or { error, errorCode } for client-tool errors). The part state becomes 'output-available'. There is no 'tool-output-error' state for tool_end chunks — that variant is reserved for the separate tool_output_error chunk type.
Reload via convertToAISDKMessages: the converter consults the persisted ToolResultMessage's metadata.helixToolFailed flag (stamped by createToolResultMessage / createSubAgentResultMessage) and routes failure to state: 'output-error' with errorText: <message>. Older tool result messages (persisted before the flag existed) fall through to a content-shape backward-compat path.

These two states are different but both are terminal — UI components that branch on state === 'output-error' should also handle state === 'output-available' with output.error set, and vice versa. (A single canonical state across both paths is a reasonable follow-up but out of scope for this fix.)

Diagnosing a stuck `'input-available'` in production

If you see a sub-agent's tool part stuck in 'input-available' despite the runtime running a fixed version:

Check warn-logs for either of these structured events:
- [subagent] Failed to write tool_end on success path (runtime-js)
- [subagent] Failed to write tool_end in recordSubSessionResult (runtime-cloudflare or runtime-temporal)
These fire when writer.write rejected mid-emission (e.g. stream evicted, connection dropped). The persisted ToolResultMessage is unaffected — reloading the session via convertToAISDKMessages will reconstruct the correct terminal state. The live UI will remain stuck until reload.
Check Temporal / CF Workflow activity retries. The recordSubSessionResult activity is not idempotent across retries: on partial-write failure, the platform re-runs the activity, which re-emits both subagent_end and tool_end. The AI SDK transformer dedupes duplicate tool_end for the same toolCallId, but observability tooling that counts events at the chunk level may double-count. For Temporal specifically, configure recordSubSessionResult's activity retry policy with bounded maximumAttempts if you need stricter event-count semantics.
Verify runtime version compat. The tool_end emission ships in runtime-js, runtime-cloudflare, and runtime-temporal. Upgrading the parent runtime is what fixes the bug — remote-agent servers (separate processes hosting sub-agents via HTTP) can lag without consequences, since the parent emits tool_end based on what it receives, not what the remote emits.

custom

Custom events emitted by tools via context.emit().

typescript

interface CustomEventChunk {
  type: 'custom';
  agentId: string;
  agentType: string;
  timestamp: number;
  eventName: string; // Event name from emit()
  data: unknown; // Event payload
}

Emitting custom events:

typescript

// In tool execute function
await context.emit('progress', { step: 1, total: 5 });
await context.emit('file_uploaded', { name: 'doc.pdf', size: 1024 });

Consuming custom events:

typescript

for await (const chunk of stream) {
  if (chunk.type === 'custom') {
    switch (chunk.eventName) {
      case 'progress':
        console.log(`Progress: ${chunk.data.step}/${chunk.data.total}`);
        break;
      case 'file_uploaded':
        console.log(`Uploaded: ${chunk.data.name}`);
        break;
    }
  }
}

state_patch

RFC 6902 JSON Patch operations representing state changes.

typescript

interface StatePatchChunk {
  type: 'state_patch';
  agentId: string;
  agentType: string;
  timestamp: number;
  patches: JSONPatchOperation[];
}

interface JSONPatchOperation {
  op: 'add' | 'remove' | 'replace' | 'move' | 'copy' | 'test';
  path: string; // JSON Pointer path
  value?: unknown; // Value for add/replace
  from?: string; // Source for move/copy
}

Example patches:

typescript

// When tool does: draft.count = 5
{ op: 'replace', path: '/count', value: 5 }

// When tool does: draft.items.push('new')
{ op: 'add', path: '/items/-', value: 'new' }

// When tool does: delete draft.temp
{ op: 'remove', path: '/temp' }

Wire format note

Array appends use RFC 6902's /- end-of-array path (e.g., add /items/-), not absolute indices. This makes parallel-tool appends compose correctly regardless of arrival order. Any patch applier consuming state_patch chunks MUST support /-. Spec-compliant libraries (fast-json-patch, rfc6902) handle this natively; hand-rolled appliers must add support.

Applying patches (client-side):

typescript

import { applyPatch } from 'fast-json-patch';

let clientState = { count: 0, items: [] };

for await (const chunk of stream) {
  if (chunk.type === 'state_patch') {
    clientState = applyPatch(clientState, chunk.patches).newDocument;
    console.log('State updated:', clientState);
  }
}

error

Error events during execution.

typescript

interface ErrorChunk {
  type: 'error';
  agentId: string;
  agentType: string;
  timestamp: number;
  error: string; // Error message
  code?: string; // Error code for categorization
  recoverable: boolean; // Whether execution can continue
}

Usage:

typescript

for await (const chunk of stream) {
  if (chunk.type === 'error') {
    if (chunk.recoverable) {
      console.warn(`Warning: ${chunk.error}`);
    } else {
      console.error(`Fatal: ${chunk.error}`);
      break;
    }
  }
}

output

Final structured output when the agent completes successfully.

typescript

interface OutputChunk {
  type: 'output';
  agentId: string;
  agentType: string;
  timestamp: number;
  output: unknown; // Validated against outputSchema
}

Usage:

typescript

for await (const chunk of stream) {
  if (chunk.type === 'output') {
    // Agent completed with structured output
    const result = chunk.output as MyOutputType;
    console.log('Final result:', result);
  }
}

forced_completion

Diagnostic progress of the forced-completion phase of an outputSchema agent. Emitted started → attempt_started → (attempt_failed)* → exactly one terminal succeeded or exhausted (started → exhausted when a forced call is refused before the model is called, e.g. a missing completion tool or a changed prompt prefix; a lone exhausted for maxCompletionRetries: 0).

typescript

interface ForcedCompletionChunk {
  type: 'forced_completion';
  agentId: string;
  agentType: string;
  timestamp: number;
  phase: 'started' | 'attempt_started' | 'attempt_failed' | 'succeeded' | 'exhausted';
  trigger: 'terminal_text' | 'max_tokens' | 'max_steps';
  expectedToolName: string; // '__finish__' or the finishWith tool
  attempt: number;
  maxAttempts: number; // maxCompletionRetries at phase entry
  stopReason?: string;
  errorCode?: string; // on attempt_failed / exhausted
  error?: string;
  // Set on attempt_started (every runtime except DBOS):
  strategy?: 'forced_tool' | 'native_output';
  capabilityProfile?: string; // e.g. 'legacy', 'anthropic-5.5'
}

strategy is the completion strategy the attempt used and capabilityProfile the profile it was selected from. Older readers ignore both fields, and DBOS does not set them yet. The chunk is a hidden diagnostic: @helix-agents/ai-sdk filters it out of the UI stream, so useChat never sees it. During a native_output attempt no text_delta chunks are written (the answer arrives as the output chunk and the persisted completion call); thinking chunks still stream.

typescript

for await (const chunk of stream) {
  if (chunk.type === 'forced_completion' && chunk.phase === 'attempt_started') {
    metrics.increment('forced_completion.attempt', { strategy: chunk.strategy ?? 'forced_tool' });
  }
}

Interrupt/Resume Chunks

These chunks are emitted during agent interrupt and resume operations.

Session vs Run Identifiers

Stream chunks include both sessionId (via agentId) and runId fields:

sessionId (shown as agentId): Primary key for all state operations. A session contains all messages, custom state, and checkpoints for a conversation.
runId: Identifies a specific execution instance within a session. Multiple runs can occur within a single session (e.g., after interrupts or when continuing a conversation).

Use sessionId for state operations (loading messages, saving checkpoints). Use runId for tracking specific executions (logging, tracing, debugging).

tool_approval_request

Emitted when a requireApproval-gated tool is reached. The run durably suspends until the frontend submits a decision via submitToolResult({ kind: 'approval-response', approvalId, approved }).

typescript

interface ToolApprovalRequestChunk {
  type: 'tool_approval_request';
  agentId: string;
  approvalId: string; // Use this in the submit call
  toolCallId: string;
  toolName: string;
  input: unknown;
  reason?: string; // Optional rationale from a function-form requireApproval
}

tool_approval_response

Emitted when an approval decision lands. Consumers see this immediately before the tool actually executes (on approve) or before the synthesized deny message is appended (on reject).

typescript

interface ToolApprovalResponseChunk {
  type: 'tool_approval_response';
  agentId: string;
  approvalId: string;
  toolCallId: string;
  approved: boolean;
  reason?: string; // Optional human-readable rationale
}

On the AI SDK UI stream (@helix-agents/ai-sdk) this chunk produces no tool-approval-response event (ai v6 has no such chunk type); a denial surfaces as a tool-output-available carrying the deny string. See the UI-chunk wire contract.

run_interrupted

Emitted when an agent is interrupted by user request.

typescript

interface RunInterruptedChunk {
  type: 'run_interrupted';
  agentId: string;
  agentType: string;
  timestamp: number;
  runId: string; // Run ID that was interrupted
  checkpointId: string | null; // Checkpoint created (null if failed)
  reason?: string; // Reason for interruption
}

Usage:

typescript

for await (const chunk of stream) {
  if (chunk.type === 'run_interrupted') {
    console.log(`Agent interrupted: ${chunk.reason}`);
    console.log(`Resume from checkpoint: ${chunk.checkpointId}`);
  }
}

run_resumed

Emitted when an interrupted or paused agent resumes execution.

typescript

interface RunResumedChunk {
  type: 'run_resumed';
  agentId: string;
  agentType: string;
  timestamp: number;
  runId: string; // Run ID that was resumed
  fromCheckpointId: string | null; // Checkpoint resumed from
  fromStepCount: number; // Step count at resume point
  mode: string; // Resume mode used
}

Usage:

typescript

for await (const chunk of stream) {
  if (chunk.type === 'run_resumed') {
    console.log(`Resumed from step ${chunk.fromStepCount}`);
    console.log(`Mode: ${chunk.mode}`);
  }
}

run_paused

Emitted when an agent pauses, typically waiting for user confirmation.

typescript

interface RunPausedChunk {
  type: 'run_paused';
  agentId: string;
  agentType: string;
  timestamp: number;
  runId: string;
  reason: string; // Why the agent paused
  pendingToolName?: string; // Tool waiting for confirmation
  pendingToolCallId?: string; // Tool call ID
}

Usage:

typescript

for await (const chunk of stream) {
  if (chunk.type === 'run_paused') {
    console.log(`Agent paused: ${chunk.reason}`);
    if (chunk.pendingToolName) {
      console.log(`Waiting for approval: ${chunk.pendingToolName}`);
    }
  }
}

Every runtime writes it when a run suspends: the stream manager's pauseStream appends the run_paused chunk and then marks the stream paused. That chunk has reason: 'hitl_suspend', step: 0, no pendingToolName, and the stream's agent (session) id in runId, not the suspending run's id. A run pauses its stream only after its suspension commit, so a paused stream means the suspension is committed and a snapshot taken then already holds the suspended step. The run's next continuation makes the stream live again. (RM-56 recorded DBOS, Temporal and Cloudflare Workflows writing no run_paused; this release pauses the stream on all of them. The catalogue record is fixed, proven by conformance cells that include stream.suspension-marker on the Cloudflare Workflows lane. See the conformance findings.)

checkpoint_created

Defined in the stream schema, but not written by any runtime today. Use stateStore.getLatestCheckpoint(sessionId) / listCheckpoints(sessionId) to observe checkpoints.

typescript

interface CheckpointCreatedChunk {
  type: 'checkpoint_created';
  agentId: string;
  agentType: string;
  timestamp: number;
  runId: string;
  checkpointId: string; // The new checkpoint ID
  stepCount: number; // Step when checkpoint was created
}

Run and step markers

These chunks mark run and step boundaries. Clients use them to attribute content to a run and to tell a committed step from a discarded or retried one. The full shapes and the per-runtime emission table are in the Stream Protocol.

Chunk	Written
`run_started`	`{ runId }`. Once per run, by the run that won its run-start commit, after it claimed the stream and made it live and before any content. A run that loses its run start writes nothing at all. The AI SDK transformer drops it.
`step_start`	The first chunk of every LLM attempt, written by the runtime itself before the model call (every runtime).
`step_end`	When the LLM adapter reports the step's end, with `usage` and `finishReason`. `@helix-agents/llm-vercel` does not report it today (RM-48).
`step_committed`	`{ runId, stepId, checkpointId, committedAttemptId }` after the step's commit. `committedAttemptId` names the attempt whose chunks are final; it is not forwarded to the UI.
`step_discarded`	`{ runId, stepId, reason }` for a step that streamed but will never commit.
`run_terminated`	`{ runId, reason: 'owner_unknown_timeout' \| 'admin_terminate' }`. The framework force-ended a run, on Cloudflare Workflows only: `owner_unknown_timeout` is written by the entry that kills an instance whose status stayed unknown past the bound, before its own `run_started`; `admin_terminate` by `executor.terminateRun()`. Forwarded as `data-run-terminated` `{ runId, reason }`.

Retried attempts. A Temporal activity retry or a Cloudflare Workflows step.do retry streams the same step again under a new attemptId. For a committed step the final attempt is the one committedAttemptId names; before the commit it is the attempt whose step_start came last in the stream. Earlier attempts' content is superseded: the AI SDK chat handler drops it from what it replays and emits a data-attempt-superseded resync trigger when a client already saw it. See Retried attempts.

step_committed

Emitted after a step's commit (the step's messages and checkpoint written atomically).

typescript

interface StepCommittedChunk {
  type: 'step_committed';
  agentId: string;
  agentType: string;
  timestamp: number;
  step: number;
  runId: string;
  stepId: string; // Step that was committed
  checkpointId: string; // Checkpoint the commit wrote
  committedAttemptId: string; // The attempt whose chunks are final ('' when none)
}

On the AI SDK UI stream this is data-step-committed { runId, stepId, checkpointId }.

step_discarded

Emitted when a step that streamed content will never commit (on interrupt, abort or a dead owner).

typescript

interface StepDiscardedChunk {
  type: 'step_discarded';
  agentId: string;
  agentType: string;
  timestamp: number;
  step: number;
  runId: string;
  stepId: string;
  reason: string; // Why the step was discarded
}

Not every runtime writes step_discarded (RM-59)

Cloudflare Workflows and DBOS write it for an interrupted, aborted or abandoned in-flight step. JS and Temporal interrupts finish and commit the step first, so they have nothing to discard. Temporal's abort() leaves the in-flight step's chunks on the stream with no step_discarded (RM-59). See the conformance findings.

stream_resync

Emitted when an entry rewound the log or removed stream chunks, so a client must re-anchor on the snapshot. It is written only by an entry that won its run start, after its entry state and before any content.

Recovery-only — not the happy path

retry() writes stream_resync every time. resume() writes one only when it rewinds (from_checkpoint) or removes a crashed step's orphan chunks. A normal client-tool resume or a follow-up turn truncates nothing and emits no resync. If you wire useAutoResync, expect it to fire rarely.

typescript

interface StreamResyncChunk {
  type: 'stream_resync';
  agentId: string;
  agentType: string;
  timestamp: number;
  step: number;
  runId?: string; // The run that wrote it; a resync is identified by (runId, checkpointId)
  checkpointId: string; // The truncate checkpoint the run-start commit wrote, else the restored one
  stepCount: number; // Step count at the restored checkpoint
  messageCount: number; // Message count after the rewind
  fromSequence: number; // Stream sequence of the rewind
  truncateCheckpointId?: string; // Set when the run-start commit truncated
  reason: 'crash_recovery' | 'rollback' | 'branch' | 'retry' | 'respawn';
}

Reasons:

retry: retry() (every runtime).
rollback: resume({ mode: 'from_checkpoint' }). A rewind writes exactly one resync.
crash_recovery: a resume that removed a crashed step's orphan chunks (runtime-js and the Cloudflare DO, DBOS, Cloudflare Workflows).
respawn: a failed / terminated persistent companion re-spawned by name on the Cloudflare DO, which starts it on a fresh session under the same id (CP-142), or a first spawn there over a session an earlier incarnation of the parent left (CP-205).
branch: reserved, not emitted.

On the AI SDK UI stream this is data-stream-resync, carrying runId and truncateCheckpointId. The React hooks re-anchor on it:

typescript

import { collectResyncTriggers, useAutoResync } from '@helix-agents/ai-sdk/react';

const { resyncError } = useAutoResync(collectResyncTriggers(chat.messages), {
  snapshotUrl: `/api/chat/${sessionId}/snapshot`,
  setMessages: chat.setMessages,
  chat, // stream control: the useHelixChat result
  onResync: (event) => toast.info(`Recovered (${event.data.reason})`),
});

useHelixChat({ resync: { snapshotUrl } }) wires the same thing in one option. See Resync for the re-anchor contract and Recovery hooks for the hooks.

Complete Streaming Example

typescript

async function runWithStreaming() {
  const handle = await executor.execute(ResearchAgent, 'Research quantum computing');
  const stream = await handle.stream();

  if (!stream) {
    console.log('No stream available');
    return;
  }

  // Track state
  let currentTool = '';
  let toolDepth = 0;

  for await (const chunk of stream) {
    switch (chunk.type) {
      case 'text_delta':
        process.stdout.write(chunk.delta);
        break;

      case 'thinking':
        if (chunk.isComplete) {
          console.log('\n[Thinking]', chunk.content);
        }
        break;

      case 'tool_start':
        currentTool = chunk.toolName;
        console.log(`\n[→ ${chunk.toolName}]`, JSON.stringify(chunk.arguments));
        break;

      case 'tool_end':
        if (chunk.error) {
          console.log(`[✗ ${chunk.toolName}]`, chunk.error);
        } else {
          console.log(`[✓ ${chunk.toolName}]`);
        }
        break;

      case 'subagent_start':
        toolDepth++;
        console.log(`${'  '.repeat(toolDepth)}[Sub-agent: ${chunk.subAgentType}]`);
        break;

      case 'subagent_end':
        console.log(`${'  '.repeat(toolDepth)}[/${chunk.subAgentType}]`);
        toolDepth--;
        break;

      case 'custom':
        if (chunk.eventName === 'progress') {
          const { step, total } = chunk.data as { step: number; total: number };
          console.log(`[Progress: ${step}/${total}]`);
        }
        break;

      case 'state_patch':
        // Optionally apply patches to local state
        break;

      case 'error':
        console.error(`\n[Error]`, chunk.error);
        if (!chunk.recoverable) {
          console.error('Execution cannot continue');
        }
        break;

      case 'output':
        console.log('\n[Output]', JSON.stringify(chunk.output, null, 2));
        break;
    }
  }

  // Get final result
  const result = await handle.result();
  console.log('\nFinal status:', result.status);
}

Stream Filtering

The framework provides utilities for filtering streams.

By Chunk Type

typescript

import { filterStreamByType } from '@helix-agents/core';

const stream = await handle.stream();

// Only text deltas
const textStream = filterStreamByType(stream, 'text_delta');
for await (const chunk of textStream) {
  process.stdout.write(chunk.delta);
}

// Only tool events
const toolStream = filterStreamByType(stream, ['tool_start', 'tool_end']);
for await (const chunk of toolStream) {
  console.log(chunk.type, chunk.toolName);
}

By Agent

typescript

import { filterStreamByAgent } from '@helix-agents/core';

const stream = await handle.stream();

// Only chunks from specific agent
const agentStream = filterStreamByAgent(stream, 'run-123');
for await (const chunk of agentStream) {
  console.log(chunk);
}

// Exclude sub-agent chunks
const parentOnly = filterStreamByAgent(stream, handle.sessionId);

Custom Filters

typescript

async function* filterStream<T>(
  stream: AsyncIterable<T>,
  predicate: (chunk: T) => boolean
): AsyncIterable<T> {
  for await (const chunk of stream) {
    if (predicate(chunk)) {
      yield chunk;
    }
  }
}

// Example: only errors and outputs
const importantStream = filterStream(
  stream,
  (chunk) => chunk.type === 'error' || chunk.type === 'output'
);

Stream Control Messages

In addition to data chunks, streams can include control messages:

stream_end

Signals the stream is complete:

typescript

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

This is a control message, not a data chunk. Most consumers don't need to handle it directly - the async iterator completes when the stream ends.

Every turn of a session writes to the same session stream. A run's endStream / failStream / pauseStream is fenced on its own run id: once a newer run has claimed the stream, a late end from the previous run is a no-op, so it cannot close the stream under the next turn. A superseded run's late chunks are dropped the same way.

Frontend Integration

For web applications, streams are typically delivered via SSE:

typescript

// Backend
app.get('/agent/stream/:sessionId', async (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');

  const stream = await getStream(req.params.sessionId);

  for await (const chunk of stream) {
    res.write(`data: ${JSON.stringify(chunk)}\n\n`);
  }

  res.end();
});

// Frontend
const eventSource = new EventSource(`/agent/stream/${sessionId}`);

eventSource.onmessage = (event) => {
  const chunk = JSON.parse(event.data);
  handleChunk(chunk);
};

See Frontend Integration for complete patterns.

Resumable Streams

Every stored chunk has a sequence number, so a reader can start from a position:

typescript

// Read from a sequence (inclusive): skip chunks the client already has.
const stream = await handle.stream({ fromSequence: lastSeenSequence + 1 });

for await (const chunk of stream) {
  // Chunks read back from the stream manager carry their position as `_sequence`.
  remember(chunk._sequence);
}

Pass signal to stop reading when the consumer goes away, for example the HTTP client's disconnect: handle.stream({ fromSequence, signal: request.signal }). When it aborts, the iteration ends and the reader is closed at once, even while it waits for the next chunk. The run itself is not affected. Every runtime honours it, and handleChatStream passes its client's signal for you (see A client that disconnects).

For a chat UI that survives a page refresh, use a snapshot plus the stream after its cursor. The full contract is in the Resumable Streams guide.

Snapshot-Based Resumption (Recommended)

buildSnapshot() reads one checkpoint: the committed messages up to its cursor (streamSequence), the state at that cursor, and a status the current run decides. handleChatStream serves the resume from that cursor.

typescript

import { handleChatStream, buildSnapshot, type HandleChatStreamParams } from '@helix-agents/ai-sdk';

const deps = { executor, stateStore, streamManager, agent: myAgent } as const;

// Every chat POST (send) and GET (resume) goes through the orchestrator.
const dispatchChat = (params: HandleChatStreamParams) => handleChatStream(deps, params);

// The snapshot: messages, state, streamSequence (the cursor), status,
// checkpointId, stepCount, resumeMessageId (when live), latestTruncateCheckpointId.
const snapshot = await buildSnapshot(deps, { sessionId });

On the client, snapshotResumeOptions(snapshot) says what to seed and how to resume:

typescript

import { snapshotResumeOptions } from '@helix-agents/ai-sdk/client';

const resumeOptions = snapshotResumeOptions(snapshot);
// active / paused: { resume: true, resumeFromSequence, existingMessageId, replayPrefix: true,
//                    messages: the snapshot WITHOUT the in-flight message }
// ended / failed:  { resume: false, messages: snapshot.messages }

Wire it into DefaultChatTransport with prepareHelixReconnectRequest and seed useHelixChat with resumeOptions.messages; see The canonical client.

Mid-Stream Page Refresh

A snapshot never carries uncommitted content. When the user refreshes while a run is streaming:

The snapshot holds every committed message up to its cursor. Its status is active (or paused) because the run is live, even in the short window after the run's start commit and before its stream goes live.
The client seeds its chat without the in-flight message (resumeMessageId) and reconnects with X-Resume-From-Sequence, X-Existing-Message-Id and X-Helix-Replay-Prefix: 1.
handleChatStream reads one snapshot, replays that message's committed prefix under the same id (between transient data-helix-replay markers), then streams the tail after that snapshot's cursor.

Each committed content chunk is rendered exactly once, on every supported ai version: ai 6.0.281 starts a resumed response from an empty message that replaces the last one, so a client that kept the message would lose its committed steps there. Replayed tool calls are never executed again. See Why the resume message is removed.

The id the run's response streams into is generateId(run.startUIMessageCount) (msg-<n> by default): the first-row index of the UI message that holds it. The snapshot's resumeMessageId, the live stream's start event and the committed message all carry it. Ids are not contiguous; see UI Messages.

Breaking: content replay and partial content are removed

The content-replay and partial-content options and the partial-content merge are gone, and so are the snapshot's timestamp, startSequence and startMessageCount. See the upgrade guide.

Stream Status

The snapshot's status tells clients whether to resume. It comes from the current run, not the stream alone:

Status	Meaning	Action
`active`	The current run is live and running	Resume from `streamSequence`
`paused`	The current run is live and suspended (HITL)	Resume from `streamSequence`
`ended`	No live run; the last one did not fail	No SSE request
`failed`	No live run; the current run or stream failed	Render the error state

SSE Event IDs

When using the AI SDK integration, each SSE event carries its chunk's sequence as id::

id: 42
data: {"type":"text-delta","id":"block-1","delta":"Hello"}

id: 43
data: {"type":"text-delta","id":"block-1","delta":" world"}

Replayed prefix events carry no id:. A reconnect may send Last-Event-ID, which the handler accepts when X-Resume-From-Sequence is absent.

For complete patterns including SSR with Next.js, see:

Resumable Streams - The snapshot, resume and resync contract
AI SDK Integration - Frontend integration with useChat
Resumable Streams Example - Full Next.js implementation with crash recovery

Streaming Metadata

When using the AI SDK integration (@helix-agents/ai-sdk), you can attach custom metadata to stream start and finish events. This is useful for request tracing, timing metrics, model attribution, and passing context to clients.

Configuration

Configure metadata via transformerOptions on the deps bundle. handleChatStream (the v7 chat orchestrator) forwards it to the stream transformer; buildSnapshot accepts the same bundle and ignores it:

typescript

import { handleChatStream, buildSnapshot, type HandleChatStreamParams } from '@helix-agents/ai-sdk';

const transformerOptions = {
  // Static metadata - same for all streams
  startMetadata: { environment: 'production', version: '1.0' },
  finishMetadata: { service: 'chat-api' },

  // Or dynamic metadata - computed per stream
  startMetadata: (agentId) => ({
    agentId,
    requestId: generateRequestId(),
    startedAt: Date.now(),
  }),
  finishMetadata: (agentId) => ({
    agentId,
    finishedAt: Date.now(),
  }),
};

const deps = {
  executor,
  stateStore,
  streamManager,
  agent: myAgent,
  transformerOptions,
} as const;

const dispatchChat = (params: HandleChatStreamParams) => handleChatStream(deps, params);
const getSnapshot = (sessionId: string) => buildSnapshot(deps, { sessionId });

Stream Events

Metadata appears in the messageMetadata field of SSE start and finish events:

data: {"type":"start","messageId":"msg-123","messageMetadata":{"requestId":"req-001","startedAt":1234567890}}

data: {"type":"finish","messageId":"msg-123","finishReason":"stop","messageMetadata":{"model":"claude-sonnet-5-5","totalTokens":150}}

Use Cases

Use Case	startMetadata	finishMetadata
Request tracing	`{ requestId, traceId }`	`{ requestId }`
Timing metrics	`{ startedAt }`	`{ duration, finishedAt }`
Model attribution	-	`{ model, provider }`
Token counting	-	`{ inputTokens, outputTokens }`
User context	`{ userId, sessionId }`	-
Feature flags	`{ features: [...] }`	-

User Message Metadata

When using the AI SDK, you can also attach metadata to user messages:

typescript

// Frontend: include metadata in request body
const { messages } = useChat({
  api: '/api/chat',
  body: {
    metadata: {
      source: 'web-ui',
      userId: currentUser.id,
      sessionId: sessionId,
    },
  },
});

This metadata is persisted with the user message and available when loading message history via getUIMessages().

See the AI SDK documentation for complete frontend integration patterns.

Best Practices

1. Handle All Chunk Types

Defensive programming for unknown chunk types:

typescript

for await (const chunk of stream) {
  switch (chunk.type) {
    case 'text_delta':
    case 'tool_start':
    case 'tool_end':
      // Handle known types
      break;
    default:
      // Log unknown types for debugging
      console.log('Unknown chunk type:', chunk.type);
  }
}

2. Buffer Text Deltas

Don't update UI on every delta - batch them:

typescript

let buffer = '';
let timeout: NodeJS.Timeout;

for await (const chunk of stream) {
  if (chunk.type === 'text_delta') {
    buffer += chunk.delta;

    // Debounce UI updates
    clearTimeout(timeout);
    timeout = setTimeout(() => {
      updateUI(buffer);
      buffer = '';
    }, 50);
  }
}

// Flush remaining
if (buffer) updateUI(buffer);

3. Track Tool Call IDs

Match tool_start with tool_end:

typescript

const pendingTools = new Map<string, ToolStartChunk>();

for await (const chunk of stream) {
  if (chunk.type === 'tool_start') {
    pendingTools.set(chunk.toolCallId, chunk);
  }
  if (chunk.type === 'tool_end') {
    const start = pendingTools.get(chunk.toolCallId);
    if (start) {
      const duration = chunk.timestamp - start.timestamp;
      console.log(`${chunk.toolName} took ${duration}ms`);
      pendingTools.delete(chunk.toolCallId);
    }
  }
}

4. Handle Errors Gracefully

Distinguish recoverable from fatal errors:

typescript

for await (const chunk of stream) {
  if (chunk.type === 'error') {
    if (chunk.recoverable) {
      // Show warning, continue streaming
      showWarning(chunk.error);
    } else {
      // Show error, stop streaming
      showError(chunk.error);
      break;
    }
  }
}

Next Steps

Frontend Integration - Display streams in web UIs
Sub-Agents - Understanding nested streams
State Management - Using state_patch chunks
Interrupt and Resume - Handle interrupt/resume stream events
Hooks - Observe agent execution with lifecycle and tool hooks

Real-Time Streaming ​

Overview ​

Stream Chunk Types ​

text_delta ​

thinking ​

tool_start ​

tool_end ​

subagent_start ​

subagent_end ​

Sub-agent chunk ordering ​

A child still running after the parent's turn ends ​

Failure-path semantics ​

Diagnosing a stuck 'input-available' in production ​

custom ​

state_patch ​

error ​

output ​

forced_completion ​

Interrupt/Resume Chunks ​

tool_approval_request ​

tool_approval_response ​

run_interrupted ​

run_resumed ​

run_paused ​

checkpoint_created ​

Run and step markers ​

step_committed ​

step_discarded ​

stream_resync ​

Complete Streaming Example ​

Stream Filtering ​

By Chunk Type ​

By Agent ​

Custom Filters ​

Stream Control Messages ​

stream_end ​

Frontend Integration ​

Resumable Streams ​

Snapshot-Based Resumption (Recommended) ​

Mid-Stream Page Refresh ​

Stream Status ​

SSE Event IDs ​

Streaming Metadata ​

Configuration ​

Stream Events ​

Use Cases ​

User Message Metadata ​

Best Practices ​

1. Handle All Chunk Types ​

2. Buffer Text Deltas ​

3. Track Tool Call IDs ​

4. Handle Errors Gracefully ​

Next Steps ​

Real-Time Streaming

Overview

Stream Chunk Types

text_delta

thinking

tool_start

tool_end

subagent_start

subagent_end

Sub-agent chunk ordering

A child still running after the parent's turn ends

Failure-path semantics

Diagnosing a stuck `'input-available'` in production

custom

state_patch

error

output

forced_completion

Interrupt/Resume Chunks

tool_approval_request

tool_approval_response

run_interrupted

run_resumed

run_paused

checkpoint_created

Run and step markers

step_committed

step_discarded

stream_resync

Complete Streaming Example

Stream Filtering

By Chunk Type

By Agent

Custom Filters

Stream Control Messages

stream_end

Frontend Integration

Resumable Streams

Snapshot-Based Resumption (Recommended)

Mid-Stream Page Refresh

Stream Status

SSE Event IDs

Streaming Metadata

Configuration

Stream Events

Use Cases

User Message Metadata

Best Practices

1. Handle All Chunk Types

2. Buffer Text Deltas

3. Track Tool Call IDs

4. Handle Errors Gracefully

Next Steps