fix(proxy/streaming): tolerate malformed upstream SSE events

[nkamali]

Description

Three parsers read upstream-controlled SSE bodies with json.loads and then
reach into the result without checking its shape:
_parse_sse_usage, _parse_sse_usage_from_buffer and _parse_sse_to_response.

A valid-JSON value of the wrong shape passes the json.JSONDecodeError guard
and then raises. Three distinct ways:

• a non-object event (["x"], "str", 42) makes .get raise AttributeError
• an explicit "key": null bypasses a .get(key, {}) default, because the
  default applies only to a missing key, not a present-and-null one
• an unhashable index (a list or dict) raises TypeError: unhashable type
  when used to key the block map

In _parse_sse_to_response the raise escapes into _finalize_stream_response
and tears down a stream the client is already reading. Several call sites reach
it from a bare finally, so there is no handler above it at all.

This is a robustness invariant, not a bug report about any specific upstream.
I am not claiming Anthropic emits these shapes; I have no evidence of that. The
Anthropic handler serves any Anthropic-shaped upstream (--backend anthropic|bedrock|openrouter|anyllm|litellm-<provider>,
ANTHROPIC_TARGET_API_URL), so Bedrock, OpenRouter, LiteLLM, vLLM and Vertex
all flow through these parsers. A parser on a proxy's critical streaming path
should not raise on a well-formed-JSON frame of unexpected shape, whatever the
source. The guard costs one isinstance; being wrong costs the user's session.

Scope: frame shape only. Value validity — a token count that is a string,
Infinity, or a >4300-digit integer literal — is a separate defect class that
reaches the same parsers and is deliberately left for a follow-up.

Type of Change

• Bug fix (non-breaking change that fixes an issue)

Changes Made

The whole change is three one-line normalizers plus their call sites
(docstrings elided here; they are in the diff):

def _sse_dict(value: Any) -> dict[str, Any]:
    return value if isinstance(value, dict) else {}

def _sse_str(value: Any) -> str:
    return value if isinstance(value, str) else ""

def _sse_index(value: Any, default: int | None = None) -> int | None:
    if isinstance(value, bool) or not isinstance(value, int):
        return default
    return value

At a call site, in _parse_sse_usage_from_buffer:

# before
msg = data.get("message", {})      # default fires only if "message" is ABSENT
msg_usage = msg.get("usage", {})   # AttributeError if msg is None / str / list

# after
msg_usage = _sse_dict(_sse_dict(data.get("message")).get("usage"))

Nested because message and message.usage are independently
upstream-controlled — either can be null or the wrong type, and the outer
guard cannot protect the inner access. The .get defaults are dropped as
redundant, since _sse_dict maps absent, null and wrong-type onto one result.

Why a helper and not .get(key) or {}: or branches on truthiness, so it
closes the null case and leaves every truthy non-dict ("overloaded", 42,
["x"]) still reaching .get. isinstance keys off the property the next line
actually requires — is this a mapping? — and it narrows the type for mypy, so
the next unguarded .get added to one of these chains becomes a type error
rather than being inferred as Any.

Well-formed frames are unaffected. Each normalizer is the identity function
on a value of the expected type, so no happy-path value is rewritten. The
real-stream happy-path fixtures pass against origin/main and this branch
alike — they sit inside the 217 of 408 cases that already passed before the fix.

A malformed frame degrades rather than being swallowed. A wrong-shaped
container becomes an empty one of the right type, so the affected field
contributes nothing: _parse_sse_usage* omits that key (or returns None) and
_parse_sse_to_response skips the block. The cost is one lost usage number,
which _finalize_stream_response already handles — it falls back to a
byte-derived estimate and tags the outcome
output_tokens_source=estimated_bytes instead of passing a guess off as
measured. The normalizers themselves are deliberately silent: they sit on a
per-frame, per-field hot path, so even a debug line would fire thousands of
times per stream.

• headroom/proxy/handlers/streaming.py

  • Add three shared normalizers — _sse_dict, _sse_str, _sse_index — so
    every upstream-derived container is shape-checked with one idiom instead of
    a mix of inline isinstance and .get(..., {}) defaults.
  • _sse_index rejects bool as well as non-int, so true cannot silently
    alias block index 1, and unhashable values cannot reach the block map.
  • Guard the positions the earlier pass missed: content_block_delta.delta,
    message_delta.delta, message_delta.usage (dict.update on a non-mapping
    raises — the message_start twin was already guarded, this sibling was not),
    the index on all three content_block_* events, and the string
    accumulators for text / partial_json / thinking.
  • Guard _parse_sse_usage, which had no shape checks at all — not on the
    top-level event, nor on the anthropic, openai or gemini branches.
  • Guard the gemini usageMetadata branch in _parse_sse_usage_from_buffer,
    which used a truthiness check where its two sibling branches used
    isinstance.
  • Stop the non-standard-block copy-through from seeding the parser's own
    scratch keys (_partial_json, thinking_buffer) via _BLOCK_SCRATCH_KEYS,
    and tolerate a non-list citations that a copy-through already placed.

• tests/test_streaming_sse_malformed_events.py — new file, 408 cases.

Testing

• Unit tests pass (pytest)
• Linting passes (ruff check, ruff format)
• Type checking passes (mypy headroom)
• New tests added

The sweep is enumerated, not randomized: the failure surface is finite, so the
cases can be exact and no hypothesis dependency is needed. This matches the
existing precedent in tests/test_toin_observation_only.py.

Happy-path fixtures are transcribed from real api.anthropic.com streams
rather than hand-written, which corrected four wrong assumptions in the
previous fixtures: message_delta.usage repeats the full usage block (not
just output_tokens); message_start.message carries stop_details: null
alongside the other two nulls; usage.cache_creation is always present as a
nested object and is what _extract_anthropic_cache_ttl_metrics reads; and a
tool_use block carries a non-standard caller field.

Test Output

$ pytest tests/test_streaming_sse_malformed_events.py
408 passed

# same file against origin/main's parsers
191 failed, 217 passed

# regression sweep: all 22 test files touching SSE parsing
710 passed, 5 skipped

$ mypy headroom --ignore-missing-imports
Success: no issues found in 509 source files

Real Behavior Proof

• Environment: macOS 15 (Darwin 25.4.0, arm64), Python 3.13.7, Rust 1.95.0
  per rust-toolchain.toml, uv sync --extra dev. Proxy build 0.33.0-dev
  (editable install of this branch), confirmed per run via /health.

• Exact command / steps: enumerated unit sweep, a parser differential against
  origin/main, HTTP-level fault injection through a local Anthropic-shaped
  upstream, then a live Claude Code session on this build.

  # 1. the enumerated sweep
  pytest tests/test_streaming_sse_malformed_events.py -q
  
  # 2. parser differential. origin/main's streaming.py is loaded via
  #    `git show origin/main:...` under a synthetic module name — NOT `git stash`,
  #    which silently no-ops once the fix is committed and yields a false pass.
  #    The same malformed frame goes into _parse_sse_usage,
  #    _parse_sse_usage_from_buffer and _parse_sse_to_response on both versions.
  
  # 3. HTTP-level fault injection. A local server stands in for one of the
  #    Anthropic-shaped upstreams this handler serves and streams a
  #    well-formed-JSON frame of the wrong shape: message_start.message = null.
  ANTHROPIC_TARGET_API_URL=http://127.0.0.1:9099 headroom proxy --port 8799
  curl -N -X POST http://127.0.0.1:8799/v1/messages \
    -H 'content-type: application/json' -H 'anthropic-version: 2023-06-01' \
    -d '{"model":"claude-fake","max_tokens":16,"stream":true,"messages":[{"role":"user","content":"hi"}]}'
  
  # 4. live positive control: a real Claude Code session on this build,
  #    in an unrelated repository
  headroom wrap claude --port 8788
  
  # 5. types, then the full local gate (rust + python + commitlint)
  mypy headroom --ignore-missing-imports
  make ci-precheck

• Observed result: (1) 408 passed. (2) all three parsers raise
  AttributeError: 'NoneType' object has no attribute 'get' on origin/main
  and return None on this branch from identical input. (3) the malformed
  stream completes end to end — HTTP 200, 568 bytes, all six SSE frames
  including the text delta delivered to the client, zero tracebacks, error: null in the request record. Usage degrades partially and correctly:
  output_tokens=7 survives from the well-formed message_delta, while
  cache_read/cache_write — which live inside the nulled
  message_start.message.usage — fall back to 0 instead of raising. (4) 421 live
  requests over 24h (claude-opus-5, claude-sonnet-5) in an unrelated
  repository, error: null on all 421 and zero streaming exceptions in the
  proxy log. Of the 340 that reached the streaming finalizer, 328 carry
  output_tokens_source=provider; 12 fell back to the byte estimate, every one
  on a stream truncated in its first frames — output_tokens = total_bytes // 40
  puts those bodies at 120-919 bytes against inputs of 149k-197k tokens, and that
  fallback is pre-existing and identical on origin/main, not a guard dropping a
  frame. The other 81 requests never reached the streaming finalizer, which sets
  that tag unconditionally, so they are non-streaming. cache_read climbed to
  256,027 with 407/421 cache hits and 120,129 tokens saved. So the guards are
  identity-transparent on well-formed frames, and the sustained prefix-cache
  growth means _parse_sse_to_response ran throughout — all three parsers
  exercised on live traffic. (5) Success: no issues found in 509 source files; ci-precheck PASSED.

• Real upstream capture: three live streaming POST /v1/messages calls to
  api.anthropic.com (claude-haiku-4-5) — a text stream, a tool_use stream
  and an extended-thinking stream — captured as raw wire bytes and used to build
  the happy-path fixtures. This is what surfaced the four fixture errors above.

• Differential: the 408-case file was run against origin/main's parsers via
  git show origin/main:... (not git stash, which silently no-ops once the fix
  is committed and produces a false pass). 191 cases fail there and pass here.

• Exhaustive shape fuzz: 700 generated cases crossing every upstream-derived
  container position in the three parsers with None, "str", 42, 3.5,
  True, False, ["x"], [], {}, {"k":"v"}. Zero raises after the change;
  before it, the positions listed under Changes Made all raise.

• Per-position legitimacy check: each injection position in the sweep was
  confirmed to produce at least one genuine failure against origin/main, so no
  case passes vacuously. This caught four delta-shaped tests that were injecting
  their fault before any content block had opened, where the parser skips the
  body and the guard is never reached.

• Not tested: the malformed frames are synthetic and the incidence claim is
  unproven — limits below.

  • No evidence these shapes occur in the wild. The fault shapes are derived
    from the parsers' own unguarded access positions, not observed in real
    upstream traffic. I searched the tracker for this crash signature and both
    function names and found no report. The claim is survivability, not incidence.
  • The real API cannot be made to emit malformed frames on demand, so the fault
    injection is synthetic; only the happy-path fixtures and the live positive
    control come from real traffic.
  • No live third-party gateway (Bedrock/OpenRouter/LiteLLM/Vertex) exercised,
    despite those being the motivating case. The fault injection used a local
    stand-in reached via ANTHROPIC_TARGET_API_URL, not a real gateway.
  • No WebSocket path, no Windows.
  • The live session is a positive control — it shows the guards do not alter
    well-formed handling. It does not show a guard firing, and cannot: the
    normalizers are silent and api.anthropic.com sends well-formed frames.
  • Value-validity faults are out of scope and still raise; see Additional Notes.

Additional Notes

Two findings left deliberately unfixed, both pre-existing and both outside the
frame-shape scope of this change:

1. Value validity. _extract_anthropic_cache_ttl_metrics calls int() on an
   upstream value with no try; _usage_int does not catch OverflowError, and
   json.loads accepts bare Infinity; except json.JSONDecodeError is too
   narrow for a >4300-digit integer literal (plain ValueError) or deep nesting
   (RecursionError); and the anthropic and gemini branches export non-int
   token values out of a dict[str, int]-annotated function.
2. Reconstruction fidelity. content_block_start copies through unknown
   fields only for non-standard block types, so the real caller field on a
   tool_use block is dropped on reconstruction.
   test_real_tool_use_stream_reconstructs_tool_input pins the current
   behaviour rather than asserting the desired one.

Review Readiness

• I have performed a self-review
• This PR is ready for human review

Checklist

• My code follows the project's style guidelines
• I have performed a self-review of my code
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation — n/a, no
  user-facing surface changed
• My changes generate no new warnings
• I have added tests that prove my fix is effective
• New and existing unit tests pass locally with my changes
• I did not edit CHANGELOG.md

No new dependencies added.

[github-actions]

PR governance

This PR does not yet satisfy the required template fields:

• Missing required section Runtime Rollout Safety.

Please update the PR body, or move the PR back to draft while it is still in progress.

[JerrettDavis]

The shape-hardening approach looks sound: expected dict/string/index values stay identity-preserving, wrong-shaped upstream frames degrade without escaping parser errors, and the reconstruction changes avoid the scratch-key/citations traps called out in the PR. Local verification passed after copying the built Rust extension into the worktree: pytest tests/test_streaming_sse_malformed_events.py -q (408 passed), nearby SSE parser tests (32 passed), pytest tests/test_streaming_sse_malformed_events.py tests/test_proxy/test_anthropic_streaming_ccr_retrieve.py tests/test_openai_responses_context_compaction.py -q (429 passed), and streaming resilience/logger/cache subsets (39 passed). Ruff passed on the touched files.

[nkamali]

@chopratejas @DevanshiVyas mind reviewing please?

[codecov-commenter]

⚠️ Please install the to ensure uploads and comments are reliably processed by Codecov.

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

[JerrettDavis]

Re-reviewed and maintained the latest source against current upstream main. The conflict resolution composes the PR's shape normalizers with main's strict stream-completeness tracking: message/error/block lifecycle flags remain intact, malformed containers and indexes still degrade safely, and stop_sequence/usage reconstruction preserves both sides' contracts. Exact merged-tree verification: 437 malformed-SSE, buffered-SSE, and streaming-CCR tests passed; targeted Ruff and the full merge pre-commit hooks passed. The branch is now mergeable. Only governance checks were scheduled remotely, which is not an author action item. No blocking finding.