7 Streaming, State, and Termination
Part II · The portable model runtime
Streaming sources checked: 2026-10-07
Streaming reduces time to first output, but it turns one response into a state machine whose transitions cross transport, provider, adapter, application, and consumer boundaries. The consumer must preserve order, correlate fragments, propagate cancellation, bound memory, and distinguish a useful partial result from a completed answer.
A correct stream consumer produces the same logical terminal state as the corresponding non-streaming response, or reports why it cannot. Rendering early is optional; preserving lifecycle and outcome is not.
7.1 A stream has three layers
Do not conflate these layers:
- Transport bytes arrive in network chunks.
- Protocol frames delimit messages, such as server-sent event data records.
- Semantic events describe text deltas, item changes, tool calls, usage, errors, and terminal state.
Byte boundaries are arbitrary. UTF-8 code points, lines, and JSON values can cross them. A transport parser must buffer incomplete framing before JSON decoding. The semantic reducer from Chapter 6 should never inspect raw chunks.
The Open Responses HTTP stream uses server-sent events carrying JSON event objects and a terminal [DONE] marker (“Open Responses Specification 2026-04-24,” n.d.). That marker closes framing; the semantic terminal event determines whether the response completed, failed, or remained incomplete.
7.2 Stream all meanings, not only text
A text-only stream is tempting. This is an API-usage fragment, not a complete consumer:
for await (const delta of result.textStream) {
render(delta);
}It is insufficient for an agent loop because it can hide tool calls, tool errors, usage, finish reasons, and abort events. The AI SDK distinguishes a text stream from its complete event stream; errors and nontext parts require the richer stream or callbacks (“AI SDK Core,” n.d.).
At the application boundary, use a discriminated union. This is an architecture sketch, not the exact v0.4.0 event type:
type ModelEvent =
| { type: "response-start"; responseId?: string }
| { type: "text-delta"; delta: string }
| { type: "tool-call"; callId: string; name: string; input: unknown }
| { type: "tool-result"; callId: string; name: string; output: unknown }
| { type: "usage"; inputTokens?: number; outputTokens?: number }
| { type: "finish"; reason: FinishReason }
| { type: "error"; error: PublicModelError };This proposed union is deliberately smaller than Open Responses. The exact v0.4.0 model event union exposes text deltas, tool calls, correlated tool-result outcomes, and finish events carrying usage and an optional response identifier; failures are thrown as sanitized errors. It does not expose separate response-start, usage, or error events. The sketch separates those meanings for discussion, not as a claim about the release API or protocol conformance.
7.3 Ordered reduction
Process semantic events serially unless the protocol explicitly permits independent partitions. A simple control-flow sketch provides an important invariant: event n finishes reduction before event n + 1 mutates state.
for await (const event of adapter.stream(request)) {
state = applyModelEvent(state, event);
await publishProjection(state, event);
}publishProjection may update a UI or send a downstream event, but it should not mutate canonical state. If publishing is slow, decide whether to block upstream consumption, buffer within limits, or decouple through a bounded queue.
Parallelizing arbitrary event handlers creates races such as a tool-result being rendered before its tool-call, or a terminal snapshot being stored before the final text delta.
7.4 Backpressure is an end-to-end policy
The Web Streams model includes backpressure: a producer can observe downstream demand rather than pushing without limit (“Streams Standard,” n.d.). But application pipelines often break that chain by reading as fast as possible into an unbounded array or message queue.
Choose explicit limits for:
- unread transport bytes;
- decoded but unapplied events;
- accumulated text and tool arguments;
- pending UI or WebSocket messages;
- concurrent tool calls; and
- persisted event volume per run.
When a limit is reached, the options are to pause, shed nonessential projections, cancel, or fail. “Keep buffering” is not a safe default.
The chapter’s executable bounded consumer blocks its producer at a configured event capacity, reduces events serially, awaits a deliberately slow projection, and keeps cancellation observable while either side waits. The complete source is code/foundations.ts; tests assert ordered delivery, a peak queue size no greater than the configured bound, and cancellation while waiting for capacity.
export const consumeBounded = async <Event, State>(options: {
source: AsyncIterable<Event>;
maximumBufferedEvents: number;
initialState: State;
reduce: (state: State, event: Event) => State;
publish: (state: State, event: Event) => void | Promise<void>;
signal?: AbortSignal;
}): Promise<{ state: State; peakBuffered: number }> => {
invariant(options.maximumBufferedEvents > 0, "buffer limit must be positive");
const queue: Event[] = [];
let peakBuffered = 0;
let done = false;
let failure: unknown;
let wakeConsumer: (() => void) | undefined;
let wakeProducer: (() => void) | undefined;
const wait = (role: "consumer" | "producer"): Promise<void> => {
if (options.signal?.aborted) return Promise.reject(options.signal.reason);
return new Promise((resolve, reject) => {
const abort = (): void => reject(options.signal?.reason);
const wake = (): void => {
options.signal?.removeEventListener("abort", abort);
resolve();
};
options.signal?.addEventListener("abort", abort, { once: true });
if (role === "consumer") wakeConsumer = wake;
else wakeProducer = wake;
});
};
const producer = (async (): Promise<void> => {
try {
for await (const event of options.source) {
while (queue.length >= options.maximumBufferedEvents) await wait("producer");
queue.push(event);
peakBuffered = Math.max(peakBuffered, queue.length);
wakeConsumer?.();
wakeConsumer = undefined;
}
} catch (error) {
failure = error;
} finally {
done = true;
wakeConsumer?.();
}
})();
let state = options.initialState;
while (!done || queue.length > 0) {
if (options.signal?.aborted) throw options.signal.reason;
if (queue.length === 0) {
await wait("consumer");
continue;
}
const event = queue.shift() as Event;
wakeProducer?.();
wakeProducer = undefined;
state = options.reduce(state, event);
await options.publish(state, event);
}
await producer;
if (failure !== undefined) throw failure;
return { state, peakBuffered };
};A client that reads slowly can turn a healthy model request into memory pressure. Test a deliberately slow sink and prove that buffering stays bounded or cancellation propagates.
7.5 Cancellation has a direction
Cancellation begins with a principal: a disconnected client, a user pressing stop, a deadline, a budget guard, or an operator. Its direction is explicit:
Client Agent run Model Tool
│ stop/disconnect │ │ │
├────────────────────►│ │ │
│ ├── abort signal ───►│ │
│ ├── abort signal ────────────────────────►│
│ │◄── cancelled / completed / unknown ─────┤
│◄── terminal outcome ┤ │ │Propagate one signal through every cancellable operation. This SDK call fragment omits surrounding construction:
const result = streamText({
model,
messages,
abortSignal: request.signal,
tools,
});The AI SDK accepts an AbortSignal for the overall operation and provides abort lifecycle handling (“AI SDK Core,” n.d.). Passing the signal only to the model is incomplete. Tool execution, sleeps, retrieval calls, subprocesses, and queue waits must also observe it.
Cancellation does not automatically roll back side effects. If a tool already created a record, stopping the model cannot uncreate it. Tools need idempotency keys, transactions, compensating actions, or a clear “outcome unknown” state.
7.5.1 Cancellation is not failure
Normalize cancellation separately from provider failure. This supports accurate telemetry and retry policy:
- a user cancellation should usually not page an operator;
- a deadline may indicate capacity or policy configuration;
- a provider timeout may be retryable;
- a tool outcome after disconnect may require reconciliation.
Cleanup can need an independent, short timeout. Reusing an already-aborted signal for mandatory cleanup guarantees that cleanup never starts.
7.6 Partial output is provisional
Streaming output may be useful before completion, but it is not yet an authoritative final answer. The following proposed projection type labels partial state explicitly:
type Projection = {
text: string;
provisional: boolean;
outcome: "running" | "completed" | "incomplete" | "failed" | "cancelled";
};A UI can display provisional text while preventing downstream automation from treating it as final. On failure or cancellation, policy determines whether to retain, hide, or annotate it. The choice should differ by use case: partial creative text may be useful; a partial JSON object or transaction instruction may be unsafe.
Never parse a structured result from an arbitrary text prefix and execute it. Wait for the structured item or terminal validation.
7.7 Snapshot consistency
The reducer should be able to create a snapshot after any accepted prefix, but only a terminal event can mark it final. At termination, verify:
- all referenced items and calls exist;
- no text or argument buffer remains syntactically unfinished;
- terminal status is unique;
- finish reason is normalized without discarding protected raw diagnostics;
- input/output usage counts preserve the distinction between unknown and measured zero, as the chapter reducer and
v0.4.0common loop do; and - any authoritative final object agrees with accumulated output.
For a completed response, a mismatch between accumulated deltas and final text should fail a consistency check. For an interrupted transport with no semantic terminal event, report an indeterminate or failed outcome rather than manufacturing completed.
7.8 Disconnect is a stream-local signal
Client disconnect, model cancellation, and durable resumption are separate features. This chapter owns the first two only: the handler must define whether reader cancellation/request abort propagates to active model and tool work, release stream-local buffers, and report the most accurate terminal knowledge available to code that can still receive it.
A cancelled connection cannot receive a terminal event on that same connection. It also cannot prove that a remote effect did not occur. Durable run identity, retained event replay, reconnect authorization, unknown-effect reconciliation, and restart policy belong to Chapter 15. Until those contracts exist, advertise a new attempt rather than resumability.
7.9 Errors before and during streaming
Errors can occur:
- before headers, when a normal HTTP error is still possible;
- after streaming begins, when the protocol needs an error event;
- inside tool execution, where a correlated tool error may be model-visible; or
- after a side effect, where the operational outcome is uncertain.
Public errors should carry a stable category, safe message, retry hint, and correlation identifier. Raw provider bodies, headers, prompts, and tool arguments belong only in protected diagnostics under a redaction policy.
The v0.4.0 reference loop disables accidental SDK error logging and maps provider failures into a sanitized AdapterError. This prevents raw provider details from reaching callers or default logs. That release does not provide a protected diagnostic store, so correlation and authorized diagnostic capture remain unimplemented operational requirements.
7.10 Retry is outside the stream reducer
A missing terminal event does not authorize another attempt. The stream layer reports what became visible, which calls started, the last accepted sequence, and why terminal meaning is absent. Chapter 15 decides whether that history permits replacement, requires reconciliation, or must stop. Keeping retry policy outside the reducer prevents a transport interruption from silently duplicating tool effects or user-visible text.
7.11 Deterministic streaming tests
A network-free suite should vary chunking independently from semantic events. Feed the same event sequence as:
- one transport chunk;
- one byte at a time;
- boundaries inside UTF-8 and JSON tokens;
- several events in one chunk; and
- a final truncated frame.
Then vary semantic behavior:
- normal text completion;
- tool call and result;
- duplicate or missing event;
- abort between deltas;
- slow downstream consumer;
- error after partial text; and
- EOF without a terminal event.
The immutable v0.4.0 reference uses AI SDK mock language models for common-loop tests and separate fixtures for its bounded consumer and deterministic HTTP/SSE endpoint. Both the reference and chapter bounded-consumer tests establish local slow-sink queue bounds and queue-wait cancellation, not network-byte limits. The tag retains a basic Pi 1.0.4 text-streaming acceptance record at tested commit 0d54a3e; extended tool, session, usage-display, and cancellation acceptance belongs to post-release work. Neither client acceptance nor complete-response SSE checks replace adversarial transport-chunking tests.
7.11.1 What the later live bridge actually proves
At the current reviewed snapshot, cancelling the response body’s reader aborts the signal supplied to a scripted provider. The handler also forwards an incoming request abort and removes its listener when work ends. This is deterministic handler-level cancellation evidence—not proof that every network disconnect, provider, or side effect stops correctly. A cancelled reader cannot receive a terminal event on that cancelled stream; cleanup and reconciliation must not depend on delivering one.
The live bridge’s producer eagerly enqueues SSE records from start() without waiting for downstream demand. It does not use the bounded consumer. Its output-token ceiling limits requested generation, not unread transport bytes, event count, or queue size. Collecting text chunks and joining once avoids repeated concatenation, but neither that optimization nor immediate forwarding establishes bounded memory or end-to-end backpressure. Slow-reader HTTP tests and explicit queue admission remain necessary.
The lesson is to keep proofs at their tested boundary: local queue bounds, response-reader cancellation, client compatibility, and network backpressure are four different claims.
7.12 Portability and evidence ledger
| Dimension | Chapter claim | Evidence | Not established |
|---|---|---|---|
| Transport | Bytes, framing, and semantic events are separate layers | Tagged deterministic HTTP/SSE checks and basic Pi text acceptance | Full adversarial byte-framing coverage |
| Provider | Common semantic consumption need not branch on provider event names | v0.4.0 injected models and retained dated text-generation checks |
All provider events or model configurations |
| Protocol | Explicit terminal meaning matters beyond transport EOF | Open Responses target and chapter reducer | Conformance or resumable protocol |
| Feature | Bounded buffering and cancellation are implementable application policies | Book and v0.4.0 local slow-sink and queue-wait abort tests |
End-to-end network backpressure |
| Semantic | Partial output is provisional | State model only | Equivalent final answers |
7.13 Pattern summary
Problem: Streaming improves responsiveness but can lose order, lifecycle, cancellation, and terminal truth.
Pattern: Separate transport parsing from semantic reduction, consume the rich event stream serially, propagate cancellation, bound buffering, and reconcile a terminal snapshot.
Result: Early output remains responsive without becoming indistinguishable from completed state.
Cost: The implementation needs explicit state, limits, cleanup policy, and adversarial stream tests.
7.14 Exercises
- Define a maximum buffer policy for a slow WebSocket client.
- Decide whether your server cancels model work when an HTTP client disconnects.
- Write a test where EOF occurs after text but before a terminal event.
- Identify a tool side effect that requires reconciliation after cancellation.
- Explain which events your UI may drop under pressure and which it must preserve.
7.15 Further reading
Compare the Open Responses streaming and semantic-event rules (“Open Responses Specification 2026-04-24,” n.d.) with the Web Streams backpressure model (“Streams Standard,” n.d.). For the reference implementation, consult the pinned AI SDK streamText contract for complete-stream, error, usage, finish, and cancellation behavior (“AI SDK Core,” n.d.).
7.16 Chapter contract
- Scope: ordered stream reduction, bounded consumption, cancellation, and termination.
- Canonical claims: a stream is a state transition sequence, not a text iterator.
- Migration inputs: former streaming chapter plus termination behavior.
- Evidence gates: ordering, limits, malformed events, cancellation, and terminal-state tests.
- Demonstration slots: deterministic scripted stream.
- Source-claim ledger: local reducer/consumer evidence is tied to
v0.4.0and book tests; handler cancellation evidence is tied to snapshot45a903c; neither is end-to-end network proof. - Exclusions: request/item vocabulary belongs to Chapter 6, tool authorization to Chapter 8, retries and durable cross-request state to Chapter 15, and OpenTelemetry to Chapter 16.