Streaming, Partial JSON & Retries

Errors are events, never thrown: stopReason error/aborted

Learning Objectives

After this chapter you will be able to:

• Explain why the "errors as events" paradigm is superior to try/catch
• Describe the push/end/error three primitives of AssistantMessageEventStream
• Explain how parsePartialJson tolerantly parses when brackets are unclosed
• Understand RetryPolicy's exponential backoff strategy and RetryCallbacks hooks
• Distinguish the two strictness levels of constrained-sampling: prefer/require

Key Source Files

File	Lines	Responsibility
packages/ai/src/utils/event-stream.ts	~180	AssistantMessageEventStream: push/end/error primitives
packages/ai/src/utils/json-parse.ts	~90	parsePartialJson + closeJsonBrackets
packages/ai/src/utils/retry.ts	227	retryAssistantCall: exponential backoff + RetryCallbacks
packages/ai/src/api/constrained-sampling.ts	~200	JSON-schema strict / Lark / regex grammar
packages/ai/src/utils/validation.ts	~120	Response validation: schema compliance check

Design Motivation: Why "Errors as Events"?

Problem: LLM streaming responses can break mid-way — network jitter, 429 rate limits, model timeouts. Under traditional try/catch, once an exception throws, all received partial content is lost.

Analogy: Like a voice recorder — even if the battery dies suddenly, the recorded portion remains on the memory card. You wouldn't discard the entire recording because the last 5 seconds weren't captured.

Alternative comparison:

Approach	Pros	Fatal Flaw
try/catch wrapping entire stream	Simple, intuitive	Loses already-received partial content
Per-chunk try/catch	Preserves partial	Fragmented code, unclear error semantics
Errors as events	Partial message + error flow together	Consumer must handle error events

pi chooses "errors as events": on failure the stream emits an error event + sets stopReason="error"/"aborted", and yields the received content as partialMessage. The upper layer then decides: display partial reply? retry? degrade?

AssistantMessageEventStream: Three Primitives

class AssistantMessageEventStream {
  push(event: AssistantMessageEvent): void   // emit one event
  end(): void                                 // normal termination
  error(err: Error): void                     // abnormal termination (no throw!)

  // Consumer side
  async *[Symbol.asyncIterator](): AsyncGenerator<AssistantMessageEvent>
}

Lifecycle:

push(text_delta) → push(text_delta) → push(toolcall_delta) → ...
    │                                                           │
    ├── normal path ─────────────────────────────→ end()         │
    │                                                           │
    └── error path ──→ error(err)                                │
                     ├── stopReason = "error"                   │
                     └── partialMessage = received content      │
                                                                ▼
                                              agent loop consumes (→ Chapter 1)

Key constraint: After error() the stream closes immediately — no more events. Consumers seeing stopReason="error" know the stream is terminated; no catch needed.

Partial JSON Parsing: Use While Writing

A tool call's argsJson arrives incrementally:

chunk 1: '{ "path": "src/a'
chunk 2: '.ts", "old"'
chunk 3: ': "x", "ne'
chunk 4: 'w": "y" }'

Waiting for complete JSON before parsing means the TUI cannot show tool parameters in real time. parsePartialJson's strategy:

function parsePartialJson(s: string): unknown {
  // ① Try direct parse first (may already be complete)
  try { return JSON.parse(s) } catch {}

  // ② Close unclosed brackets, then parse
  try { return JSON.parse(closeBrackets(s)) } catch {}

  // ③ If all else fails, return null — never throws
  return null
}

function closeBrackets(s: string): string {
  let depth = 0, inStr = false
  for (const ch of s) {
    if (ch === '"' && !inStr) inStr = true
    else if (ch === '"' && inStr) inStr = false
    else if (!inStr && ch === '{') depth++
    else if (!inStr && ch === '}') depth--
  }
  return s + '}'.repeat(Math.max(0, depth))
}

Effect: When chunk 2 arrives, '{ "path": "src/a.ts", "old"' is closed to '{ "path": "src/a.ts", "old"}' → parses to {path: "src/a.ts"} — TUI can display "editing src/a.ts" in real time.

Analogy: Like reading a half-written letter — you don't need the author to finish before understanding the paragraphs already written.

RetryPolicy: Exponential Backoff + Hooks

interface RetryCallbacks {
  onRetry?: (attempt: number, delay: number, error: Error) => void
  onGiveUp?: (totalAttempts: number, lastError: Error) => void
}

async function retryAssistantCall(
  fn: () => Promise<AssistantMessageEventStream>,
  policy: RetryPolicy,
  callbacks?: RetryCallbacks
): Promise<AssistantMessageEventStream> {
  for (let attempt = 1; attempt <= policy.maxAttempts; attempt++) {
    const stream = await fn()
    const result = await collectStream(stream)

    if (result.stopReason !== "error") return stream  // success

    // Only back off on retryable errors
    if (!isRetryable(result.errorCode)) break  // 400/401 not retried

    const delay = policy.baseDelay * 2 ** (attempt - 1)  // exponential backoff
    callbacks?.onRetry?.(attempt, delay, result.error)
    await sleep(delay + jitter())
  }
  callbacks?.onGiveUp?.(policy.maxAttempts, lastError)
  throw new RetryExhausted(lastError)
}

Retryable errors: 429 (rate limit), 529 (overloaded), overloaded_error, network timeout. Non-retryable: 400 (bad params), 401 (auth failure) — retrying won't help.

Why hooks? The TUI needs to show "Retrying (2/3)..."; telemetry needs to record retry counts. Hooks decouple retry logic from display/monitoring.

Constrained Sampling: Making Models Output Compliant JSON

interface ConstrainedSampling {
  mode: "prefer" | "require"
  schema?: JsonSchema        // JSON Schema strict
  grammar?: string           // Lark / regex grammar (OpenAI)
}

• prefer: Hint in system prompt for JSON output, but don't enforce — suits "structured is nice, free text is fine" scenarios
• require: Force compliant JSON via API's response_format / tool_choice — suits tool call parameters

Collaboration with validation.ts: Even in require mode, models occasionally produce non-compliant JSON. validation.ts performs a final check — non-compliance triggers a "correction retry" (feeding the error back to the model for self-correction).

Engineering Insights

1. event-stream.ts ~180 lines: Core is an internal queue + async iterator. Complexity lies in handling the race "events buffered after error" — pi uses a closed flag to discard subsequent pushes after error.

2. closeBrackets doesn't handle arrays: The teaching Demo tracks both [] and {}, but production only closes } — because tool arguments are always JSON Objects, never bare arrays. Simplification is correctness.

3. Division of labor between retry and lazyStream: lazyStream (→ Chapter 7) handles "connection failure," retry handles "response failure." The former is SDK loading/network establishment; the latter is model returning errors. Two independent layers combine to cover all failure paths.

Cross-Chapter Links

Direction	Chapter	Relationship
Prerequisite	Ch01 Agent Loop	agentLoop consumes AssistantMessageEventStream
Prerequisite	Ch07 Wire Protocols	lazyStream produces the stream; this chapter handles errors and retries on it
Follow-up	Ch02 AgentHarness	Harness wraps another "whole-turn retry" above retry
Follow-up	Ch14 Model Runtime	Runtime dynamically downgrades model based on retry results

Check Your Understanding

1. If parsePartialJson fails even after bracket-closing (returns null), how should the TUI behave? Ignore silently or display raw text?
2. Why is 429 retryable but 400 not? What would happen if all errors were retried?
3. Why doesn't constrained-sampling's "prefer" mode simply use response_format to force? In what scenarios could forcing actually be harmful?

Takeaways

• Errors are events, not exceptions — partial message + error flow together; recover, don't catch
• parsePartialJson tolerantly closes brackets, parses incrementally, never throws
• Exponential backoff targets only retryable errors; hooks decouple display from logic
• Constrained sampling prefer/require + validation as final safety net

---

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