Streaming and Mid-Stream Tool Calls

The last chapter wrapped the loop in a command line that holds it to account: a meter that sums every token, a transcript that records every event in order on disk. But the loop it wraps still calls the model the slow way. It sends the request and waits — for the whole response, every token, the tool call and its arguments and the closing summary — and only when the last byte lands does anything happen: the text prints, the tool runs. For a two-sentence reply that wait is nothing. For a model that thinks for thirty seconds before it decides to run a command, the user stares at a blank line the entire time, and the command that could have started running at second three does not start until second thirty.

Streaming closes that gap: the model sends its response as it produces it, token by token, and the loop shows the work as it arrives. The catch is the tool call. Buffered, a tool call is one clean object — a name, a string of arguments — handed over whole. Streamed, it arrives in pieces: the name in one fragment, the arguments dribbled across a dozen more, with no byte that says “this call is complete.” The loop has to stitch those fragments back into a call and know when it is whole — before the turn is over. The question this chapter answers is whether that reassembly is faithful: does the streamed path reconstruct the same message a buffered call would return, and can the loop act on the tool call before the stream has closed?

4.1 Bytes are not messages

The wire does not deliver JSON objects. It delivers bytes, in whatever chunks the network hands over, and a single read can split a line down the middle or carry three lines at once. Before anything can be parsed, the byte stream has to be reframed into the discrete data: events the protocol defines.

export async function* sseData(body: ReadableStream<Uint8Array>): AsyncGenerator<string> {
  const decoder = new TextDecoder();
  const reader = body.getReader();
  let buffer = "";
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) return;
      buffer += decoder.decode(value, { stream: true });
      let newline = buffer.indexOf("\n");
      while (newline !== -1) {
        const line = buffer.slice(0, newline).trimEnd();
        buffer = buffer.slice(newline + 1);
        newline = buffer.indexOf("\n");
        if (!line.startsWith("data:")) continue;
        const payload = line.slice(5).trim();
        if (payload === "[DONE]") return;
        if (payload) yield payload;
      }
    }
  } finally {
    reader.releaseLock();
  }
}

Listing 4.1 · framing the byte stream. Each read is appended to a buffer, complete lines are cut off at every newline, and only data: payloads survive.

The load-bearing line is buffer += decoder.decode(value, { stream: true }). Each read is appended to what came before, not treated as a message on its own, because a read boundary and a line boundary have nothing to do with each other — a chunk can end in the middle of {"command": with the rest arriving next read. The stream: true flag tells the decoder the same thing about bytes: a multi-byte character split across two reads is held, not mangled. Only once a full \n is in the buffer is a line cut out, trimmed, and checked. Lines that are not data: are dropped, the sentinel [DONE] ends the generator, and every real payload is yielded as a string. What comes out is a clean sequence of JSON payloads — one model chunk each — with the byte-level mess already absorbed.

4.2 Reassembling a tool call from fragments

Each payload is a chunk, and a chunk carries a delta — a scrap of content, or a scrap of a tool call. A tool call’s name arrives in one fragment; its arguments arrive as raw string pieces across the chunks that follow. Something has to accumulate those pieces into one call, and — because the stream never announces it — infer the moment the call is finished. That something is the active accumulator and its finish helper.

  let content = "";
  const usage: Usage = { promptTokens: 0, completionTokens: 0 };
  const calls: ToolCall[] = [];
  let active: ToolCall | null = null;
  let activeIndex = -1;

  const finish = (): ToolCall | null => {
    if (active === null) return null;
    const done = active;
    calls.push(done);
    active = null;
    return done;
  };

  for await (const payload of sseData(response.body)) {
    const chunk = JSON.parse(payload) as WireChunk;
    if (chunk.usage) {
      usage.promptTokens = chunk.usage.prompt_tokens ?? 0;
      usage.completionTokens = chunk.usage.completion_tokens ?? 0;
    }
    const choice = chunk.choices?.[0];
    const delta = choice?.delta;
    if (choice?.finish_reason) {
      const flushed = finish();
      if (flushed !== null) yield { type: "tool-call", call: flushed };
    }
    if (delta === undefined) continue;
    if (delta.content) {
      // a content delta means the active call's fragments are over
      const flushed = finish();
      if (flushed !== null) yield { type: "tool-call", call: flushed };
      content += delta.content;
      yield { type: "text", piece: delta.content };
    }
    for (const fragment of delta.tool_calls ?? []) {
      const index = fragment.index ?? 0;
      if (index !== activeIndex) {
        const done = finish();
        if (done !== null) yield { type: "tool-call", call: done };
        active = { id: fragment.id ?? `call_${index}`, name: "", arguments: "" };
        activeIndex = index;
      }
      if (active === null) continue;
      if (fragment.id) active.id = fragment.id;
      if (fragment.function?.name) active.name = fragment.function.name;
      if (fragment.function?.arguments) active.arguments += fragment.function.arguments;
    }
  }
  const done = finish();
  if (done !== null) yield { type: "tool-call", call: done };

  const message: Message = { role: "assistant", content };
  if (calls.length) message.toolCalls = calls;
  yield { type: "done", message, usage };
}

Listing 4.2 · the streaming state machine. One active call collects fragments; finish flushes it the instant the next thing arrives.

The line that does the stitching is active.arguments += fragment.function.arguments — every argument fragment is concatenated onto the running string, in order, so ten pieces become one JSON blob. The subtler half is when the call is declared complete, because no fragment is marked “last.” Three signals end it, and each calls finish: a content delta arrives — the comment says it plainly, “a content delta means the active call’s fragments are over” — or the index changes, meaning a second tool call has started, or the chunk carries a finish_reason. Whichever fires first flushes active and yields it as a tool-call event, while the stream is still open. Only after the loop drains does the final finish catch a call that ran to the very end. The reassembled call and the accumulated content are then handed over together in the closing done event, alongside the usage — the same shape a buffered call returns, rebuilt from fragments.

4.3 Acting before the turn ends

A reassembled call is only worth streaming if the loop can act on it early. The buffered turn waits for the whole message, then dispatches. The streamed turn consumes the events as they come.

  const events = provider.stream(messages, tools, maxTokens);
  for await (const event of events) {
    switch (event.type) {
      case "text": {
        emit({ kind: "text-piece", piece: event.piece });
        break;
      }
      case "tool-call": {
        emit({ kind: "tool-call", name: event.call.name, arguments: event.call.arguments });
        const result = await gatedDispatch(event.call, context, options, session, tools);
        emit({ kind: "tool-result", name: event.call.name, result });
        results.push({
          callId: event.call.id,
          name: event.call.name,
          output: result.output,
          error: result.error,
        });
        break;
      }
      case "done": {
        assistant = event.message;
        usage = event.usage;
        emit({ kind: "usage", usage: event.usage });
        break;
      }
    }
  }

Listing 4.3 · consuming the stream. Text pieces are emitted as they arrive; a tool call is dispatched the moment it is parsed.

The line that earns the whole chapter is const result = await gatedDispatch(event.call, context, options, session, tools), inside the tool-call case. When a tool-call event arrives, the tool runs then — not after the stream closes — so a command the model decided on early begins executing while the model is still emitting its trailing summary. The text case fans each piece out as a text-piece event for the terminal to print live. The done case captures the final message and usage. One honest detail keeps the transcript straight: the tool runs mid-stream, but its result message is collected into results and only pushed onto the message list and appended to the session after done lands. Execution is early; the bookkeeping stays in the order the loop appends it, so the record the last chapter built reads the same whether the turn streamed or not.

4.4 Correct: streamed equals buffered

Two things have to hold for streaming to be a free win rather than a source of drift: the reassembled call must be byte-for-byte what a buffered call would have returned, and the call must be detected before the turn is done. The companion driver checks both against one fixture, with no network and no model. It stubs the transport to answer the same content two ways — buffered, as one whole message, and streamed, as a tool call fragmented across ten pieces followed by trailing text — then compares what chat returns against what chatStream reassembles:

The streamed arguments reassemble to {"command":"echo measured"}, character for character the string the buffered call returns. The tool-call event lands at event 0 of 3, with a text piece still arriving after it and the done event only at index 2 — the call was in hand before the turn finished. Content and usage match across both paths. What this does not prove is that a real provider fragments a call exactly this way, or that two concurrent calls interleave correctly — the fixture streams a single call. It proves the narrower, load-bearing thing: given a stream of fragments, hoist’s reassembly returns the same message a buffered decode of the same content would, and surfaces the call mid-stream.

The streamed turn records the same assistant message, the same tool results, and the same usage into the SessionLog as the buffered turn — the transcript from the last chapter comes out identical no matter which path produced it. That record is about to carry more weight than a log you read after the fact. Because every event is captured in order, on disk, a run that dies halfway — the process killed, the machine rebooted — can be picked up from the log alone and continued as if it never stopped. That is the next box: an event-sourced session you can kill mid-run and resume from the transcript, with the loop none the wiser.