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

## 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

- [x] Bug fix (non-breaking change that fixes an issue)

## Changes Made

- **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

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check`, `ruff format`)
- [x] Type checking passes (`mypy headroom`)
- [x] 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

```text
$ 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`.
- **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:**
  - **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 come from live traffic.
  - No live third-party gateway (Bedrock/OpenRouter/LiteLLM/Vertex) exercised,
    despite those being the motivating case.
  - No WebSocket path, no Windows, not run against a live Claude Code session.
  - 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

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] 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
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective
- [x] New and existing unit tests pass locally with my changes
- [x] I did **not** edit `CHANGELOG.md`

No new dependencies added.

Author: nkamali

headroom/proxy/handlers/streaming.py

@@ -30,6 +30,49 @@
 logger = logging.getLogger("headroom.proxy")
 
 
+# Upstream controls every SSE event body, so the shape of anything reached
+# through `json.loads` is not guaranteed. These normalize a wrong-shaped value
+# to a harmless one instead of letting it raise on the streaming path, where
+# the raise escapes into `_finalize_stream_response` and tears down a stream
+# the client is already reading. They guard *shape*, not value validity.
+
+
+def _sse_dict(value: Any) -> dict[str, Any]:
+    """Upstream-supplied object, or ``{}`` when it is not one.
+
+    An explicit ``"key": null`` bypasses a ``.get(key, {})`` default — the
+    default only applies to a *missing* key — so nested objects are
+    type-checked rather than defaulted.
+    """
+    return value if isinstance(value, dict) else {}
+
+
+def _sse_str(value: Any) -> str:
+    """Upstream-supplied text, or ``""`` when it is not a string.
+
+    Accumulating a non-string delta would raise ``TypeError`` on ``+``.
+    """
+    return value if isinstance(value, str) else ""
+
+
+def _sse_index(value: Any, default: int | None = None) -> int | None:
+    """Upstream-supplied block index, or ``default`` when unusable as a key.
+
+    ``index`` keys ``blocks_by_index`` and ``appended_block_keys``; a list or
+    dict there raises ``TypeError: unhashable type``. ``bool`` is rejected too
+    so ``true`` cannot silently alias index ``1``.
+    """
+    if isinstance(value, bool) or not isinstance(value, int):
+        return default
+    return value
+
+
+# Keys `_parse_sse_to_response` owns on its in-progress block dicts. Upstream
+# must not seed them via the non-standard-block copy-through, or a non-string
+# would reach the delta accumulators. `type` is set explicitly, not copied.
+_BLOCK_SCRATCH_KEYS = frozenset({"type", "_partial_json", "thinking_buffer"})
+
+
 def _parse_completion_tokens_from_sse_chunk(chunk_bytes: bytes) -> int | None:
     """Extract `usage.completion_tokens` from a single SSE chunk if present.
 
@@ -179,6 +222,9 @@ def _parse_sse_usage(self, chunk: bytes, provider: str) -> dict[str, int] | None
                 except json.JSONDecodeError:
                     continue
 
+                if not isinstance(data, dict):
+                    continue
+
                 usage = {}
 
                 if provider == "anthropic":
@@ -187,8 +233,7 @@ def _parse_sse_usage(self, chunk: bytes, provider: str) -> dict[str, int] | None
                     event_type = data.get("type", "")
 
                     if event_type == "message_start":
-                        msg = data.get("message", {})
-                        msg_usage = msg.get("usage", {})
+                        msg_usage = _sse_dict(_sse_dict(data.get("message")).get("usage"))
                         if msg_usage:
                             usage["input_tokens"] = msg_usage.get("input_tokens", 0)
                             usage["cache_read_input_tokens"] = msg_usage.get(
@@ -204,24 +249,24 @@ def _parse_sse_usage(self, chunk: bytes, provider: str) -> dict[str, int] | None
                             usage["cache_creation_ephemeral_1h_input_tokens"] = cache_write_1h
 
                     elif event_type == "message_delta":
-                        delta_usage = data.get("usage", {})
+                        delta_usage = _sse_dict(data.get("usage"))
                         if delta_usage:
                             usage["output_tokens"] = delta_usage.get("output_tokens", 0)
 
                 elif provider == "openai":
                     # OpenAI sends usage in final chunk (when stream_options.include_usage=true)
-                    chunk_usage = data.get("usage")
+                    chunk_usage = _sse_dict(data.get("usage"))
                     if chunk_usage:
                         usage["input_tokens"] = chunk_usage.get("prompt_tokens", 0)
                         usage["output_tokens"] = chunk_usage.get("completion_tokens", 0)
                         # OpenAI has cached tokens in prompt_tokens_details
-                        details = chunk_usage.get("prompt_tokens_details") or {}
+                        details = _sse_dict(chunk_usage.get("prompt_tokens_details"))
                         usage["cache_read_input_tokens"] = details.get("cached_tokens", 0)
 
                 elif provider == "gemini":
                     # Gemini sends usageMetadata in each streaming chunk
                     # Format: {"usageMetadata": {"promptTokenCount": N, "candidatesTokenCount": M}}
-                    usage_meta = data.get("usageMetadata")
+                    usage_meta = _sse_dict(data.get("usageMetadata"))
                     if usage_meta:
                         usage["input_tokens"] = usage_meta.get("promptTokenCount", 0)
                         usage["output_tokens"] = usage_meta.get("candidatesTokenCount", 0)
@@ -275,11 +320,18 @@ def _parse_sse_usage_from_buffer(
             except json.JSONDecodeError:
                 continue
 
+            # The upstream controls the event body. A valid-JSON non-object
+            # (a bare array or string, e.g. from an overloaded frontend's
+            # error page framed as SSE) makes `.get` raise AttributeError,
+            # which the JSONDecodeError guard above does not catch. Skip it
+            # like any other unusable event.
+            if not isinstance(data, dict):
+                continue
+
             if provider == "anthropic":
                 event_type = data.get("type", "")
                 if event_type == "message_start":
-                    msg = data.get("message", {})
-                    msg_usage = msg.get("usage", {})
+                    msg_usage = _sse_dict(_sse_dict(data.get("message")).get("usage"))
                     if msg_usage:
                         usage_found["input_tokens"] = msg_usage.get("input_tokens", 0)
                         usage_found["cache_read_input_tokens"] = msg_usage.get(
@@ -299,7 +351,7 @@ def _parse_sse_usage_from_buffer(
                             f"cache_write={usage_found.get('cache_creation_input_tokens')}"
                         )
                 elif event_type == "message_delta":
-                    delta_usage = data.get("usage", {})
+                    delta_usage = _sse_dict(data.get("usage"))
                     if delta_usage:
                         usage_found["output_tokens"] = delta_usage.get("output_tokens", 0)
 
@@ -339,7 +391,7 @@ def _usage_int(value: Any) -> int:
                         )
 
             elif provider == "gemini":
-                usage_meta = data.get("usageMetadata")
+                usage_meta = _sse_dict(data.get("usageMetadata"))
                 if usage_meta:
                     usage_found["input_tokens"] = usage_meta.get("promptTokenCount", 0)
                     usage_found["output_tokens"] = usage_meta.get("candidatesTokenCount", 0)
@@ -402,29 +454,36 @@ def _parse_sse_to_response(self, sse_data: str, provider: str) -> dict[str, Any]
             except json.JSONDecodeError:
                 continue
 
+            # Same upstream-controlled shape guard as
+            # ``_parse_sse_usage_from_buffer``: a valid-JSON non-object event
+            # would raise AttributeError below, and here that escapes into
+            # ``_finalize_stream_response`` and tears down a stream the client
+            # was already reading.
+            if not isinstance(data, dict):
+                continue
+
             event_type = data.get("type", "")
 
             if event_type == "message_start":
-                msg = data.get("message", {})
+                msg = _sse_dict(data.get("message"))
                 response["id"] = msg.get("id")
                 response["model"] = msg.get("model")
                 response["role"] = msg.get("role", "assistant")
                 response["stop_reason"] = msg.get("stop_reason")
                 if "stop_details" in msg:
                     response["stop_details"] = msg["stop_details"]
-                if msg.get("usage"):
-                    response["usage"].update(msg["usage"])
+                response["usage"].update(_sse_dict(msg.get("usage")))
 
             elif event_type == "content_block_start":
-                block = data.get("content_block", {})
-                block_index = data.get("index", len(response["content"]))
+                block = _sse_dict(data.get("content_block"))
+                block_index = _sse_index(data.get("index"), len(response["content"]))
                 btype = block.get("type")
                 current_block = {
                     "type": btype,
                     "index": block_index,
                 }
                 if btype == "text":
-                    current_block["text"] = block.get("text", "")
+                    current_block["text"] = _sse_str(block.get("text"))
                 elif btype == "tool_use":
                     current_block["id"] = block.get("id")
                     current_block["name"] = block.get("name")
@@ -433,7 +492,7 @@ def _parse_sse_to_response(self, sse_data: str, provider: str) -> dict[str, Any]
                     # Thinking block — accumulate text via
                     # `thinking_delta`; signature arrives via
                     # `signature_delta` (single value, not accumulated).
-                    current_block["thinking_buffer"] = block.get("thinking", "")
+                    current_block["thinking_buffer"] = _sse_str(block.get("thinking"))
                     if "signature" in block:
                         current_block["signature"] = block["signature"]
                 elif btype == "redacted_thinking":
@@ -450,31 +509,33 @@ def _parse_sse_to_response(self, sse_data: str, provider: str) -> dict[str, Any]
                     # {type, index}. Mirrors the sibling reconstructor
                     # `_reconstruct_anthropic_response`, which does `dict(block)`.
                     for _k, _v in block.items():
-                        if _k != "type":
+                        if _k not in _BLOCK_SCRATCH_KEYS:
                             current_block[_k] = _v
                 blocks_by_index[block_index] = current_block
 
             elif event_type == "content_block_delta":
                 # Resolve the target block by index (preferred) or fall
                 # back to current_block for legacy linear streams.
-                idx = data.get("index")
+                idx = _sse_index(data.get("index"))
                 target = (blocks_by_index.get(idx) if idx is not None else None) or current_block
                 if target is not None:
-                    delta = data.get("delta", {})
+                    # `"delta": null` bypasses a `.get(..., {})` default the same
+                    # way `"message": null` does on message_start.
+                    delta = _sse_dict(data.get("delta"))
                     dtype = delta.get("type")
                     if dtype == "text_delta":
-                        target["text"] = target.get("text", "") + delta.get("text", "")
+                        target["text"] = _sse_str(target.get("text")) + _sse_str(delta.get("text"))
                     elif dtype == "input_json_delta":
                         # Accumulate partial JSON for tool input.
-                        partial = delta.get("partial_json", "")
-                        target["_partial_json"] = target.get("_partial_json", "") + partial
+                        partial = _sse_str(delta.get("partial_json"))
+                        target["_partial_json"] = _sse_str(target.get("_partial_json")) + partial
                     elif dtype == "thinking_delta":
                         # Accumulate thinking text into the dedicated
                         # buffer so it never collides with `text` on
                         # text blocks (separate field per guide §2.7).
-                        target["thinking_buffer"] = target.get("thinking_buffer", "") + delta.get(
-                            "thinking", ""
-                        )
+                        target["thinking_buffer"] = _sse_str(
+                            target.get("thinking_buffer")
+                        ) + _sse_str(delta.get("thinking"))
                     elif dtype == "signature_delta":
                         # Single value, not accumulated. Last-write
                         # wins per Anthropic spec.
@@ -485,13 +546,18 @@ def _parse_sse_to_response(self, sse_data: str, provider: str) -> dict[str, Any]
                         # list so multi-citation blocks reconstruct
                         # correctly. Per guide §2.5: each delta carries
                         # one full citation object under `citation`.
-                        citations = target.setdefault("citations", [])
+                        # A non-standard block start can copy a non-list
+                        # `citations` through, which `.append` would reject.
+                        citations = target.get("citations")
+                        if not isinstance(citations, list):
+                            citations = []
+                            target["citations"] = citations
                         citation = delta.get("citation")
                         if citation is not None:
                             citations.append(citation)
 
             elif event_type == "content_block_stop":
-                idx = data.get("index")
+                idx = _sse_index(data.get("index"))
                 target = (blocks_by_index.get(idx) if idx is not None else None) or current_block
                 if target is not None:
                     # Parse accumulated JSON into `input` for any block that
@@ -525,13 +591,15 @@ def _parse_sse_to_response(self, sse_data: str, provider: str) -> dict[str, Any]
                     current_block = None
 
             elif event_type == "message_delta":
-                delta = data.get("delta", {})
+                delta = _sse_dict(data.get("delta"))
                 if "stop_reason" in delta:
                     response["stop_reason"] = delta["stop_reason"]
                 if "stop_details" in delta:
                     response["stop_details"] = delta["stop_details"]
-                if data.get("usage"):
-                    response["usage"].update(data["usage"])
+                # Type-checked, not truthiness-checked: `dict.update` on a
+                # non-mapping raises. The message_start twin above already
+                # guarded this; this sibling did not.
+                response["usage"].update(_sse_dict(data.get("usage")))
 
         return response if response.get("content") else None