Wire Protocols & Lazy SDKs

lazyStream hides async auth+SDK loading behind a synchronous stream

Learning Objectives

After this chapter you will be able to:

• Explain how lazyStream achieves "synchronous stream return, asynchronous SDK loading"
• Describe how event normalization hides per-vendor SSE/chunk differences
• Explain the impact of lazy imports on Bun bundle size
• Understand why "never throw" is the iron law of stream infrastructure

Key Source Files

File	Lines	Responsibility
packages/ai/src/api/anthropic-messages.ts	1345	Anthropic Messages wire protocol
packages/ai/src/api/openai-completions.ts	1504	OpenAI Completions wire protocol
packages/ai/src/api/openai-codex-responses.ts	1635	OpenAI Codex Responses wire protocol
packages/ai/src/api/lazy.ts	75	lazyStream: sync stream + async backplane

Design Motivation: Why lazyStream?

Problem: pi supports 40+ providers, each SDK is sizeable (openai ~200KB, @anthropic-ai/sdk ~150KB). Importing all at startup bloats the initial bundle to MB scale.

But: the agent loop's streamFn needs to return a stream synchronously — callers consume with for await and don't want to "wait for SDK loading before getting a stream."

Tension: We want lazy loading (async), but callers need a synchronous stream handle.

Analogy: Like a video site that "shows the player frame immediately while the video buffers behind the scenes" — the user sees the play interface (sync), video data loads in the background (async).

lazyStream: 75 Lines of Elegance

// lazy.ts — the entire file is only 75 lines
export function lazyStream(
  setup: () => Promise<AssistantMessageEventStream>
): AssistantMessageEventStream {
  // ① Immediately create a "deferred stream" — callers can for-await right away
  const stream = AssistantMessageEventStream.defer()

  // ② Run setup asynchronously (resolve auth, dynamic import SDK, connect)
  setup()
    .then(realStream => stream.connect(realStream))  // plug in when ready
    .catch(error => stream.error(error))             // failure goes through stream events

  // ③ Return synchronously — callers never wait
  return stream
}

Key design decisions:

• defer() creates an "empty shell stream" — internally buffers events until connect() plugs in the real stream
• Callers doing for await (const ev of stream) naturally wait, unaware of SDK loading
• Failure routes through stream.error() — never throws; upper layer handles via stopReason="error"

Per-Provider Lazy Wrappers

// anthropic-messages.lazy.ts (conceptual)
export function streamAnthropic(opts: StreamOptions): AssistantMessageEventStream {
  return lazyStream(async () => {
    // Dynamic import — SDK loaded only when Anthropic is actually used
    const { Anthropic } = await import("@anthropic-ai/sdk")
    const client = new Anthropic({ apiKey: opts.auth.apiKey })
    const raw = await client.messages.stream({ ... })
    return normalizeAnthropicEvents(raw)  // normalize to unified events
  })
}

Tree-shaking friendly: Each provider factory imports only its own lazy file. If the user only uses OpenAI, the Anthropic SDK is never loaded — the Bun build artifact stays lean.

Event Normalization: A Unified Language

Each vendor's SSE/chunk format is completely different:

Anthropic:  event: content_block_delta  data: {"delta":{"type":"text_delta","text":"Hello"}}
OpenAI:     data: {"choices":[{"delta":{"content":"Hello"}}]}
Google:     data: {"candidates":[{"content":{"parts":[{"text":"Hello"}]}}]}

But all are converted by their respective implementations into a unified event stream:

type AssistantMessageEvent =
  | { type: "text_delta"; text: string }
  | { type: "toolcall_delta"; id: string; argsJson: string }
  | { type: "thinking_delta"; text: string }
  | { type: "stop"; stopReason: StopReason }

The upper-layer agent knows only this shape — it neither knows nor cares whether the backend is Anthropic or OpenAI.

The "Never Throw" Iron Law

The first principle of stream infrastructure: errors are events, not exceptions.

// Error handling pattern
stream.on("error", (err) => {
  // Produce partial message + error info
  partialMessage.stopReason = "error"
  partialMessage.errorMessage = err.message
})

Why? Because the stream may have already yielded half a reply. Throwing an exception loses received content. Routing through events enables:

• Displaying the partial reply to the user ("the model stopped mid-sentence")
• Passing partial result + error together to retry logic
• TUI gracefully showing errors instead of crashing

Engineering Insights

1. 1345–1635 lines per protocol: Looks like a lot, but 80% handles vendor quirks — Anthropic's thinking blocks, OpenAI's function_call format evolution, Bedrock's base64 encoding. The cost of unified abstraction lives in the adapter layer.

2. lazy.ts is only 75 lines: The hallmark of a good abstraction — the core mechanism is minimal, complexity pushed into each provider's setup().

3. Collaboration with Ch09: lazyStream handles "connection," retry handles "reconnect after failure," partial JSON handles "parse even if incomplete" — the three combine into a complete streaming fault-tolerance chain.

Cross-Chapter Links

Direction	Chapter	Relationship
Prerequisite	Ch06 Models & Providers	Models.stream() delegates to provider.api.stream()
Prerequisite	Ch08 Auth & OAuth	setup() resolves auth internally
Follow-up	Ch09 Streaming & Retry	Retry and partial JSON after stream breaks
Follow-up	Ch14 Model Runtime	Rebuilding lazyStream on runtime provider switch

Check Your Understanding

1. If lazyStream were removed and replaced with "await import SDK first, then return stream," what impact would that have on agent loop callers?
2. Why does event normalization produce only 4 event types (text/toolcall/thinking/stop)? Is that sufficient?
3. In what scenarios is the "never throw" law superior to try/catch? In what scenarios might it mask problems?

Takeaways

• lazyStream = sync stream shell + async SDK loading — callers never wait
• SDKs imported on demand; Bun artifacts stay lean (tree-shaking friendly)
• Unified event stream (4 types) hides per-vendor SSE differences
• Errors are events, not exceptions — partial message + error flow together

---

This chapter is based on the main-branch source of earendil-works/pi.