Migration Diary: Normalize Stream Events Before You Leave a Paid Agent Gateway

You should not cut a paid agent gateway until every client can consume a vendor-neutral stream. The model answer can look correct while leftover SSE shapes, partial tool deltas, and silent usage frames still break your UI. Capture a replay tape first, then normalize events, then dual-run the same fixture on a free runtime before you drain the paid path. This diary treats the stream contract as leftover infrastructure rather than a cosmetic logging concern you can fix later.

The leftover is the event grammar, not the paragraph

Paid agent gateways rarely emit a plain token stream that your existing client can treat as opaque text. They interleave text deltas, tool-call fragments, heartbeat comments, citation objects, and a final usage block that your client already parses. When you move the same prompt onto another runtime, the text can match while the event grammar changes under the client. You then spend the cutover week chasing "random" spinner stalls that are really missing done events.

You do not need a new agent framework to survive that cutover if you already own the client parser. You need a frozen tape, a tiny schema, and tests that fail when an unknown frame appears. The current industry chatter about agents being fancy loops over tools does not cancel this work. A loop still has to stream, and your frontend still has to render partial tool names without crashing.

Freeze four artifacts before you touch DNS or API keys

Do not start the migration by swapping base URLs or by rewriting the production streaming client. Freeze these four artifacts so the paid path and the free path can be compared without storytelling.

1. Capture a raw SSE tape from the paid gateway for three representative prompts you already ship.
2. Freeze a canonical event schema with a closed set of types your clients already understand.
3. Ship a normalizer that maps vendor frames onto that schema and rejects the rest.
4. Run a dual-run harness that replays the same prompts against the destination runtime.

You should store the tapes next to the schema, not in a personal Downloads folder. If the vendor frame names drift during your notice period, the tape is the only honest regression surface you will have. Commit the tapes with redacted headers so every pull request can replay the leftover frames.

Step 1: Record a paid-gateway tape you can replay

Run the client against the paid gateway with stream logging enabled, and write every frame including comments. Keep the HTTP status, headers, and body bytes, because some gateways hide heartbeats as SSE comments rather than JSON. Label each tape with the prompt id, tool list hash, and whether the run used tools.

Here is a small recorder you can drop in front of any SSE endpoint you already call. Treat the recorder as a proposal you should adapt rather than as a production proxy. Redact authorization headers before the tape file lands in git or in a shared fixture bucket.

# stream_tape.py — proposal: record SSE bytes for later normalization tests
from __future__ import annotations

import json
import time
from pathlib import Path

import httpx


def record_sse_tape(
    url: str,
    payload: dict,
    headers: dict,
    out_path: Path,
) -> Path:
    started = time.time()
    frames: list[dict] = []
    with httpx.Client(timeout=120.0) as client:
        with client.stream("POST", url, json=payload, headers=headers) as resp:
            resp.raise_for_status()
            event_name = "message"
            data_lines: list[str] = []
            for raw_line in resp.iter_lines():
                if raw_line == "":
                    frames.append(
                        {
                            "t": round(time.time() - started, 4),
                            "event": event_name,
                            "data": "\n".join(data_lines),
                        }
                    )
                    event_name = "message"
                    data_lines = []
                    continue
                if raw_line.startswith("event:"):
                    event_name = raw_line[6:].strip()
                    continue
                if raw_line.startswith("data:"):
                    data_lines.append(raw_line[5:].lstrip())
                    continue
                if raw_line.startswith(":"):
                    frames.append(
                        {
                            "t": round(time.time() - started, 4),
                            "event": "comment",
                            "data": raw_line[1:].lstrip(),
                        }
                    )
    out_path.write_text(json.dumps({"url": url, "frames": frames}, indent=2))
    return out_path

Record one tape that streams a tool call across several deltas and one tape that is text only. Record a third tape that errors after a partial answer so the client can show a stable failure. Those three shapes cover most leftover bugs you will actually meet on cutover night.

Step 2: Declare a canonical event schema your clients already speak

Do not invent a grand agent protocol just because the destination runtime uses different frame names. List the events your UI and your tool dispatcher already handle, then refuse everything else. A closed schema is the cutover valve: paid frames map into it, and destination frames must map into it too.

# canonical.py — proposal: closed event set for cutover
from __future__ import annotations

from typing import Literal, TypedDict


class CanonicalEvent(TypedDict):
    type: Literal[
        "text_delta",
        "tool_call_delta",
        "tool_call_finish",
        "error",
        "usage",
        "done",
    ]
    request_id: str
    text: str
    tool_name: str
    tool_args_delta: str
    tool_call_id: str
    error_code: str
    input_tokens: int
    output_tokens: int

You should keep the request_id field mandatory even if the destination runtime never emits one. You can mint a client-side id because leftover log joins and support tickets still key off that field. Treat empty strings and zeros as acceptable placeholders, but let missing keys fail the normalizer. That single rule keeps paid tapes and shadow tapes joinable in the same support query.

Step 3: Normalize vendor frames and fail on unknown leftovers

Write the mapper as pure functions over tape frames so you can unit-test without the network. Unknown event names, leftover citation objects, and vendor-only safety labels should raise in staging builds. In production you may log and skip, but the dual-run harness should fail closed so leftovers cannot hide.

# normalize.py — proposal: map SSE frames to CanonicalEvent
from __future__ import annotations

import json
from typing import Iterable

from canonical import CanonicalEvent


class LeftoverFrameError(ValueError):
    pass


def _base(request_id: str) -> CanonicalEvent:
    return CanonicalEvent(
        type="text_delta",
        request_id=request_id,
        text="",
        tool_name="",
        tool_args_delta="",
        tool_call_id="",
        error_code="",
        input_tokens=0,
        output_tokens=0,
    )


def normalize_frame(frame: dict, request_id: str) -> list[CanonicalEvent]:
    if frame.get("event") == "comment":
        return []  # heartbeats are leftovers; do not forward them
    raw = frame.get("data") or ""
    if raw == "[DONE]":
        event = _base(request_id)
        event["type"] = "done"
        return [event]
    try:
        body = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise LeftoverFrameError(f"non-json frame: {raw[:80]}") from exc

    if "choices" in body:
        return _from_openai_shape(body, request_id)
    if body.get("type") in {"content_block_delta", "message_stop", "message_delta"}:
        return _from_anthropic_shape(body, request_id)
    raise LeftoverFrameError(f"unmapped leftover: {sorted(body.keys())}")


def normalize_tape(frames: Iterable[dict], request_id: str) -> list[CanonicalEvent]:
    out: list[CanonicalEvent] = []
    for frame in frames:
        out.extend(normalize_frame(frame, request_id))
    if not out or out[-1]["type"] != "done":
        raise LeftoverFrameError("tape ended without a canonical done event")
    return out

You should implement the vendor mappers against your real tapes rather than against documentation screenshots. Vendor docs omit the frames that appear only when a tool name is streamed one character at a time. Those partial tool names are the exact frames that freeze a naive streaming UI during cutover.

Add a focused test that replays the captured tool-call tape and checks argument reassembly. Keep the fixture path boring so a missing tape fails CI the same way a mapper bug fails CI. Label the test as unexecuted until the three paid tapes actually exist in the repo.

# test_normalize.py — proposal: fail if leftover frames appear
import json
from pathlib import Path

import pytest

from normalize import LeftoverFrameError, normalize_tape


def test_tool_call_tape_emits_deltas_then_done():
    tape = json.loads(Path("tapes/paid_tool_call.json").read_text())
    events = normalize_tape(tape["frames"], request_id="req_paid_001")
    types = [e["type"] for e in events]
    assert "tool_call_delta" in types
    assert types[-1] == "done"
    joined = "".join(e["tool_args_delta"] for e in events)
    json.loads(joined)  # partial args must reassemble


def test_unknown_vendor_frame_is_a_leftover():
    frames = [{"event": "message", "data": "{\"safety_ribbon\": true}"}]
    with pytest.raises(LeftoverFrameError):
        normalize_tape(frames, request_id="req_paid_002")

Run the tests from the repo root with a plain pytest invocation so the tape stays in source control. Keep secrets out of the tape by redacting headers in the recorder before you commit. If pytest cannot find the tape, the cutover is not ready, even when live demos look fine.

pytest test_normalize.py -q

Step 4: Dual-run the same prompts on a free model path

Keep the paid gateway as the source of truth until the normalizer emits identical canonical sequences for the destination. Identical text is not enough; event types, tool argument assembly, and the terminal done event must match. Where tokenizers differ, compare usage with a tolerance, and never block cutover on exact token counts.

A second runtime earns its keep in this window because you need a shadow stream, not a new vendor story. Disclosure: This article was prepared as part of MonkeyCode's product outreach. You can point the harness at MonkeyCode's free model access and free server option as a shadow destination. Store those shadow frames beside the paid tape so the mapper sees both grammars in one fixture. The product is relevant only as a second place to replay the fixture you already own.

# dual_run.py — proposal: compare canonical streams from two runtimes
from __future__ import annotations

from normalize import normalize_tape


def event_signature(events: list[dict]) -> list[tuple]:
    return [
        (e["type"], e["tool_name"], bool(e["text"]), bool(e["tool_args_delta"]))
        for e in events
    ]


def assert_shadow_compatible(paid_frames, dest_frames, request_id: str) -> None:
    paid = normalize_tape(paid_frames, request_id)
    dest = normalize_tape(dest_frames, f"{request_id}_shadow")
    if event_signature(paid) != event_signature(dest):
        raise AssertionError(
            f"shadow stream grammar drifted: {event_signature(paid)} vs "
            f"{event_signature(dest)}"
        )

If the destination cannot stream tools yet, you should not pretend that text deltas are enough. Park those prompts on the paid path, and cut only the text-only routes that already match. Partial cutovers that follow event signatures stay safer than one big-bang base URL swap.

Step 5: Drain leftovers that will not survive the paid contract

After the grammar matches, you still need to walk leftovers that are not stream events. Paid gateways often inject request ids, retry-after hints, citation bundles, and prompt-cache hashes that your operators still mention in runbooks. Write each leftover down, then assign an owner and a kill date before traffic moves.

1. Replace vendor request ids with client-minted ids in logs, traces, and support macros before you drain traffic.
2. Drop SSE comments that only existed to keep a vendor load balancer from closing the socket.
3. Stop parsing citation or retrieval objects unless you rebuilt that index on the destination side.
4. Treat prompt-cache hashes as vendor memory, and do not expect them to appear on the free path.
5. Rehearse a mid-stream disconnect, because the destination will not share the old idle timer.

The leftover that bites most teams is the spinner that waits for a usage frame. If your UI treats usage as the real done, a destination that omits usage will spin forever even though done already arrived. Flip that spinner dependency in the streaming client before you lower any paid traffic.

Decision table for the night you cut

Use this table on the night you cut instead of debating whether the answer looks fine. Each row is a leftover signal you can observe from the dual-run harness without reading the prose. If a prompt class does not fit a row, it stays on the paid gateway until you add one.

Signal on the shadow path	Action	Do not do
Canonical types match, text close enough	Cut that prompt class	Wait for token counts to match
Tool args reassemble, names differ	Freeze tool names in your dispatcher	Let the model alias tools
Unknown frame in destination tape	Keep paid path; extend the mapper	Strip the frame in the UI
No done event	Keep paid path; fix the adapter	Time out and show partial text
Usage missing but done present	Cut after the UI ignores usage	Block cutover for billing fields

Print the table next to the deploy checklist so the on-call person does not improvise. Arguments about polished paragraphs should lose to a matching row in that table. Keep the table in git beside the tapes so the next destination inherits the same valve.

Limitations, and who should skip this diary

This workflow assumes you own a streaming client and you can record production-like prompts without leaking secrets. If your product only uses unary completions, a stream normalizer is ceremony, and you should instead snapshot request and response JSON. If you cannot dual-run because of data-handling rules, stay on one runtime until you have a redacted fixture set.

The mapper above is labeled as a proposal, not as a certified adapter for every vendor. It does not claim compatibility with every vendor frame, and it does not measure latency, cost, or quality. Token counts will drift across tokenizers, so you should meter those separately rather than treating them as stream correctness. Teams that need vendor citations, built-in file search, or hosted memory should not treat a free model path as a full substitute.

What you should have in git after the cutover

You should have three tapes, one schema, one normalizer, and a dual-run assertion that still runs on pull requests. The paid gateway can then go quiet without taking your event grammar with it into shutdown. If a later destination appears, you record a new tape and extend the mapper instead of rewriting the UI. When those three tapes pass on a free runtime, you can cancel the paid stream path with a lot less theater.