Streaming Reconstruction: From a Delta Stream to Structured Messages流式重建：从 delta 流到结构化消息

Some background: I, JollySammy, am a thrifty little squirrel. Whenever I build an agent, I can’t help wondering — could I wire up a cheaper third-party API as the agent’s built-in model (a real one, mind you, not the kind where you ask “which model (A) are you?” and it cheerfully answers “I’m XXX, made by B”)? Building agents this way, I kept tripping over the same handful of problems: streaming output comes in a thousand shapes and needs a sane way to be rebuilt back into context; and some truly infuriating vendors slap an “XYZ-compatible” label on the box while quietly cutting corners behind the curtain — trimming return fields, or outright misbehaving (missing events, duplicated events, out-of-order events). Under those conditions, how do you keep your agent faithfully mapped to the right information and instructions?

First, the essence of it: when a model streams its output, it doesn’t hand you one complete assistant message — it emits it a sliver at a time. The agent harness has to reassemble those slivers into a structured message that can be saved, replayed, continued, and executed as tools.

What the user finally sees is:

{
  "role": "assistant",
  "content": [
    { "type": "text", "text": "Let me read the file first." },
    {
      "type": "tool_use",
      "id": "tool_1",
      "name": "Read",
      "input": { "file_path": "main.py" }
    }
  ]
}

But the streaming API may actually deliver it in dribs and drabs like this:

TEXT_DELTA: "Let me"
TEXT_DELTA: " read the"
TEXT_DELTA: " file first."

TOOL_USE_START: id="tool_1", name="Read"

TOOL_INPUT_DELTA: "{\"file"
TOOL_INPUT_DELTA: "_path\":"
TOOL_INPUT_DELTA: "\"main.py\""
TOOL_INPUT_DELTA: "}"

So if we’re going to build an agent harness, we obviously can’t just print(delta) — we need a streaming reconstructor.

1. Streaming output is a very “dirty” thing

“Streaming output” sounds like the model speaking one character at a time. In an ordinary LLM call, that’s a lovely way to watch the answer land in real time. But drop it into an agent setting and the stream carries more than plain text:

plain text
thinking / reasoning
the signature of the thinking block
tool call start
tool name
fragments of the tool input JSON
message stop
usage token counts

The problem is that at the API layer none of this is natively structured — it arrives as many separate events.

For example, a tool’s arguments end up as:

{ "file_path": "src/query_loop.py", "limit": 100 }

but they might be split into:

"{\"file"
"_path\": \"src/"
"query_loop.py\","
" \"limit\": "
"100}"

and might even be split in the middle of an escape sequence:

"{\"query\": \"hello \\"
"n world\"}"

So you cannot parse the JSON as it arrives. You must accumulate the argument string in full, and only once the tool block is confirmed finished do you json.loads() it in one shot.

2. The reconstructor is really a “state manager”

Picture the reconstructor as a secretary keeping the books. It holds a few ledgers:

accumulated_content: content blocks already finalized

pending_text:      plain text being accumulated
pending_thinking:  thinking being accumulated
pending_signature: thinking signature being accumulated

current_tool: the tool call being accumulated
  - id
  - name
  - input_json_buffer

For instance, the reconstructor in JollySammy’s favourite piece of harness code sets up exactly these few locals.

As long as a content block hasn’t ended, it stays in “pending / current.” Once a clear boundary arrives, it gets archived into accumulated_content. “Archiving” just means turning the temporary buffer into a finalized content block. For example:

pending_text = "Let me read the file first."

when a tool starts, becomes:

{ "type": "text", "text": "Let me read the file first." }

and is appended to accumulated_content.

3. Why blockwise archiving matters

Because text, thinking, and tool calls carry ordering semantics. Say the model’s real output order is:

thinking: "I need to look at the file first"
text:     "Let me read the file first."
tool_use: Read(...)

You must reconstruct it as:

[
  { "type": "thinking", "thinking": "I need to look at the file first", "signature": "..." },
  { "type": "text", "text": "Let me read the file first." },
  { "type": "tool_use", "id": "tool_1", "name": "Read", "input": { "file_path": "main.py" } }
]

not as something that scrambles the model’s true output order. So the rule is: when a TEXT_DELTA arrives while thinking is still accumulating, finalize the thinking first; when a THINKING_DELTA arrives while text is accumulating, finalize the text first; before a tool starts, both text and thinking must be finalized. That way every block boundary lines up with the real switch point in the stream. In fact we should also guarantee that thinking aggregates into a single block rather than one-block-per-delta. The thinking signature (SIGNATURE_DELTA) may also arrive incrementally, and is finalized together with the thinking at the next archiving boundary.

4. A concrete example: text + thinking + tool call

Suppose the model streams:

THINKING_DELTA: "The user wants code analysis, "
THINKING_DELTA: "so I need to read the file first."
SIGNATURE_DELTA: "sig_abc"

TEXT_DELTA: "Let me look at the relevant file."

TOOL_USE_START: id="tool_1", name="Read"
TOOL_INPUT_DELTA: "{\"file_path\":"
TOOL_INPUT_DELTA: "\"jollysammy/nuts/nuts.py\"}"
MESSAGE_STOP

The state machine walks like this. Step 1, thinking arrives:

pending_thinking = "The user wants code analysis, so I need to read the file first."
pending_signature = "sig_abc"

Step 2, text arrives. Because text and thinking are different blocks, flush thinking first, then start accumulating text. Step 3, the tool starts; flush the text first, then create the current tool. Step 4, the argument deltas arrive and fill input_json_buffer. Step 5, message stop — no more content, so finalize the tool. The reconstructed assistant message is:

{
  "role": "assistant",
  "content": [
    { "type": "thinking", "thinking": "The user wants code analysis, so I need to read the file first.", "signature": "sig_abc" },
    { "type": "text", "text": "Let me look at the relevant file." },
    { "type": "tool_use", "id": "tool_1", "name": "Read", "input": { "file_path": "jollysammy/nuts/nuts.py" } }
  ]
}

5. How do OpenAI-style and Anthropic-style streams differ?

Anthropic’s streaming protocol is fairly “block-shaped.” It tells you:

content_block_start
content_block_delta
content_block_delta
content_block_stop

So the protocol tells you exactly when a tool call starts and stops. But the OpenAI Chat Completions style — especially many OpenAI-compatible proxies (which shall remain nameless — no free “praise” from me here) — commonly looks like:

delta.tool_calls[0].function.name = "Read"
delta.tool_calls[0].function.arguments = "{\"file"
delta.tool_calls[0].function.arguments = "_path\":"
delta.tool_calls[0].function.arguments = "\"main.py\"}"
finish_reason = "tool_calls"

There is no separate tool_call_stop event. So the harness must define its own rule for “tool finished.” One rule JollySammy wrote is: the current tool has exactly two end signals:

1. the next TOOL_USE_START arrives
2. MESSAGE_STOP arrives

So if tool B starts while tool A is accumulating, the harness treats A as finished — finalize A, then open B:

on TOOL_USE_START (tool_2):
    finalize tool_1
    parse {"file_path":"main.py"}
    submit Read(main.py) to the tool executor
    start accumulating tool_2

on MESSAGE_STOP:
    finalize tool_2
    parse {"pattern":"TODO","path":"."}

In one sentence: OpenAI has no “tool end” event; the next tool start, or message stop, is the previous tool’s end. So — plenty of the lucky folks who’ve only ever used a single SDK, or a single official-API quota assume tool calls naturally arrive as complete objects. But in production you routinely get a string of half-formed incremental events.

6. Can you execute tools while still streaming?

The textbook SDK approach is:

wait until the whole message has streamed
rebuild all tool calls
then start executing tools

Stable — but on a long trace, astonishingly slow. So what latency optimization can we make? JollySammy’s move: the instant a tool block is finalized, _finalize_tool_block — the routine that closes out an accumulated tool block (parse its buffered input_json_buffer into real JSON, infer a missing name from the schema if needed, then emit a finished tool_use block) — submits it straight to the streaming executor, without waiting for the whole assistant message to finish (submission happens at exactly those two end signals from the previous section).

Say the model wants to call two tools back-to-back:

Read(file_path="main.py")
Grep(pattern="TODO", path=".")

The moment the second tool starts, the first is known to be finished, so Read("main.py") can run right away while the network stream keeps receiving the second tool’s arguments. That gives you overlap:

network keeps streaming the model output
tool executor runs the finished tools at the same time

This is one of the harness’s most important latency wins. But there’s a precondition: it’s only safe for concurrency-safe, approximately-read-only, execute-early-tolerant tools. These are usually fine to run early:

Read
Grep
Glob
LS

But for side-effecting tools like these:

DeleteFile
WriteFile
Bash("rm -rf ...")
SubmitPayment
SendEmail

streaming execution is dangerous: if the model errors in its second half, the request is downgraded and retried, or the message is never accepted, the side effect has already happened and cannot be rolled back.

7. What traps do backstreet proxies hide?

The previous section covered how OpenAI- and Anthropic-style protocols differ. In practice, official APIs are usually tidy, but proxies and OpenAI-compatible services often aren’t. Here are three common traps and how the harness defends against each.

Trap 1: a duplicated TOOL_USE_START. A normal stream should look like:

TOOL_USE_START: id="tool_1", name="Read"
TOOL_INPUT_DELTA: "{\"file_path\":\"main.py\"}"
MESSAGE_STOP

But some proxies gild the lily and emit a second TOOL_USE_START with the same id:

TOOL_USE_START: id="tool_1", name="Read"
TOOL_INPUT_DELTA: "{\"file_path\":\"main.py\"}"

TOOL_USE_START: id="tool_1", name="Read"
MESSAGE_STOP

That second TOOL_USE_START is just a spurious closing signal. Treat it as new and you may run Read twice, or worse, run a write twice. Defense:

if the incoming TOOL_USE_START has the same id as current_tool:
    do not open a new tool slot
    only fill in missing fields, e.g. the name

In other words, a repeated start with the same id is treated as supplementary info for the same tool.

Trap 2: headless fragments — arguments before the start, or a start that never comes. Normally TOOL_USE_START precedes TOOL_INPUT_DELTA. But a non-conforming proxy might just hurl arguments at you with no start:

TOOL_INPUT_DELTA: "{\"file_path\":\"main.py\"}"
MESSAGE_STOP

A rigid state machine would cry “arguments before a tool start?!” and error out. A production harness can’t be that brittle — CC-Py’s approach is to catch the arguments and refuse to drop them:

TOOL_INPUT_DELTA arrives but current_tool is None:
    synthesize a current_tool
    id   = randomly generated
    name = None
    input_json_buffer = this input fragment

If a tool name shows up later, it’s filled in. If it never does, we fall into the next recovery.

Trap 3: the name is lost, only arguments remain. Say the tool call the proxy finally hands you has no name:

{
  "id": "tool_abc",
  "name": null,
  "input": {
    "file_path": "main.py"
  }
}

But a reasonable harness should have registered every tool’s schema:

Read:  required: file_path
Grep:  required: pattern, path
Bash:  required: command
Edit:  required: file_path, old_string, new_string

So it looks at the arguments { "file_path": "main.py" }, finds that only Read’s required fields match, and infers name = "Read"; for {"pattern": "TODO", "path": "."} it might infer Grep. CC-Py’s inference lives in _finalize_tool_block. It isn’t chasing perfection — just best-effort recovery when a proxy drops a field. If several schemas match, or none do, the safe move is to not execute and mark it an invalid tool call.

The engineering philosophy behind all three is one sentence:

do not trust the shape of the upstream stream
the local rebuild must keep its own rules intact

The upstream may be chaotic; the local rebuild must not be.

8. Why does usage accounting also bite?

Streaming usage events are often cumulative values, not increments. One message might report at message_start:

{ "input_tokens": 1000, "cache_read_tokens": 800, "output_tokens": 0 }

and later at message_delta:

{ "input_tokens": 1000, "cache_read_tokens": 800, "output_tokens": 120 }

If you just add them, you get input_tokens = 2000 and cache_read_tokens = 1600 — both wrong. The correct totals are 1000 / 800 / 120. So in principle, within a single message you take the latest non-zero value per field. A delta reporting 0 won’t clobber the non-zero value from start, while output_tokens keeps moving forward. Only across messages do you add: message 1 input 1000 + message 2 input 600 = 1600 total. Get this wrong and your later cost accounting, cache-hit rate, and token-usage analysis all become distorted.

9. Why must thinking and signature be bound together?

In Anthropic’s extended thinking, thinking is not plain text. The server attaches a signature to it:

{ "type": "thinking", "thinking": "I need to read the file, then analyze the error.", "signature": "sig_xxx" }

The signature is the server’s integrity credential for the thinking content. If you later send the conversation back to the API, thinking and signature must be paired exactly as given. You cannot keep the thinking and drop the signature, nor alter the thinking text while reusing the old signature — the API may reject it. So the state machine must archive pending_thinking and pending_signature together as one atomic block.

10. What if a tool already ran but the stream errors later?

That’s the price of streaming execution. Example:

model starts emitting a tool call: Read(file_path="main.py")
harness sees the tool block complete, runs Read immediately.
then the streaming API fails: connection reset / rate limit / server error

Read has already run. For a read-only tool, no big deal. But for writing a file, sending an email, or deleting a file, there’s no rollback. The correct strategy:

• Don’t pretend to roll back side effects that already happened — record them honestly in the log and the message;
• Non-downgrade error path: fill in results for all tool_use blocks and terminate;
• Downgrade-retry path: mark the half-finished assistant message as a tombstone — i.e. flag it as discarded so it’s no longer used as normal context — and discard() the executor: cancel-behavior tasks are cancelled, block-behavior ones are let run to completion;
• Already-completed side effects are not undone.

This is why a production harness distinguishes tool types: concurrency-safe tools can execute mid-stream; side-effecting tools should wait until the message is fully confirmed, or require human confirmation.

11. What if two tool blocks share an id?

Normally each tool call has a unique id (tool_1 → Read, tool_2 → Grep). But a proxy bug can produce two different tools with the same id (tool_1 → Read, tool_1 → Grep). The harness’s dedup logic treats the second as a supplement to the first rather than a new tool — which means one of the tool calls can’t be paired correctly. But this is a deliberate trade-off, because the conversation has one rule that must always hold: every tool_use id must correspond to exactly one tool_result. If you let one id become two different tool calls, sending it back to the API is more likely to 400. So the harness would rather let a local task fail than blow up the message structure:

the proxy already corrupted the ids; the true intent is unknowable.
choose the conservative path: keep the protocol structure closed,
instead of letting the whole request blow up.