Part III Chapter 2: Connecting the First Pipe — ModelClient

Chapter guide: You will write the ModelClient interface and the ModelEvent type in src/types.ts, then translate the Anthropic SDK's events into internal events in src/model-client.ts. At the core is one discipline: business code never touches the SDK's event naming directly.

With the scope settled, Xiao A opened the editor, and the first thing he did was stare blankly at types.ts. He asked Lao Z, "How do we write the model integration? Can't we just call the SDK directly?"

"You can, but you will regret it," Lao Z said. "Isolate the model protocol first. ModelClient.stream() declared in src/types.ts produces the internal ModelEvent; only AnthropicModelClient in src/model-client.ts depends on the official SDK. Business code never reads the SDK's event naming."

Xiao A: "Then what exactly does this middle layer isolate?"

Start with the internal message contract

Lao Z typed out a snippet in types.ts:

// src/types.ts, excerpt
export type ModelEvent =
  | { type: "text"; text: string }
  | { type: "tool_call"; call: ToolCall }
  | { type: "done" };

"Text, tool calls, and completion are split into three kinds because they arrive at different times in the stream," Lao Z said. "done does not carry the conclusion that 'the answer is certainly valid'; the loop still has to check whether it accumulated any text or tool calls. With this small contract in place, business code no longer needs to read the SDK's event naming."

Xiao A pressed: "Then why not just design a single message event that carries text, calls, and completion all at once?"

"Because they arrive at different times, and downstream can consume them at different moments too. Merge them into one event and either you wait for the whole stream to finish before emitting anything — then stream() no longer deserves to be called streaming — or you emit a half-finished product and make the loop guess which part is executable. The three events correspond to three facts that can be consumed independently in time: text can be appended to the display the moment it arrives; tool_call is a complete, executable call the moment it arrives; done means the protocol layer has wrapped up. The granularity of the events is decided by 'what is the earliest downstream can do', not by what the provider's events look like."

Design question	This artifact's choice	The alternative dropped	Why it was dropped
Event granularity	three atomic events	one aggregated message	aggregation means waiting for everything, or shipping a half-finished product
Argument completeness	tool_call emits complete input only	incremental argument fragments	downstream has no way to judge when it is executable
done semantics	only stream wrap-up	"the answer is valid"	validity must be checked by the loop itself
The protocol visible to business code	the three internal events	raw SDK event names	zero changes in the business layer when switching vendors

Lao Z sketched a mapping diagram:

Raw Anthropic SDK events                 Internal ModelEvent
─────────────────────────────            ───────────────────
content_block_delta (text_delta) ───────► { type:"text" }
content_block_start (tool_use)
content_block_delta (input_json_delta)   (accumulated, emits no event)
content_block_stop               ───────► { type:"tool_call" }
message_delta / message_stop     ───────► { type:"done" }

"Notice the middle part: input_json_delta has no corresponding row in the mapping table. It emits no internal event; it only updates internal state. SDK events are the input to a state machine; internal events are what it produces after parsing."

Message mapping is not a type assertion

Xiao A stared at the toolInput function for a long time, then asked: "Tool input is unknown in TypeScript. Why still check it during mapping? Aren't the types already declared?"

"Because both the model and the network can hand you arbitrary JSON." Lao Z opened model-client.ts:

function toolInput(input: unknown): Record<string, unknown> {
  if (input === null || typeof input !== "object" || Array.isArray(input)) {
    throw new Error("tool input must be an object");
  }
  return input as Record<string, unknown>;
}

"The first line keeps the untrusted type; the condition rules out the three kinds of JSON that are not objects; the final assertion happens only after the runtime check," Lao Z said. "It also does not verify whether the fields match a tool schema. That is the tool execution layer's job."

Xiao A: "Then what exactly does this layer guard against, and how far?"

"It guards against 'structural impossibility', not 'business illegality'," Lao Z said. "Tool input must be a JSON object. null, an array, a string are structurally impossible as tool arguments, so this layer rejects structure only: an input like { path: 42 }, whose field type is wrong, passes through and is left for stringField inside execute to reject. Two lines of defence, each covering its own stretch: one asks 'is it an object', the other asks 'are the fields right'. Comparing the real code shows the layering: toolInput in model-client.ts does only the structural check; the first line of read.execute in tools.ts is const path = stringField(input, 'path'). Each layer answers only the question it owns. Shift the questions and you get either duplicated validation or missing validation."

"There is one more detail hidden here: the tool input must be an object thrown by toolInput does not satisfy the startsWith("Anthropic") check in toModelEvents, so it gets wrapped into a protocol error of the form tool_use block N returned non-object input. The error message carries the block index, so when you locate the fault you know which stretch of the stream went wrong."

The state of streamed tool arguments

Xiao A changed his angle: "What about streaming output? Tool arguments arrive in several pieces. How do I know when they are complete?"

"Now we reach the most tangled part of this chapter." Lao Z pointed at toModelEvents, which AnthropicModelClient.stream() delegates to:

// src/model-client.ts, toModelEvents (simplified: the index lookup of activeTool and the error wrapping are omitted)
if (event.type === "content_block_delta" && event.delta.type === "input_json_delta") {
  activeTool.json += event.delta.partial_json;
}
if (event.type === "content_block_stop" && activeTool) {
  input = JSON.parse(activeTool.json) as unknown;
}

"The normal sequence is: the tool block's start creates { index, id, name, startInput, json }; several deltas with the same index append JSON; only the block stop with that same index parses it and produces the internal call. When start already carries a complete object and no delta follows, that object is used directly — that is how {} works for a zero-argument tool. At any earlier moment the delta string may be incomplete, and it must not be handed over for execution."

Xiao A: "What if the stream goes wrong partway through?"

"A parse failure, a non-object input, a JSON delta for an unknown index, or a tool block still open when the stream ends — any of them interrupts that response. The stop of a text block does not close a tool block."

"What if several tools come at once?" Xiao A went on.

"Multiple tool blocks can arrive interleaved, so a single 'current tool' variable cannot associate them: tool calls are emitted in the order of their own stops, but the arguments always come from the start/delta/stop of one and the same index."

"Then what does the lifecycle of a single tool block look like, drawn out?" Xiao A asked.

Lao Z drew a state diagram:

               content_block_start (tool_use)
                                │ create ActiveTool{id,name,startInput,json:"",received:false}
                                ▼
               ┌─────────────────────────────────┐
               │     block with no delta yet     │
               │  (startInput is authoritative)  │
               └─────────────────────────────────┘
                                │
        input_json_delta (same index)
                                │ append json + received=true
                                ▼
               ┌─────────────────────────────────┐
               │   block with deltas received    │
               │     (json is authoritative)     │
               └─────────────────────────────────┘
                                │
                                │ content_block_stop
                                ▼
        JSON.parse(json) → toolInput → yield tool_call → delete from Map

"Two rules hold it together: only the block stop is allowed to parse and emit, and a delta is only ever appended, never parsed. Why not try JSON.parse on every delta? Because partial_json is usually a fragment that has not closed yet — something like {"path":"a" parses at no moment. Trying to parse piece by piece only spits out a pile of noise errors; genuinely bad JSON can be confirmed only at stop time."

Xiao A: "And two tools arriving interleaved?"

Two tool blocks arriving interleaved in one stream (the scenario model-client.test.ts covers)

start(1) → start(3) → delta(3) → delta(1) → stop(1) → stop(3)
   │          │          │          │           │          │
activeTools: {1}        {1,3}      {1,3}      {1,3}       {3}       {}
                                                                     │
Emitted events:                                   tool_call(1)  tool_call(3)  done

"The key is the last row: index 1 and index 3 each accumulate their own JSON, without interfering with each other. With a single 'current tool' variable, the second start would overwrite the first and the first tool's arguments would be lost. That is why toModelEvents uses a Map<index, ActiveTool> instead of one variable — the next section looks at it closely."

toModelEvents is a small state machine

Xiao A: "So how does the real toModelEvents actually handle all of this? Is it enough to write an activeTool variable following the concept?"

"Not enough." Lao Z opened the real model-client.ts and pointed at the middle section: "What it maintains is a Map<index, ActiveTool>, because multiple tool blocks may arrive interleaved:"

// examples/mini-agent/src/model-client.ts, excerpt
const activeTools = new Map<number, ActiveTool>();
let stopReason: Anthropic.StopReason | null | undefined;
let toolCallCount = 0;
let receivedMessageStop = false;

"Every open block is stored under its own index. Look at the full ActiveTool fields:"

type ActiveTool = {
  id: string;
  name: string;
  startInput: unknown;
  json: string;
  receivedInputJsonDelta: boolean;
};

Xiao A pointed at receivedInputJsonDelta and asked: "What is this boolean for?"

"This is the easiest detail to miss," Lao Z said. "The input in a content_block_start event may already carry the complete arguments (say {} for a zero-argument tool), or it may be only the beginning. receivedInputJsonDelta records 'whether any incremental JSON arrived this time': if it did, you need to JSON.parse the accumulated json again; if it did not, startInput is used directly. Look at this part:"

// examples/mini-agent/src/model-client.ts, excerpt
let input: unknown = activeTool.startInput;
if (activeTool.receivedInputJsonDelta) {
  try {
    input = JSON.parse(activeTool.json) as unknown;
  } catch {
    throw protocolError(
      `tool_use block ${event.index} had invalid input_json_delta JSON`,
    );
  }
}

Xiao A picked up the thread: "So once a delta has arrived, json decides?"

"Right. Once a delta exists, json is authoritative: if the accumulation is incomplete or parsing fails, it throws immediately."

Xiao A: "Those bad cases you listed earlier — how does the real code end each of them?"

Lao Z counted the error paths: "An input_json_delta with no open block to match it, a duplicate tool_use start, a JSON parse failure at block stop, a block still unclosed when the stream ends, a missing message_stop, an unsupported stop reason, or a stop reason that contradicts the actual call count. Every one of them does throw protocolError and produces no done."

// examples/mini-agent/src/model-client.ts, excerpt
if (activeTools.size > 0) {
  throw protocolError(
    `stream ended with unclosed tool_use block(s): ${[...activeTools.keys()].join(", ")}`,
  );
}
if (!receivedMessageStop) {
  throw protocolError("stream ended before message_stop");
}

Xiao A glanced over: "All these errors carry the Anthropic stream protocol error: prefix?"

"That way the caller can tell a protocol problem from a business problem at a glance," Lao Z said. "This is fail loud: when the stream state is incomplete, the whole response fails rather than handing half-finished arguments over as a complete call."

Lao Z collected the error paths of this state machine into a table, all carrying the Anthropic stream protocol error: prefix:

Trigger	Error message (prefix omitted)	Why not skip it silently
input_json_delta with no open block to match	input_json_delta arrived without an open tool_use block at index N	the ordering is already broken; staying silent would mix streams
duplicate tool_use start	duplicate tool_use start for block N	two starts at the same index leave the arguments' owner unclear
JSON parse failure at block stop	tool_use block N had invalid input_json_delta JSON	a half-assembled argument string must not pass as a complete call
stream ended with unclosed blocks	stream ended with unclosed tool_use block(s): …	a call without its result leaves the model hanging
no message_stop received	stream ended before message_stop	cannot confirm this round ended whole
missing stop reason	message_delta did not provide a stop_reason	cannot decide whether the loop should continue

"Checked against model-client.test.ts, every row above is watched by an offline test. Protocol errors are caught by tests going red, not by human observation."

Xiao A was a little unconvinced: "A single variable isn't completely unusable, is it? You never run into interleaving in normal use."

Approach	Basis for association	When events arrive interleaved
a single activeTool variable	"the tool currently being assembled"	the second start overwrites the first; the arguments are lost
Map<index, ActiveTool>	the content block index the provider supplies	each index accumulates on its own, without interference

"A single variable is not wrong every time — it is wrong only when events interleave, which is exactly what makes it tempting. But interleaving is a normal case the provider permits, and the test associates simultaneous tool JSON with their content-block indexes watches it specifically."

The stop reason is a precondition for committing

Xiao A spotted the stop_reason field in the log and said offhandedly: "Isn't that just display information?"

"No." Lao Z's tone turned serious. "message_delta.delta.stop_reason is not display information. This adapter currently accepts only two complete meanings: end_turn, which must come with no completed tool call; and tool_use, which must come with at least one completed tool call."

"max_tokens, pause_turn, refusal, stop_sequence, a missing stop reason, and the two accepted reasons when they contradict the actual tool calls all throw an error carrying the provider protocol context, and none of them produces done," Lao Z said. "That way truncated text cannot be mistaken by the downstream loop for a complete answer. If a tool block is still open when message_stop arrives or when the iterator ends naturally, it errors the same way."

stop_reason	Condition accepted	Meaning	The loop's action
end_turn	no completed tool call	the model's answer is complete	commit the text, return complete
tool_use	at least one completed call	the model requests a tool	dispatch the tool, continue the next round
max_tokens	not supported	the output was truncated	throw, emit no done
pause_turn	not supported	the model paused its turn	throw, emit no done
refusal	not supported	the model refused	throw, emit no done
stop_sequence	not supported	a custom stop sequence	throw, emit no done
missing	not supported	the protocol is incomplete	throw, emit no done

Xiao A: "max_tokens is common in real products, isn't it? Is refusing it across the board too conservative?"

"Because max_tokens means the output was cut off: the text stops halfway, or the tool arguments stop halfway through assembly. Committing it as a complete answer misleads; dispatching it as a tool request may hand over arguments that are not executable at all. This artifact would rather fail the whole round than pretend 'truncated' is 'complete'. Supporting it later means first designing semantics like 'keep the half-finished text or not, auto-retry or not, retry how many times'. That is a new requirement, not this chapter's default."

Normal completion, abort, and the evidence boundary

Xiao A remembered something else: "Then what about cancellation? The user pressed stop — what should I do?"

"The abort path has two checkpoints." Lao Z opened agent.ts:

// examples/mini-agent/src/agent.ts, excerpt
for await (const event of options.client.stream(request)) {
  if (aborted(options.signal)) {
    return { messages, stopReason: "aborted" };
  }
  // collect this round's events
}
if (aborted(options.signal)) {
  return { messages, stopReason: "aborted" };
}

"The first check keeps us from going on absorbing events once cancellation is already known; the second sits between stream completion and the assistant commit. These two checkpoints do not provide forced preemption at any moment: if the underlying async iterator never returns and does not respond to the signal, the host still cannot get past the for await it is waiting on."

Xiao A: "So cancellation is... cooperative?"

"Yes. The AbortSignal is passed to the SDK, but if the client does not respond to the signal promptly, the loop can only observe cancellation at its checkpoints — after an event returns, or after the stream ends. Partial text already received is never committed as a complete assistant message."

Lao Z laid out the verification boundary of the whole chain in a table:

Fact	How this round verifies it	Not verified
the pure mapping rejects a null input	fake unit test	what the SDK's server side accepts
tools accumulate by index, start input, block closure	unit tests over constructed offline SDK events	the real Provider's stream ordering
unsupported or missing stop reason never yields done	offline tests constructing max_tokens, a missing reason, and so on	how the Provider actually delivers an abnormal response
the fake round trip works	npm test	the real API, cost, rate limits
the end-of-stream race does not commit an assistant	regression test with a non-cooperative fake stream	the SDK's upper bound on cancellation latency

"Read those two columns together: 'how this round verifies it' is the promise, 'not verified' is the part not yet honoured."

"So where exactly does cancellation land inside the loop?" Xiao A asked.

Lao Z drew the checkpoints of one runAgent turn as a diagram:

Cancellation checkpoints in a single runAgent turn (agent.ts)

turn starts
  │  ① aborted? (before this turn's loop) ──────────yes──► return aborted
  ▼
for await (client.stream(...))
  │  ② aborted? (after each event) ─────────────────yes──► return aborted
  ▼
stream ends
  │  ③ aborted? (before the assistant commit) ──────yes──► return aborted (no commit on the race)
  ▼
commit the assistant (throw when text=="" && calls==0)
  ▼
for call of calls
  │  ④ aborted? (before each tool executes) ────────yes──► record aborted before execution
  │  ⑤ aborted? (after approval)
  ▼
tool.execute(call.input, signal)
  │  ⑥ aborted? (after the tool executes) ──────────yes──► record the rest as not executed
  ▼
next call / next round

"Pay attention to gate ③: the stream has already ended naturally, but the signal happens to be raised exactly as the last piece of it arrives. Skip this check and half-finished text gets committed as a complete assistant message. The abort after a non-cooperative stream completes prevents assistant commit case in cancellation.test.ts was written for this gate alone."

Summary

The key to connecting the model stream is to leave an adapter seam between the vendor protocol and the internal events. ModelClient exposes only a stream() method, taking system, messages, tools, and an optional signal, and producing three internal events: text, tool_call, done. The loop consumes only these three and never needs to know whether Anthropic, OpenAI, or something else is behind it; the whole value of the adapter layer sits in that thin layer of isolation. The event granularity is not arbitrary: text can be displayed the moment it arrives, tool_call means one complete executable call the moment it arrives, and done promises only that the protocol has wrapped up, not that the answer is valid — the latter is for the loop to check itself, by counting the text and the calls.

toModelEvents is a small state machine, and its reason to exist is those crossing paths. Tool arguments accumulate in a Map keyed by the provider's content block index; receivedInputJsonDelta distinguishes "the arguments came complete with start" from "the arguments are assembled out of deltas". Only the block stop parses and emits; half-assembled JSON must never be handed to execution at any moment. Protocol errors uniformly carry the Anthropic stream protocol error: prefix — better a red test, better a failed round, than passing half-finished arguments off as a complete call. The stop reason is equally strict: outside the two complete meanings end_turn and tool_use, max_tokens and refusal throw outright, because truncated text should not be treated as a complete answer.

Cancellation is cooperative. The AbortSignal is passed to the SDK, but the loop observes cancellation only at a few fixed checkpoints: before the stream for each round starts, while the stream runs, between the end of the stream and the assistant commit, and before and after each tool executes. The gate after the stream ends matters most: the signal arrives at exactly the last piece of the stream, and without the check you would commit half-finished text as a complete assistant message. Every checkpoint trades in the same direction — under-commit rather than over-commit — so the signal's propagation path is predictable, and debugging can pin an interruption to one specific gate.

But this seam does not adapt to every vendor by itself. "The adapter runs" only proves that the current vendor's event format is translated correctly. Switching vendors means rewriting the message role conversion, the partial JSON assembly, and the stop semantics mapping, and each of those has to be verified with fake streams plus controlled tests against the real service. State the boundary of this artifact honestly: a real Anthropic end-to-end stream has not been verified, and timeouts, retries, and usage accounting are not implemented either. Treating "the adapter is wired up" as "every vendor works" is the easiest pitfall to fall into in this chapter.

Stage Acceptance

Run npm test and npm run typecheck; do not set up real credentials for this.