Part II Chapter 3: Issue the Ticket First, Do the Work After — pi-ai

Chapter guide: Stop ② on the source route (core layer). Still in the pi-ai package, but deeper into api/lazy.ts and api/transform-messages.ts. What this stop has to take apart is the AssistantMessageEventStream that left Xiao A staring blankly: how vendors' scattered incremental events get gathered into the unified shape the loop can consume directly.

With Part II Chapter 2 done, Xiao A finally got the model to "depart". But what the Provider handed back froze him again: not a complete answer, but an AssistantMessageEventStream. He stared at the stream and asked Lao Z, "How am I supposed to consume this? Wait for it to finish entirely, or take the events one at a time?"

"Neither," Lao Z said. "Part II Chapter 2: The Catalog and the Household Register — pi-ai finished the dispatch from 'model object to Provider', but what the Provider returns is not a finished string — it is an AssistantMessageEventStream. Three things you now have to verify: why the asynchronous setup does not block the entry point; how one real adapter turns the vendor's stream into unified events; and what status cancellation and exceptions ultimately report upward."

The evidence in this chapter is pinned to the upstream pi tag v1.0.0 (commit a13d35a742c6ef8462812a28fbe1d8c8b7431c32) and @earendil-works/pi-ai 1.0.0. This chapter picks packages/ai/src/api/openai-responses.ts to read as one real adapter. That is not a claim that every Provider or API adopts its fields and failure strategy; it does not cover every vendor protocol, model pricing rule, or WebSocket adapter, and it does not write the existence of a general retry helper as "every call retries".

Before returning a stream, leave the setup for later

"ModelsImpl.stream and streamSimple both call lazyStream. The function immediately puts an iterable stream into the caller's hands, while auth resolution, dynamic module loading, and the Provider call all proceed in the background," Lao Z said. "So holding the stream is not the same as 'the request has certainly started', and it is far from an ordinary Promise<AssistantMessage>."

// packages/ai/src/api/lazy.ts, lazyStream (v1.0.0, excerpt)
export function lazyStream(model, setup): AssistantMessageEventStream {
	const outer = new AssistantMessageEventStream();
	setup()
		.then((inner) => forwardStream(outer, inner))
		.catch((error) => {
			const message = createSetupErrorMessage(model, error);
			outer.push({ type: "error", reason: "error", error: message });
			outer.end(message);
		});
	return outer;
}

Xiao A: "Why go the long way around like this? Can't we just wait for setup to finish and then return?"

"Look at it paragraph by paragraph." Lao Z pointed at the code. "The first line constructs outer, so the call site synchronously gets the same output type and need not distinguish 'still authenticating' from 'already receiving network events'. The second part does not await setup(); it attaches forwardStream through a promise continuation — that is exactly the deferred setup: the requireProvider, applyAuth, and provider.stream* calls inside ModelsImpl are all placed into the async callback that gets passed in. Note that 'deferred' is said relative to the synchronous return; it does not promise to wait until the first for await to run — setup() is called right inside the function body."

"The third part normalizes a setup rejection into an AssistantMessage: empty content, zero usage, stopReason: "error", and the error text; then it pushes an error event and calls end(message). So an unknown Provider or a failed authentication presents itself to the consumer as an in-stream error, not a synchronous throw from the call to Models.stream."

Metaphor: issue the ticket first, do the work after

lazyStream is like handing you a pickup number first (the outer stream) while the kitchen (setup) takes its time cooking. If the dish is ruined, you still hear it through the pickup window — "this order can't be made" — rather than having a plate smashed at you the moment you place the order.

forwardStream pushes the inner stream's events one by one, and after the iteration ends it closes outer with the inner stream's .result(), if there is one. This rule preserves the final message the adapter produced; do not infer from it that any event must contain a text delta.

Stage	Product	Normal next step	When setup fails
Models.stream* returns	outer stream	the caller subscribes / iterates	no synchronous exception yet
setup()	inner async iterable	forwardStream	.catch creates the error message
inner events	start, delta, done / error, etc.	pushed to outer as-is	the adapter decides the event contents
inner completion	result() or undefined	outer.end	determined by the adapter's termination behaviour

The lazy API is a second layer of delay, not a second event protocol

Xiao A followed up: "What about the adapter modules inside a Provider? Are they lazy-loaded too?"

"Yes, but that is the second layer of delay." Lao Z opened lazy.ts:

lazyApi(load, capabilities?) returns a ProviderStreams object. Its two stream methods each wrap the awaited module's method in lazyStream again; the capabilities added in v1.0.0 declare whether this API still supports retrieving and cancelling deferred responses — not expanded in this chapter:

// packages/ai/src/api/lazy.ts, lazyApi (v1.0.0, excerpt)
const api: ProviderStreams = {
	stream: (model, context, options) =>
		lazyStream(model, async () => (await load()).stream(model, context, options)),
	streamSimple: (model, context, options) =>
		lazyStream(model, async () => (await load()).streamSimple(model, context, options)),
};

"The outer lazy stream in ModelsImpl covers 'Provider / auth setup'; the lazy stream here covers 'adapter module loading'. Both layers produce the same AssistantMessageEventStream, so you get nested forwarding. Whether the module import cache de-duplicates is the host runtime's responsibility — a source comment does mention the host import cache, but no performance outcome has been proven for arbitrary runtime environments."

"The benefit: unused adapters need not load at the root entry point. The cost: module loading errors are postponed until the first request, and when you debug, you still have to tell the three layers apart — auth setup, import setup, and the network request."

One normal event flow through a real adapter

Xiao A suggested: "Pick a real adapter and see how the happy path runs?"

"Look at openai-responses.ts." Lao Z said. "Its stream creates the AssistantMessageEventStream at the very start of the function, then uses resolveTranscript to process the transcript per compat: when the model does not accept system messages mid-conversation, all system messages are merged into a single one at the beginning. Then it starts an async IIFE. The IIFE first builds an output with stopReason: "pending", pushes start only after success, lets processResponsesStream update it, and then pushes done. The tool list is no longer read from context.tools either; getDeclaredTools replays it from the system message."

// packages/ai/src/api/openai-responses.ts, stream (v1.0.0, excerpt)
const { data: openaiStream, response } = await retryProviderRequest(
	() => client.responses.create(params, requestOptions).withResponse(),
	{ maxRetries: options?.maxRetries, maxRetryDelayMs: options?.maxRetryDelayMs, signal: options?.signal },
);
await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
stream.push({ type: "start", partial: output });
await processResponsesStream(openaiStream, output, stream, model, { /* excerpt */ });
stream.push({ type: "done", reason: output.stopReason, message: output });
stream.end();

"Pay attention to the order in this snippet." Lao Z warned. "The request goes through retryProviderRequest, but the adapter explicitly passes maxRetries: 0 to the OpenAI SDK itself and hands the retry policy to the outer helper. Whether a retry happens is still decided by options.maxRetries, the signal, and the helper's retryability check — do not read the presence of this layer in the code as 'every error must be retried'."

"onResponse fires after the HTTP response arrives and before start is pushed, giving you the status and the headers converted into a record. The processResponsesStream that follows is where upstream events actually get converted into unified deltas. This chapter does not walk through its OpenAI event cases one by one: those cases are details of this one adapter, not a cross-Provider contract."

A normal sequence can be read as:

buildParams / createClient
  → retryProviderRequest issues the Responses request
  → HTTP response arrives, onResponse is called
  → push start(partial output)
  → processResponsesStream keeps modifying output and pushes text/thinking/tool-call events
  → output gets a stopReason that is neither pending nor aborted/error
  → push done(message), end()

"For the Agent, what matters most is not what the upstream event is called, but that it first sees an updatable partial and finally receives a complete AssistantMessage. That is exactly how streamAssistantResponse in the next chapter can write start into the context and turn deltas into message_update."

transform and compat: a unified shape does not mean the differences are gone

Xiao A put the two names side by side: "Earlier, applyAuth had transformHeaders, and here there's getCompat. Are they the same thing?"

"No. They operate at different layers," Lao Z said. "The transformHeaders in ModelsImpl.applyAuth is a Models-specific transformation applied before the request is sent; openai-responses.ts, for its part, builds a default compatibility configuration for the model's compat."

// packages/ai/src/api/openai-responses.ts, getCompat (v1.0.0, excerpt)
function getCompat(model: Model<"openai-responses">): Required<OpenAIResponsesCompat> {
	return {
		supportsDeveloperRole: model.compat?.supportsDeveloperRole ?? true,
		supportsMidConvoSystemMessages: model.compat?.supportsMidConvoSystemMessages ?? false,
		sessionAffinityFormat: model.compat?.sessionAffinityFormat ?? detectSessionAffinityFormat(model),
		supportsLongCacheRetention: model.compat?.supportsLongCacheRetention ?? true,
		// remaining capability flags elided
	};
}

"transformHeaders takes the already-merged header record as input, which suits a host adding tracking or routing information; it does not touch the adapter's message structure. getCompat infers capability flags from model.compat plus the provider / baseUrl, and the flags are then used by paths such as resolveTranscript, buildParams, and grammar tool input. Capability differences can still exist under the same API name, and a default of true does not mean the remote end necessarily accepts the capability."

Mechanism	Where it happens	What it acts on	Failure or limits
transformHeaders	ModelsImpl.applyAuth	the merged headers	a throw becomes a lazy setup error
onPayload	after the adapter assembles params	the provider payload	the return value can replace params; a failing hook enters the catch
getCompat	before the adapter builds params	Model capability flags	the defaults are a local choice, not server-side proof
processResponsesStream	after the request is established	upstream events → unified events	the concrete fields apply only to this adapter

Xiao A pointed at the getCompat row in the table: "That sessionAffinityFormat falls through to detectSessionAffinityFormat by default — what does it actually look at?"

"Two checks," Lao Z said. "One: model.provider === "openrouter", or the baseUrl contains openrouter.ai. If either hits, it picks openrouter; otherwise the default is openai. This is a local heuristic, not a server-side declaration. Later, when createClient builds the session headers, it uses the result: with a sessionId and openrouter, it sends x-session-id; on the openai branch it sends session_id plus x-client-request-id. So the compat flag does change the headers that get sent. As for the default value, it only represents the local side's choice."

"Then does that branch count as an error path?" Xiao A pressed.

"What deserves remembering more is the positional difference of getClientApiKey. In stream it sits inside the async IIFE's try, so a missing key throws into the catch and becomes an error event; in streamSimple it is called synchronously, before stream is constructed. So if you call the adapter's streamSimple directly, a missing key is a synchronous throw and does not go through in-stream error handling; only when it runs through the Models collection's async setup closure does it get normalized into an error event. This echoes the line from the last section: bypass ModelsImpl, and satisfying the contract is on you."

Aborts, errors, and incomplete streams must not be swallowed

Xiao A listed the situations that worried him all at once: "abort, a stream cut off before ending, an exception — how does each get handled mid-stream?"

"The tail of a real adapter has three checks: after the request, if signal.aborted it throws; if the upstream stream ends while still pending, it reports 'no stop reason'; if the output is already aborted / error, that also turns into the error path, carrying the errorMessage the output already has. A stream without complete termination information does not get misreported as a success here."

// packages/ai/src/api/openai-responses.ts, the tail of stream and its catch (v1.0.0, excerpt)
if (options?.signal?.aborted) throw new Error("Request was aborted");
if (output.stopReason === "pending") throw new Error("OpenAI Responses stream ended without a stop reason");

// inside the catch:
output.stopReason = options?.signal?.aborted ? "aborted" : "error";
const errorMessage = formatProviderError(
	normalizeProviderError(error),
	`${model.provider === "openai" ? "OpenAI" : model.provider} API error`,
);
output.errorMessage = /* when subscription quota is exhausted, a ChatGPT usage-page link is appended */ errorMessage;
stream.push({ type: "error", reason: output.stopReason, error: output });
stream.end();

"Before the catch, it also deletes the index, partialJson, and customInput fields on the content blocks that exist only for streaming parsing. This is a verifiable cleanup behaviour — don't exaggerate it into 'intermediate data is never kept': logging and stores in other layers are all outside this function."

Path	Trigger condition	Event / final stop reason	Recoverability boundary
Normal completion	the parser fills in a valid stop reason	done, the original stop reason	the caller may continue processing tool calls
Deliberate abort	the AbortSignal is already aborted	error, aborted	whether to retry is decided by the layer above
Request / parse exception	the SDK, a hook, params, or the network throws	error, error	errorMessage is normalized by the formatter
Upstream has no stop reason	ends while still pending	error, error	prevents treating a half-finished response as success
Lazy setup failure	auth, Provider, or module loading rejects	error, error	this adapter's normal start was never reached

Xiao A stared at the table: "Why does an abort also use the error event instead of done?"

"The event type here expresses the 'not-normal-completion channel'; it is the reason that distinguishes aborted from error," Lao Z said. "So a consumer must not look at the event type alone. In Part II Chapter 4: The Double Loop — pi-agent-core you will see that the loop stops according to the stop reason, rather than treating aborted as 'tool calls are empty'."

"When the catch synthesizes the errorMessage, is the HTTP status thrown away? What can I actually read?"

"It is synthesized by formatProviderError(normalizeProviderError(error), prefix). The prefix is 'OpenAI API error' when the provider is openai, and otherwise the provider ID. normalize first probes the status by SDK field shape — statusCode, status, $metadata.httpStatusCode, $response.statusCode — taking the first number in that order. For the body it accepts only plain objects or strings: class instances and stream objects both count as no body, and the content is truncated at 4000 characters. format assembles prefix (status): body only when it has both a status and a body; otherwise it keeps the SDK's own error.message, so useful text is not replaced by noise. When the message contains subscription_sharing_usage_limit_exceeded, the adapter also appends a line with the ChatGPT usage-page link."

"So the errorMessage is a synthesized display string, not a structured error object," Lao Z summed up. "It carries a prefix and preserves the status and body as best it can. Consumers that need to branch should look at the stopReason and the event itself first — don't parse this text. The difference between aborted and error is already given by the reason; there is no need to guess it from the text."

The consumer should keep the final message, not concatenate the text itself

Xiao A wanted a shortcut: "So if I just concatenate the text_deltas, I get the complete answer, right?"

"You would lose things," Lao Z said. "The adapter publishes partial events in the stream so the consumer can display and update state promptly. The final AssistantMessage carries more than text: usage, provider, model, stop reason, and possibly tool-call content. Concatenating only the text_deltas into a string throws away all the information the Agent needs for its next step."

Consumption strategy	What you get	What is lost or at risk
Display deltas only	a low-latency text experience	tool calls, usage, stop reason are incomplete
Wait only for the result	the complete message	no streaming visibility
Consume both events and result	partial UI plus the final state	you must avoid treating partials as persisted final values
Immediately ignore an error	quick return	erases the difference between aborted and a provider error

"forwardStream in lazyStream does not rewrite the inner events, so the fine-grained semantics of the event sequence are still decided by the adapter. A setup failure is uniformly synthesized into an error event, so callers can handle configuration failures through one and the same subscription surface; but that is no guarantee that 'all failures carry the same message text' — createSetupErrorMessage fills the errorMessage from error.message or String(error), and an adapter may use its own formatter."

"Retry has to be looked at outside the consumption semantics. retryProviderRequest covers only the await in which this adapter issues the Responses request. It does not enclose the Agent's subsequent tool execution, nor the session append. If the request had already begun emitting output before being interrupted, whether you may retry depends on whether the caller accepts duplicate generation. For tool results with external side effects, the retry rules for model requests cannot simply be copied over. That is a general engineering judgement, not an automatic protection of the OpenAI Responses adapter."

Benefits, costs, and scope not covered

From the source we can infer this: unifying API adapters behind an event stream lets the Agent display text, thinking, and tool-call increments inside one and the same loop, while Providers can still preserve their differences through typed options, compat, and payload hooks. The cost is that error sources span multiple layers, and one generic event type automatically smooths away neither the remote semantics of billing, caching, nor cancellation.

This chapter did not verify whether live services respond, the full algorithm of retry backoff, every event mapping in processResponsesStream, or Anthropic, Google, or WebSocket adapters. Those require reading each implementation separately and running them under controlled credentials; the OpenAI Responses code cannot stand in for them.

Summary

What this chapter answered: how exactly to consume the AssistantMessageEventStream a Provider returns. It is not a complete object you wait for and then read, but an incremental event flow with a lifecycle: lazyStream gives the entry point a unified stream handle immediately and uniformly converts auth and module-loading failures into an error message; the adapter's normal path is request, start, event transformation, done; the error path uses reason to tell abort and error apart, and refuses half-finished streams that have no stop reason.

• Boundary: the header transform, the payload hook, and compat each solve different problems at different layers; none substitutes for another. Consumers must not look at the event type alone but also at the reason: aborted and error are two different recovery paths.

Remember: a stream has a lifecycle; it is not one return. Incremental events, the stop reason, and the refusal of half-finished streams — these three together define a stream's terminal shape. The loop must decide its next step from the stop reason, not treat aborted as "the tool calls are empty".

Source Walkthrough

1. Read createSetupErrorMessage, forwardStream, and lazyStream in packages/ai/src/api/lazy.ts; confirm that a setup error ends the outer stream with an assistant message carrying zero usage.
2. Search for lazyStream(model in packages/ai/src/models.ts; mark out the setup closures of stream and streamSimple separately, and confirm that both call the Provider inside the closure.
3. Open stream in packages/ai/src/api/openai-responses.ts; follow from retryProviderRequest to stream.push({ type: "start" and write down at which stage of the HTTP response the start appears.
4. Search for getCompat and buildParams; pick one compatibility flag and confirm it is used in parameter construction, not decided automatically by the root Models.
5. In the catch of the same stream, inspect the removal of partialJson and the assignment to stopReason; reason through how the reason differs when the signal is already aborted versus when an ordinary Error is thrown.