One API for every model

pi-ai lets a caller talk to forty-odd providers through one call, models.stream(model, context), which returns immediately with an async-iterable stream of normalized events and a promise for the final message. This chapter covers the public contract: the Models collection, the normalized transcript, the event protocol, the queue that implements the stream, partial JSON for streaming tool calls, argument validation, token estimation and cost. Chapter 4 opens the provider adapters behind it.

Learning goals. Use pi-ai from code; state the invariants of the event protocol; explain how EventStream bridges a push-based producer to a pull-based consumer; explain why setup errors are delivered inside the stream; describe how partial tool-call arguments are parsed while they stream.

3.1 Using it#

The package README's quick start shows the whole surface in a few lines:

import { Type, type Context, type Tool } from '@earendil-works/pi-ai';
import { builtinModels } from '@earendil-works/pi-ai/providers/all';

const models = builtinModels();
const model = models.getModel('openai', 'gpt-4o-mini')!;

const tools: Tool[] = [{
  name: 'get_time',
  description: 'Get the current time',
  parameters: Type.Object({
    timezone: Type.Optional(Type.String({ description: 'Optional timezone (e.g., America/New_York)' }))
  })
}];

const context: Context = {
  systemPrompt: 'You are a helpful assistant.',
  messages: [{ role: 'user', content: 'What time is it?', timestamp: Date.now() }],
  tools
};

const s = models.stream(model, context);
for await (const event of s) { /* text_delta, toolcall_end, done, … */ }

Three nouns carry the design (all in packages/ai/src/models.ts):

• A Provider is a runtime unit for one vendor: an id, an auth description, getModels(), and the operations its models support (stream, streamSimple, optional deferred responses, image generation, classification).
• A Models collection holds providers, resolves credentials, and routes each request to the provider that owns the model. Its doc comment states the division of labour: "Providers own request behavior; Models resolves auth and delegates each request to the provider that owns the model."
• A Model is a catalog record (Chapter 5): provider, api, baseUrl, limits, prices.

complete() is literally this.stream(model, context, options).result() — the non-streaming call is the streaming call with the events ignored. One implementation, two conveniences.

3.2 Normalizing the request: the transcript and a branded type#

Callers pass a loose Context (systemPrompt?, messages, tools?). Providers receive a TranscriptContext, in which the prompt and tools have been folded into a leading SystemMessage. normalizeContext() in utils/transcript.ts is the only function that builds one:

export function normalizeContext(context: Context): TranscriptContext {
	const initialMessage = createInitialSystemMessage(context.systemPrompt, context.tools);
	const messages = initialMessage ? [initialMessage, ...context.messages] : context.messages;
	return { messages } as TranscriptContext;
}

How does the type system stop a raw Context from reaching provider code? types.ts declares a brand:

declare const transcriptContextBrand: unique symbol;
export type TranscriptContext = {
	messages: Message[];
	readonly [transcriptContextBrand]: true;
};

TypeScript types are structural: two types with the same shape are interchangeable. A property keyed by a unique symbol that exists only at the type level makes the type nominal in practice — no ordinary object literal has that key, so the only way to obtain a TranscriptContext is through the function that casts to it. The symbol is declared, never created, so it costs nothing at runtime.

Because Pi lets system messages appear mid-transcript (Chapter 1), the transcript module also knows how to replay them. getCurrentTools() walks all system messages in order, applying toolsRemoved then toolsAdded to a Map keyed by tool name. getCurrentSystemMessage() concatenates instruction text and patches named sections (a null deletes a section). For APIs that cannot take a system message in the middle, resolveTranscript() calls collapseSystemMessages(), which emits one replayed system message followed by every non-system message. This is a small instance of event sourcing: the current state is not stored, it is the fold of a log of changes.

3.3 The event protocol#

Every adapter, whatever its vendor, must emit this union (quoted from types.ts):

export type AssistantMessageEvent =
	| { type: "start"; partial: AssistantMessage }
	| { type: "text_start"; contentIndex: number; partial: AssistantMessage }
	| { type: "text_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
	| { type: "text_end"; contentIndex: number; content: string; partial: AssistantMessage }
	| { type: "thinking_start"; contentIndex: number; partial: AssistantMessage }
	| { type: "thinking_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
	| { type: "thinking_end"; contentIndex: number; content: string; partial: AssistantMessage }
	| { type: "toolcall_start"; contentIndex: number; partial: AssistantMessage }
	| { type: "toolcall_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
	| { type: "toolcall_end"; contentIndex: number; toolCall: ToolCall; partial: AssistantMessage }
	| { type: "done"; reason: Extract<StopReason, "stop" | "length" | "toolUse" | "deferred">; message: AssistantMessage }
	| { type: "error"; reason: Extract<StopReason, "aborted" | "error">; error: AssistantMessage };

The comment above it states the protocol's rules, which you can read as a small grammar:

stream  := start block* (done | error)  |  error
block   := text_start text_delta* text_end
         | thinking_start thinking_delta* thinking_end
         | toolcall_start toolcall_delta* toolcall_end

• "Successful streams emit start before partial updates and terminate with done."
• "A stream may terminate directly with error when request setup fails before generation starts."
• "Updates and done must never appear before start."
• partial "is the shared live response-so-far helper, not an event-time snapshot": every event carries a reference to the same growing AssistantMessage. If you store events and inspect partial later, you will see the final state, not the state at that event. Copy it if you need a snapshot.
• contentIndex points into partial.content, so blocks can be interleaved (a provider may open a tool call while text is still streaming) and consumers update the right block.

Notice the type-level precision: done.reason is Extract<StopReason, "stop" | "length" | "toolUse" | "deferred">, so a done event can never claim "aborted", and an error event can never claim "stop". Illegal states are unrepresentable.

3.4 The stream itself: a queue between a producer and a consumer#

The adapter is a producer: network chunks arrive whenever they arrive and it calls push(event). The caller is a consumer: it pulls with for await. These run at different speeds. EventStream<T, R> in utils/event-stream.ts connects them with two queues:

• queue holds events that were pushed before anyone asked for them.
• waiting holds resolver functions of consumers who asked before any event existed.

push(event: T): void {
	if (this.done) return;
	if (this.isComplete(event)) {
		this.done = true;
		this.resolveFinalResult(this.extractResult(event));
	}
	// Deliver to waiting consumer or queue it
	const waiter = this.waiting.dequeue();
	if (waiter) {
		waiter({ value: event, done: false });
	} else {
		this.queue.enqueue(event);
	}
}

async *[Symbol.asyncIterator](): AsyncIterator<T> {
	while (true) {
		if (this.queue.length > 0) {
			yield this.queue.dequeue()!;
		} else if (this.done) {
			return;
		} else {
			const result = await new Promise<IteratorResult<T>>((resolve) => this.waiting.enqueue(resolve));
			if (result.done) return;
			yield result.value;
		}
	}
}

At any moment at most one of the two queues is non-empty: either events are waiting for consumers or consumers are waiting for events. A terminal event (done or error) flips done and resolves a separate finalResultPromise, which is what result() returns. end() wakes every waiting consumer with { done: true } so no for await hangs forever.

The queue class is itself a classic: FifoQueue implements a first-in-first-out queue with two stacks.

dequeue(): T | undefined {
	if (this.outgoing.length === 0) {
		while (this.incoming.length > 0) {
			this.outgoing.push(this.incoming.pop()!);
		}
	}
	return this.outgoing.pop();
}

JavaScript's Array.prototype.shift() is O(n) because it re-indexes the array. Pushing onto incoming and popping from outgoing is O(1); when outgoing runs dry, all of incoming is reversed into it. Each element moves at most twice, so every operation is amortized O(1). This matters when a fast provider emits thousands of small deltas.

Two properties to note. The stream is unbounded: if nobody consumes, events accumulate in memory. That is acceptable because one response is finite and bounded by maxTokens; there is no backpressure signal to slow the producer, and none is needed. And the stream is single-consumer: two for await loops on the same stream would split the events between them, not each receive all of them. Consumers that need fan-out (the agent loop, Chapter 6) re-emit events themselves.

AssistantMessageEventStream subclasses it and adds timing: the final message gets durationMs measured with performance.now(), a monotonic clock. Wall-clock time (Date.now()) can jump backwards when the system clock is corrected; a monotonic clock cannot, so durations computed from it are never negative.

3.5 Errors travel inside the stream#

Models.stream() must resolve credentials (possibly refreshing an OAuth token over the network) and may need to import the adapter module lazily. Both are asynchronous, yet stream() returns synchronously. lazyStream() in api/lazy.ts makes that possible:

export function lazyStream(
	model: Model<Api>,
	setup: () => Promise<AsyncIterable<AssistantMessageEvent>>,
): AssistantMessageEventStream {
	const startedAt = Date.now();
	const outer = new AssistantMessageEventStream();

	setup()
		.then((inner) => forwardStream(outer, inner))
		.catch((error) => {
			const message = createSetupErrorMessage(model, error, startedAt);
			outer.push({ type: "error", reason: "error", error: message });
			outer.end(message);
		});

	return outer;
}

It returns an empty outer stream at once, runs setup in the background, and forwards the inner stream's events when ready. If setup throws, the error becomes an error event carrying an AssistantMessage with stopReason: "error" and an errorMessage. The StreamFunction contract in types.ts makes this a rule for every adapter: "Once a stream is returned, request/model/runtime failures should be encoded in that stream. Error termination must produce an AssistantMessage with stopReason "error" or "aborted"."

Why not just throw? Because consumers would need two error paths — a try/catch around the call and error events inside the loop — and an error that happens after some text has streamed cannot be thrown to a caller that has already received a stream. Encoding failures as data gives one path, and the error message can be persisted into the session like any other assistant message. The same idea appears in generateImages() and classify(), documented as "Never rejects".

lazyApi() applies the trick to modules: it wraps a dynamic import() of an adapter as a ProviderStreams object, so the Bedrock SDK, for example, is loaded only the first time a Bedrock model is used.

3.6 Routing a request#

ModelsImpl.stream() normalizes the context, then inside lazyStream finds the provider and applies auth:

stream<TApi extends Api>(model: Model<TApi>, context: Context, options?: ModelsApiStreamOptions<TApi>) {
	const transcript = normalizeContext(context);
	return lazyStream(model, async () => {
		const provider = this.requireChatProvider(model);
		const { requestModel, requestOptions } = await this.applyAuth(model, options as ModelsApiStreamOptions<Api> | undefined);
		return provider.stream(requestModel, transcript, requestOptions as ApiStreamOptions<TApi>);
	});
}

applyAuth() merges in a fixed precedence: "Explicit request options win per-field; the Models-only transform runs last." An explicit apiKey beats the resolved credential; caller headers override provider headers; a credential may also redirect baseUrl (for example to a gateway). Providers are usually assembled by createProvider(), which accepts either one api implementation or a map from api name to implementation and dispatches on model.api — the strategy pattern with a lookup table instead of a class hierarchy. A model whose api has no implementation gets, unsurprisingly, an error stream.

3.7 stream versus streamSimple: provider-specific and provider-neutral options#

Each adapter accepts its own options type (Anthropic's thinking display, OpenAI's reasoning summary, …), mapped at the type level by ApiOptionsMap and ApiStreamOptions<TApi>, a conditional type that resolves to the concrete options for known APIs. streamSimple() takes SimpleStreamOptions instead, whose key addition is a provider-neutral reasoning?: ThinkingLevel. Each adapter's streamSimple translates the level into its vendor's knob.

Pi's thinking levels are ordered: off, minimal, low, medium, high, xhigh, max. Not every model supports every level; a model's thinkingLevelMap maps levels to vendor values, with null meaning unsupported. clampThinkingLevel() picks the nearest supported level, searching upward first and then downward:

for (let i = requestedIndex; i < EXTENDED_THINKING_LEVELS.length; i++) {
	const candidate = EXTENDED_THINKING_LEVELS[i];
	if (availableLevels.includes(candidate)) return candidate;
}
for (let i = requestedIndex - 1; i >= 0; i--) {
	const candidate = EXTENDED_THINKING_LEVELS[i];
	if (availableLevels.includes(candidate)) return candidate;
}

For token-budget providers, DEFAULT_THINKING_BUDGETS in api/simple-options.ts maps levels to budgets (minimal: 1024, low: 2048, medium: 8192, high: 16384 tokens), and MIN_ANSWER_TOKENS = 1024 reserves room for the visible answer when thinking and answer share one ceiling. clampMaxTokensToContext() shrinks maxTokens so that estimated input + output + a 4,096-token safety margin fits the context window.

3.8 Streaming tool calls: parsing JSON that is not finished#

Tool-call arguments stream as JSON text fragments: {"pa, th": "src/, index.ts"}. A UI wants to show the path as soon as possible, but JSON.parse('{"path": "src/') throws. utils/json-parse.ts provides a ladder of increasingly forgiving strategies in parseStreamingJson():

1. JSON.parse, and if that fails, JSON.parse of a repaired string;
2. the partial-json library, which closes open strings, arrays and objects and returns the best prefix value;
3. partial-json on the repaired string;
4. {}.

repairJson() is a single-pass scanner with one bit of state, inString. Outside strings it copies characters. Inside strings it escapes raw control characters (a literal newline becomes \n), keeps valid escapes, keeps \uXXXX with four hex digits, and doubles a backslash before anything else. Models often produce Windows paths like "C:\src", where \s is not a legal JSON escape; repair turns it into "C:\\src".

Once a call is finished the rules tighten. parseToolCallArguments() never keeps a partial prefix: if the final text is not a JSON object, the call gets arguments: {} and an argumentsError that includes the text the model sent, "so tool validation can report it back to the model." The asymmetry is deliberate: during streaming, a best guess is fine for display; at execution time, guessing would run a tool with arguments the model did not intend.

3.9 Validating arguments against the schema#

validateToolArguments() in utils/validation.ts is the gate before execution:

1. If argumentsError is set, throw it.
2. structuredClone the arguments (never mutate the message), normalize optional nulls, and let TypeBox's Value.Convert coerce obvious mismatches (the string "10" for an integer field).
3. Check with a compiled validator. Compile(schema) generates a specialized checking function once; results are cached in a WeakMap keyed by the schema object, so the cache entry disappears when the schema is garbage-collected.
4. On failure, throw a message listing each failing path and the received arguments.

The thrown message does not crash anything: the agent loop turns it into an error tool result (Chapter 6), so the model reads exactly what was wrong and usually retries correctly. Validation errors are feedback, not exceptions in the usual sense.

3.10 Estimating tokens without a tokenizer#

Exact token counts require each vendor's tokenizer, which pi-ai does not ship. utils/estimate.ts uses a heuristic (CHARS_PER_TOKEN = 3.5, images counted as 4,800 characters) and anchors it to ground truth: estimateContextTokens() finds the most recent successful assistant message, takes its provider-reported usage as exact for everything up to that point, and estimates only the messages after it. A subtle check skips usage that no longer describes the prefix: if a newer message was inserted before that response ("for example, a compaction summary"), its usage is ignored. This hybrid — exact where known, estimated at the tail — keeps the error small because the tail is short.

3.11 Money#

calculateCost() in models.ts turns Usage into dollars. Prices are per million tokens. Some models have tiered prices above an input size; the function picks the tier with the largest threshold that the input exceeds. Anthropic bills 1-hour cache writes at twice the base input rate, so writes are split:

const longWrite = usage.cacheWrite1h ?? 0;
const shortWrite = usage.cacheWrite - longWrite;
usage.cost.input = (rates.input / 1000000) * usage.input;
usage.cost.output = (rates.output / 1000000) * usage.output;
usage.cost.cacheRead = (rates.cacheRead / 1000000) * usage.cacheRead;
usage.cost.cacheWrite = (rates.cacheWrite * shortWrite + rates.input * 2 * longWrite) / 1000000;

Costs are IEEE-754 floating-point dollars. For display and budgeting this is fine; an accounting system would use integer micro-units to avoid accumulated rounding error.

3.12 Concepts in this chapter#

• Async iterators and generators. An object with [Symbol.asyncIterator] can be consumed with for await. An async function* produces such an iterator; yield hands one value to the consumer and suspends until the next is requested.
• Producer/consumer with a rendezvous queue. Pending items or pending takers are queued, whichever side arrives first; the other side completes the pair. Go channels and Java's SynchronousQueue are relatives.
• Amortized analysis. An operation that is occasionally expensive can still be O(1) on average if the expensive work is paid for by many cheap operations (each element is moved at most twice in the two-stack queue).
• Backpressure. A way for a slow consumer to slow a fast producer. Its absence is a deliberate choice when the total data is bounded.
• Errors as values. Encoding failure in the result type (here, an error event and stopReason) gives one control path and makes failures persistable.
• Branded (nominal) types. A phantom unique symbol property makes structurally identical types incompatible, forcing construction through a trusted function.
• Event sourcing. State as the left fold of a log of changes; replaying system messages is a tiny example.
• Incremental and error-tolerant parsing. Parsing a prefix of a language and recovering from known malformations.
• Schema validation and memoization. Compile a checker once per schema; cache by object identity in a WeakMap so the cache never leaks.
• Monotonic vs wall clock. Measure durations with a clock that never goes backwards.
• Strategy pattern / dispatch tables. Select behaviour at runtime from a map keyed by data (model.api).

Exercises#

1. Write a small checker that consumes an AssistantMessageEventStream and throws if the grammar in 3.3 is violated (for example, a text_delta whose contentIndex was never opened).
2. Prove that in EventStream, queue and waiting are never both non-empty. What would break if they were?
3. Replace FifoQueue with an array using shift() and benchmark pushing and draining 1,000,000 events. Explain the result.
4. Show what repairJson returns for {"cmd": "echo "hi""}. Why can't a single-bit scanner fix this case?
5. A consumer stores every text_delta event in an array and later reads event.partial.content[0].text from the first one. What does it see and why?
6. clampThinkingLevel searches upward before downward. Argue for and against that choice for a user who asked for medium on a model that supports only off and high.
7. Modify calculateCost to use integer micro-dollars. Where must rounding happen?