oh-my-pi/docs/provider-streaming-internals.md

Provider streaming internals

This document explains how token/tool streaming is normalized in @oh-my-pi/pi-ai, then propagated through @oh-my-pi/pi-agent-core and coding-agent session events.

End-to-end flow

1. streamSimple() (packages/ai/src/stream.ts) maps generic options and dispatches to a provider stream function.
2. Provider stream functions translate provider-native stream events into the unified AssistantMessageEvent sequence. Current built-ins include Anthropic, OpenAI Responses/Completions/Codex/Azure Responses, Google Gemini/Gemini CLI/Vertex, Bedrock Converse, Ollama, Cursor, pi-native gateway transport, plus GitLab Duo/Kimi/Synthetic wrappers and extension-registered custom APIs.
3. Each provider pushes events into AssistantMessageEventStream (packages/ai/src/utils/event-stream.ts), which throttles delta events and exposes:
   • async iteration for incremental updates
   • result() for final AssistantMessage
4. agentLoop (packages/agent/src/agent-loop.ts) consumes those events, mutates in-flight assistant state, and emits message_update events carrying the raw assistantMessageEvent.
5. AgentSession (packages/coding-agent/src/session/agent-session.ts) subscribes to agent events, persists messages, drives extension hooks, and applies session behaviors (retry, compaction, TTSR, streaming-edit abort checks).

Unified stream contract in @oh-my-pi/pi-ai

All providers emit the same shape (AssistantMessageEvent in packages/ai/src/types.ts):

• start
• content block lifecycle triplets:
  • text: text_start → text_delta* → text_end
  • thinking: thinking_start → thinking_delta* → thinking_end
  • tool call: toolcall_start → toolcall_delta* → toolcall_end
• terminal event:
  • done with reason: "stop" | "length" | "toolUse"
  • or error with reason: "aborted" | "error"

AssistantMessageEventStream guarantees:

• final result is resolved by terminal event (done or error)
• deltas are batched/throttled (~50ms)
• buffered deltas are flushed before non-delta events and before completion

Delta throttling and harmonization behavior

AssistantMessageEventStream treats text_delta, thinking_delta, and toolcall_delta as mergeable events:

• buffered deltas are merged only when type + contentIndex match
• merge keeps the latest partial snapshot
• non-delta events force immediate flush

This smooths high-frequency provider streams for TUI/event consumers, but is not provider backpressure: providers still produce at full speed, while the local stream buffers.

Provider normalization details

Anthropic (anthropic-messages)

Source: packages/ai/src/providers/anthropic.ts

Normalization points:

• message_start initializes usage (input/output/cache tokens)
• content_block_start maps to text/thinking/toolcall starts
• content_block_delta maps:
  • text_delta → text_delta
  • thinking_delta → thinking_delta
  • input_json_delta → toolcall_delta
  • signature_delta updates thinkingSignature only (no event)
• content_block_stop emits corresponding *_end
• message_delta.stop_reason maps via mapStopReason()

Tool-call argument streaming:

• each tool block carries internal partialJson
• every JSON delta appends to partialJson
• arguments are reparsed on each delta via parseStreamingJson()
• toolcall_end reparses once more, then strips partialJson

OpenAI Responses family (openai-responses, openai-codex-responses, azure-openai-responses)

Sources: packages/ai/src/providers/openai-responses.ts, openai-codex-responses.ts, and azure-openai-responses.ts

Normalization points:

• response.output_item.added starts reasoning/text/function-call/custom-tool blocks
• reasoning summary events (response.reasoning_summary_text.delta) and raw reasoning events (response.reasoning_text.delta) become thinking_delta
• output/refusal deltas become text_delta
• response.function_call_arguments.delta and response.custom_tool_call_input.delta become toolcall_delta
• response.output_item.done emits thinking_end / text_end / toolcall_end
• response.completed maps status to stop reason and usage; response.failed / SDK error events throw into the wrapper's terminal error path

Tool-call argument streaming:

• same partialJson accumulation pattern as Anthropic for function-call JSON arguments
• custom tools stream raw string input and expose final arguments as { input: <raw> }
• providers that send only response.function_call_arguments.done still populate final args
• tool call IDs are normalized as "<call_id>|<item_id>"

Google Generative AI (google-generative-ai)

Source: packages/ai/src/providers/google.ts

Normalization points:

• iterates candidate.content.parts
• text parts are split into thinking vs text by isThinkingPart(part)
• block transitions close previous block before starting a new one
• part.functionCall is treated as a complete tool call (start/delta/end emitted immediately)
• finish reason mapped by mapStopReason() from google-shared.ts

Tool-call argument streaming:

• function call args arrive as structured object, not incremental JSON text
• implementation emits one synthetic toolcall_delta containing JSON.stringify(arguments)
• no partial JSON parser needed for Google in this path

Partial tool-call JSON accumulation and recovery

Shared behavior for Anthropic/OpenAI Responses uses parseStreamingJson() (packages/ai/src/utils/json-parse.ts):

1. try JSON.parse
2. fallback to partial-json parser for incomplete fragments
3. if both fail, return {}

Implications:

• malformed or truncated argument deltas do not crash stream processing immediately
• in-progress arguments may temporarily be {}
• later valid deltas can recover structured arguments because parsing is retried on every append
• final toolcall_end performs one more parse attempt before emission

Stop reasons vs transport/runtime errors

Provider stop reasons are mapped to normalized stopReason:

• Anthropic: end_turn→stop, max_tokens→length, tool_use→toolUse, safety/refusal cases→error
• OpenAI Responses: completed→stop, incomplete→length, failed/cancelled→error
• Google: STOP→stop, MAX_TOKENS→length, safety/prohibited/malformed-function-call classes→error

Error semantics are split in two stages:

1. Model completion semantics (provider reported finish reason/status)
2. Transport/runtime failure (network/client/parser/abort exceptions)

If provider stream throws or signals failure, each provider wrapper catches and emits terminal error event with:

• stopReason = "aborted" when abort signal is set
• otherwise stopReason = "error"
• errorMessage = formatErrorMessageWithRetryAfter(error)

Malformed chunk / SSE parse failure behavior

Most provider paths delegate chunk/SSE framing to vendor SDK streams (Anthropic SDK, OpenAI SDK, Google SDK). The Codex SSE fallback uses readSseJson() directly, and websocket Codex frames are normalized through the same event handler.

Observed behavior in current implementation:

• malformed SDK stream parsing surfaces as an exception or stream error event
• malformed Codex SSE JSON/framing throws from the local SSE reader
• provider wrapper converts failures into unified terminal error events
• no provider-specific resume/retry inside the stream function itself, except Codex websocket-to-SSE transport fallback before replay-unsafe output is emitted
• higher-level retries are handled in AgentSession auto-retry logic (message-level retry, not stream-chunk replay)

Cancellation boundaries

Cancellation is layered:

• AI provider request: options.signal is passed into provider client stream call.
• Provider wrapper: after stream loop, aborted signal forces error path ("Request was aborted").
• Agent loop: checks signal.aborted before handling each provider event and can synthesize an aborted assistant message from the latest partial.
• Session/agent controls: AgentSession.abort() -> agent.abort() -> shared abort controller cancellation.

Tool execution cancellation is separate from model stream cancellation:

• tool runners use AbortSignal.any([agentSignal, steeringAbortSignal])
• steering interrupts can abort remaining tool execution while preserving already-produced tool results

Backpressure boundaries

There is no hard backpressure mechanism between provider SDK stream and downstream consumers:

• EventStream uses in-memory queues with no max size
• throttling reduces UI update rate but does not slow provider intake
• if consumers lag significantly, queued events can grow until completion

Current design favors responsiveness and simple ordering over bounded-buffer flow control.

How stream events surface as agent/session events

agentLoop.streamAssistantResponse() bridges AssistantMessageEvent to AgentEvent:

• on start: pushes placeholder assistant message and emits message_start
• on block events (text_*, thinking_*, toolcall_*): updates last assistant message, emits message_update with raw assistantMessageEvent
• on terminal (done/error): resolves final message from response.result(), emits message_end

AgentSession then consumes those events for session-level behaviors:

• TTSR watches message_update.assistantMessageEvent for text_delta, thinking_delta, and toolcall_delta
• streaming edit guard inspects toolcall_delta/toolcall_end on edit calls and can abort early
• persistence writes finalized messages at message_end
• auto-retry examines assistant stopReason === "error" plus errorMessage heuristics

Unified vs provider-specific responsibilities

Unified (common contract):

• event shape (AssistantMessageEvent)
• final result extraction (done/error)
• delta throttling + merge rules
• agent/session event propagation model

Provider-specific (not fully abstracted):

• upstream event taxonomies and mapping logic
• stop-reason translation tables
• tool-call ID conventions
• reasoning/thinking block semantics and signatures
• usage token semantics and availability timing
• message conversion constraints per API

Implementation files

• ../../ai/src/stream.ts — provider dispatch, option mapping, API key/session plumbing, custom API dispatch, and provider-specific credential handling.
• ../../ai/src/utils/event-stream.ts — generic stream queue + assistant delta throttling.
• ../../ai/src/utils/json-parse.ts — partial JSON parsing for streamed tool arguments.
• ../../ai/src/providers/anthropic.ts — Anthropic event translation and tool JSON delta accumulation.
• ../../ai/src/providers/openai-responses.ts, openai-responses-shared.ts, openai-codex-responses.ts, azure-openai-responses.ts — Responses-family event translation and status mapping.
• ../../ai/src/providers/google.ts, google-gemini-cli.ts, google-vertex.ts — Gemini stream chunk-to-block translation variants.
• ../../ai/src/providers/google-shared.ts — Gemini finish-reason mapping and shared conversion rules.
• ../../ai/src/providers/amazon-bedrock.ts, openai-completions.ts, ollama.ts, cursor.ts, pi-native-client.ts — additional built-in stream adapters using the same event contract.
• ../../agent/src/agent-loop.ts — provider stream consumption and message_update bridging.
• ../src/session/agent-session.ts — session-level handling of streaming updates, abort, retry, and persistence.