fix(ccr): preserve Anthropic re-stream shape (#1854)

## Description

Buffered Anthropic CCR re-streaming now preserves response shape instead
of normalizing newer Anthropic/Fable fields away during SSE
reconstruction.

Related upstream traffic checked before opening:

- #1451 added the direct streaming CCR buffered path and already
preserves thinking/signature/citation fields in
`StreamingMixin._response_to_sse`.
- #1825 / #1806 cover unknown Anthropic content block types such as
`server_tool_use`; this PR does not duplicate that fix.
- No open or closed issue/PR search result mentioned `stop_details`,
`signature_delta thinking_delta`, `refusal stop_reason`, `Fable CCR`, or
`re-stream thinking` as this exact gap.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Preserve `thinking`, `redacted_thinking`, `signature_delta`,
`citations_delta`, `stop_details`, and verbatim `stop_reason` while
parsing Anthropic SSE in `StreamingCCRHandler`.
- Reuse the shared proxy Anthropic SSE renderer for the legacy
`StreamingCCRHandler` output path so it preserves the same shape as the
direct buffered streaming CCR path.
- Preserve `stop_details` and stop defaulting missing `stop_reason` to
`end_turn` in `StreamingMixin._response_to_sse`.
- Add focused regressions for empty thinking blocks, signatures,
redacted thinking data, `refusal`, and `stop_details`.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [ ] Manual testing performed

### Test Output

```text
$ uv run --frozen --extra dev pytest tests/test_ccr_response_handler_extra.py tests/test_sse_thinking_blocks.py -q
18 passed in 0.29s

$ uv run --frozen --extra dev pytest tests/test_proxy/test_anthropic_streaming_ccr_retrieve.py -q
4 passed, 1 warning in 10.25s

$ uv run --frozen --extra dev ruff check headroom/ccr/response_handler.py headroom/proxy/handlers/streaming.py tests/test_ccr_response_handler_extra.py tests/test_sse_thinking_blocks.py
All checks passed!

$ uv run --frozen --extra dev ruff format --check headroom/ccr/response_handler.py headroom/proxy/handlers/streaming.py tests/test_ccr_response_handler_extra.py tests/test_sse_thinking_blocks.py
4 files already formatted
```

## Real Behavior Proof

- Environment: local worktree based on current `origin/main` after `git
fetch origin --prune && git rebase origin/main`.
- Exact command / steps: parse and re-emit a synthetic Anthropic SSE
stream containing an empty `thinking` block, `signature_delta`,
`redacted_thinking.data`, `message_delta.stop_reason = "refusal"`, and
`message_delta.stop_details`.
- Observed result: the reconstructed response and re-emitted SSE retain
the thinking/signature/redacted data plus `refusal` and `stop_details`;
a missing `stop_reason` is no longer rewritten to `end_turn`.
- Not tested: live upstream Fable/Opus traffic against the proxy.

## 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
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A

---------

Co-authored-by: JerrettDavis <mxjerrett@gmail.com>

CHANGELOG.md

@@ -104,6 +104,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * **cli:** the startup banner no longer advertises `HEADROOM_COMPRESSION_STABLE_AFTER_TURN` and `HEADROOM_STALE_READ_COMPRESS_AFTER_TURNS` as tuning knobs. Both were read only to render the `Performance Tuning` banner section and were never wired into the compression path, so setting them changed the banner but had no effect on behavior. The banner now surfaces only the embedding sidecar, which is a real, consumed setting.
 * **memory/embedder:** cap CPU thread oversubscription in the local torch/sentence-transformers embedder. Concurrent encodes previously each fanned out to ~`os.cpu_count()` BLAS/OpenMP threads, so under load the memory path starved the asyncio event loop and spiked `/livez` latency to several seconds. CPU encodes now run on a dedicated, size-limited executor whose workers each pin their thread pool, bounding total embedding threads to `HEADROOM_EMBED_CONCURRENCY` × `HEADROOM_EMBED_NUM_THREADS` (defaults `min(4, cpu)` × 1). The ONNX embedder already capped its threads; this brings the torch path to parity ([#198](https://github.com/headroomlabs-ai/headroom/issues/198)).
 * **proxy:** Buffered passthrough routes (e.g. `GET /v1/models`) no longer return an opaque HTTP 502 when an OpenAI-compatible upstream closes a pooled keep-alive connection mid-response (`httpx.RemoteProtocolError` / "incomplete chunked read"). Headroom now retries the request once on a fresh connection — mirroring a direct `curl` — and only returns a clear `upstream_protocol_error` 502 if the upstream is genuinely sending an incomplete response ([#1112](https://github.com/chopratejas/headroom/issues/1112)).
+* **ccr:** buffered Anthropic CCR re-streaming now preserves adaptive-thinking response shape, including empty `thinking` blocks, `signature_delta`, `redacted_thinking.data`, verbatim `stop_reason` values such as `refusal`, and `stop_details`.
 * **cursor:** `headroom wrap cursor` no longer injects the `rtk` custom-instructions block into `.cursorrules` when rtk's own native Cursor hook registers successfully. rtk supports a real hook for Cursor via `rtk init --agent cursor` (the same mechanism headroom already uses for Claude Code), which rewrites shell commands transparently — the injected `.cursorrules` text duplicated that guidance for no benefit. `wrap cursor` now tries the native hook first and only falls back to injecting `.cursorrules` if hook registration fails (#756).
 * **proxy:** The Headroom dashboard no longer tunnels `GET /favicon.ico` to the wrapped upstream provider. No route matched that path, so it fell through to the proxy's catch-all passthrough route and was forwarded to the configured Anthropic/OpenAI/etc. backend — burning a real upstream request (and possibly failing auth) for a browser's automatic favicon fetch on `/dashboard`. A dedicated `/favicon.ico` route now answers with `204 No Content` directly, registered ahead of the passthrough catch-all (#1787).
 

headroom/ccr/response_handler.py

@@ -734,62 +734,103 @@ class StreamingCCRHandler:
             "usage": {},
         }
 
-        current_text = ""
-        current_tool: dict[str, Any] | None = None
+        blocks_by_index: dict[int, dict[str, Any]] = {}
+        current_block: dict[str, Any] | None = None
 
         for event in events:
             event_type = event.get("type", "")
 
             if event_type == "content_block_start":
                 block = event.get("content_block", {})
-                if block.get("type") == "text":
-                    current_text = block.get("text", "")
-                elif block.get("type") == "tool_use":
-                    current_tool = {
-                        "type": "tool_use",
-                        "id": block.get("id", ""),
-                        "name": block.get("name", ""),
-                        "input": {},
-                    }
-
-            elif event_type == "content_block_delta":
-                delta = event.get("delta", {})
-                if delta.get("type") == "text_delta":
-                    current_text += delta.get("text", "")
-                elif delta.get("type") == "input_json_delta":
-                    # Accumulate JSON for tool input
-                    if current_tool is not None:
-                        partial = delta.get("partial_json", "")
-                        # This is tricky - partial JSON needs accumulation
-                        # For simplicity, we'll try to parse when complete
-                        current_tool["_partial_json"] = (
-                            current_tool.get("_partial_json", "") + partial
-                        )
-
-            elif event_type == "content_block_stop":
-                if current_text:
-                    response["content"].append(
+                block_index = event.get("index", len(blocks_by_index))
+                btype = block.get("type")
+                current_block = {"type": btype}
+                if btype == "text":
+                    current_block["text"] = block.get("text", "")
+                elif btype == "tool_use":
+                    current_block.update(
                         {
-                            "type": "text",
-                            "text": current_text,
+                            "id": block.get("id", ""),
+                            "name": block.get("name", ""),
+                            "input": {},
                         }
                     )
-                    current_text = ""
-                if current_tool:
-                    # Parse accumulated JSON
-                    partial = current_tool.pop("_partial_json", "")
-                    if partial:
-                        try:
-                            current_tool["input"] = json.loads(partial)
-                        except json.JSONDecodeError:
-                            current_tool["input"] = {}
-                    response["content"].append(current_tool)
-                    current_tool = None
+                elif btype == "thinking":
+                    current_block["thinking_buffer"] = block.get("thinking", "")
+                    if "signature" in block:
+                        current_block["signature"] = block["signature"]
+                elif btype == "redacted_thinking":
+                    if "data" in block:
+                        current_block["data"] = block["data"]
+                elif btype:
+                    current_block = dict(block)
+                blocks_by_index[block_index] = current_block
+
+            elif event_type == "content_block_delta":
+                idx = event.get("index")
+                target = (blocks_by_index.get(idx) if idx is not None else None) or current_block
+                if target is None:
+                    continue
+                delta = event.get("delta", {})
+                dtype = delta.get("type")
+                if dtype == "text_delta":
+                    target["text"] = target.get("text", "") + delta.get("text", "")
+                elif dtype == "input_json_delta":
+                    if target.get("type") == "tool_use":
+                        partial = delta.get("partial_json", "")
+                        target["_partial_json"] = target.get("_partial_json", "") + partial
+                elif dtype == "thinking_delta":
+                    target["thinking_buffer"] = target.get("thinking_buffer", "") + delta.get(
+                        "thinking", ""
+                    )
+                elif dtype == "signature_delta":
+                    if "signature" in delta:
+                        target["signature"] = delta["signature"]
+                elif dtype == "citations_delta":
+                    citation = delta.get("citation")
+                    if citation is not None:
+                        target.setdefault("citations", []).append(citation)
+
+            elif event_type == "content_block_stop":
+                idx = event.get("index")
+                target = (blocks_by_index.get(idx) if idx is not None else None) or current_block
+                if target is not None:
+                    if target.get("type") == "tool_use" and "_partial_json" in target:
+                        partial = target.pop("_partial_json", "")
+                        if partial:
+                            try:
+                                target["input"] = json.loads(partial)
+                            except json.JSONDecodeError:
+                                target["input"] = {}
+                    if target.get("type") == "thinking" and "thinking_buffer" in target:
+                        target["thinking"] = target.pop("thinking_buffer")
+                    if target not in response["content"]:
+                        response["content"].append(target)
+                    current_block = None
+
+            elif event_type == "message_start":
+                msg = event.get("message", {})
+                if "id" in msg:
+                    response["id"] = msg["id"]
+                if "model" in msg:
+                    response["model"] = msg["model"]
+                if "role" in msg:
+                    response["role"] = msg["role"]
+                if "stop_reason" in msg:
+                    response["stop_reason"] = msg["stop_reason"]
+                if "stop_details" in msg:
+                    response["stop_details"] = msg["stop_details"]
+                if msg.get("usage"):
+                    response["usage"].update(msg["usage"])
 
             elif event_type == "message_delta":
                 delta = event.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 event.get("usage"):
+                    response["usage"].update(event["usage"])
 
             elif event_type == "message_stop":
                 pass
@@ -859,11 +900,10 @@ class StreamingCCRHandler:
         to chunk the response more granularly.
         """
         if self.provider == "anthropic":
-            # Anthropic SSE format
-            yield b"event: message_start\n"
-            yield f"data: {json.dumps({'type': 'message_start', 'message': response})}\n\n".encode()
-            yield b"event: message_stop\n"
-            yield b'data: {"type": "message_stop"}\n\n'
+            from headroom.proxy.handlers.streaming import StreamingMixin
+
+            for chunk in StreamingMixin()._response_to_sse(response, "anthropic"):
+                yield chunk
         else:
             # OpenAI SSE format
             yield f"data: {json.dumps(response)}\n\n".encode()

headroom/proxy/handlers/streaming.py

@@ -384,6 +384,8 @@ class StreamingMixin:
                 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"])
 
@@ -482,8 +484,10 @@ class StreamingMixin:
 
             elif event_type == "message_delta":
                 delta = data.get("delta", {})
-                if delta.get("stop_reason"):
+                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"])
 
@@ -627,9 +631,14 @@ class StreamingMixin:
             events.append(f"event: content_block_stop\ndata: {json.dumps(block_stop)}\n\n".encode())
 
         # message_delta
+        msg_delta_payload: dict[str, Any] = {}
+        if "stop_reason" in response:
+            msg_delta_payload["stop_reason"] = response["stop_reason"]
+        if "stop_details" in response:
+            msg_delta_payload["stop_details"] = response["stop_details"]
         msg_delta = {
             "type": "message_delta",
-            "delta": {"stop_reason": response.get("stop_reason", "end_turn")},
+            "delta": msg_delta_payload,
             "usage": {"output_tokens": response.get("usage", {}).get("output_tokens", 0)},
         }
         events.append(f"event: message_delta\ndata: {json.dumps(msg_delta)}\n\n".encode())

tests/test_ccr_response_handler_extra.py

@@ -30,6 +30,19 @@ async def _async_iter(items: list[bytes]):
         yield item
 
 
+def _sse_json_events(chunks: list[bytes]) -> list[dict[str, Any]]:
+    events = []
+    for chunk in chunks:
+        for line in chunk.decode("utf-8").splitlines():
+            if not line.startswith("data: "):
+                continue
+            payload = line[len("data: ") :]
+            if payload == "[DONE]":
+                continue
+            events.append(json.loads(payload))
+    return events
+
+
 def test_extract_tool_calls_google_and_invalid_shapes() -> None:
     handler = CCRResponseHandler()
     google_response = {
@@ -202,23 +215,43 @@ def test_streaming_buffer_and_parse_sse_helpers() -> None:
     # Per SSE spec each event is terminated by `\n\n`. The byte-buffer
     # parser introduced in PR-A8 requires the spec terminator so partial
     # multi-byte UTF-8 reads don't corrupt event boundaries.
+    stop_details = {"type": "refusal", "message": "policy refusal"}
     anthropic_data = b"\n\n".join(
         [
+            b'data: {"type":"message_start","message":{"id":"msg_1","model":"claude-fable-5","role":"assistant","usage":{"input_tokens":7}}}',
+            b'data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}',
+            b'data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":""}}',
+            b'data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig_fable"}}',
+            b'data: {"type":"content_block_stop","index":0}',
+            b'data: {"type":"content_block_start","index":1,"content_block":{"type":"redacted_thinking","data":"ENC:abc"}}',
+            b'data: {"type":"content_block_stop","index":1}',
             b'data: {"type":"content_block_start","content_block":{"type":"text","text":"Hel"}}',
             b'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"lo"}}',
             b'data: {"type":"content_block_stop"}',
             b'data: {"type":"content_block_start","content_block":{"type":"tool_use","id":"tool_1","name":"headroom_retrieve"}}',
             b'data: {"type":"content_block_delta","delta":{"type":"input_json_delta","partial_json":"{\\"hash\\":\\"abc\\"}"}}',
             b'data: {"type":"content_block_stop"}',
-            b'data: {"type":"message_delta","delta":{"stop_reason":"tool_use"}}',
+            (
+                b'data: {"type":"message_delta","delta":{"stop_reason":"refusal",'
+                b'"stop_details":{"type":"refusal","message":"policy refusal"}},'
+                b'"usage":{"output_tokens":3}}'
+            ),
             b"data: [DONE]\n\n",
         ]
     )
     parsed = handler._parse_sse_stream(anthropic_data)
-    assert parsed["content"][0] == {"type": "text", "text": "Hello"}
-    assert parsed["content"][1]["name"] == "headroom_retrieve"
-    assert parsed["content"][1]["input"] == {"hash": "abc"}
-    assert parsed["stop_reason"] == "tool_use"
+    assert parsed["content"][0] == {
+        "type": "thinking",
+        "signature": "sig_fable",
+        "thinking": "",
+    }
+    assert parsed["content"][1] == {"type": "redacted_thinking", "data": "ENC:abc"}
+    assert parsed["content"][2] == {"type": "text", "text": "Hello"}
+    assert parsed["content"][3]["name"] == "headroom_retrieve"
+    assert parsed["content"][3]["input"] == {"hash": "abc"}
+    assert parsed["stop_reason"] == "refusal"
+    assert parsed["stop_details"] == stop_details
+    assert parsed["usage"]["output_tokens"] == 3
 
     openai_handler = StreamingCCRHandler(CCRResponseHandler(), provider="openai")
     parsed_openai = openai_handler._reconstruct_openai_response(
@@ -349,9 +382,46 @@ async def test_streaming_handler_falls_back_to_buffer_on_processing_error(
 async def test_response_to_sse_formats() -> None:
     anthropic = StreamingCCRHandler(CCRResponseHandler(), provider="anthropic")
     anthropic_chunks = [chunk async for chunk in anthropic._response_to_sse({"content": []})]
-    assert anthropic_chunks[0] == b"event: message_start\n"
-    assert anthropic_chunks[-1] == b'data: {"type": "message_stop"}\n\n'
+    assert anthropic_chunks[0].startswith(b"event: message_start\n")
+    assert anthropic_chunks[-1] == b'event: message_stop\ndata: {"type": "message_stop"}\n\n'
 
     openai = StreamingCCRHandler(CCRResponseHandler(), provider="openai")
     openai_chunks = [chunk async for chunk in openai._response_to_sse({"choices": []})]
     assert openai_chunks == [b'data: {"choices": []}\n\n', b"data: [DONE]\n\n"]
+
+
+@pytest.mark.asyncio
+async def test_response_to_sse_preserves_anthropic_shape() -> None:
+    handler = StreamingCCRHandler(CCRResponseHandler(), provider="anthropic")
+    stop_details = {"type": "refusal", "message": "policy refusal"}
+    response = {
+        "id": "msg_1",
+        "model": "claude-fable-5",
+        "role": "assistant",
+        "content": [
+            {"type": "thinking", "thinking": "", "signature": "sig_fable"},
+            {"type": "redacted_thinking", "data": "ENC:abc"},
+            {"type": "text", "text": "done"},
+        ],
+        "stop_reason": "refusal",
+        "stop_details": stop_details,
+        "usage": {"input_tokens": 7, "output_tokens": 3},
+    }
+
+    chunks = [chunk async for chunk in handler._response_to_sse(response)]
+    events = _sse_json_events(chunks)
+    message_delta = next(event for event in events if event["type"] == "message_delta")
+
+    assert message_delta["delta"]["stop_reason"] == "refusal"
+    assert message_delta["delta"]["stop_details"] == stop_details
+    assert any(event.get("delta", {}).get("type") == "signature_delta" for event in events)
+    assert any(
+        event.get("content_block", {}).get("type") == "redacted_thinking" for event in events
+    )
+
+    parsed = handler._parse_sse_stream(b"".join(chunks))
+    assert parsed["content"][0]["signature"] == "sig_fable"
+    assert parsed["content"][0]["thinking"] == ""
+    assert parsed["content"][1]["data"] == "ENC:abc"
+    assert parsed["stop_reason"] == "refusal"
+    assert parsed["stop_details"] == stop_details

tests/test_sse_thinking_blocks.py

@@ -218,6 +218,7 @@ def test_response_to_sse_preserves_server_tool_use_blocks() -> None:
 def test_response_to_sse_preserves_thinking_redacted_and_citations() -> None:
     parser = _Parser()
     redacted_blob = "ENC:" + ("y" * 200)
+    stop_details = {"type": "refusal", "message": "policy refusal"}
     response = {
         "id": "msg_2",
         "model": "claude-opus-4",
@@ -237,7 +238,8 @@ def test_response_to_sse_preserves_thinking_redacted_and_citations() -> None:
             },
             {"type": "redacted_thinking", "data": redacted_blob},
         ],
-        "stop_reason": "end_turn",
+        "stop_reason": "refusal",
+        "stop_details": stop_details,
         "usage": {"input_tokens": 10, "output_tokens": 3},
     }
 
@@ -255,6 +257,22 @@ def test_response_to_sse_preserves_thinking_redacted_and_citations() -> None:
     assert round_tripped["content"][0]["signature"] == "sig_123"
     assert round_tripped["content"][1]["citations"][0]["cited_text"] == "abc"
     assert round_tripped["content"][2]["data"] == redacted_blob
+    assert round_tripped["stop_reason"] == "refusal"
+    assert round_tripped["stop_details"] == stop_details
+
+
+def test_response_to_sse_does_not_default_missing_stop_reason() -> None:
+    parser = _Parser()
+    sse_text = b"".join(parser._response_to_sse({"content": []}, "anthropic")).decode("utf-8")
+    events = [
+        json.loads(line[len("data: ") :])
+        for line in sse_text.splitlines()
+        if line.startswith("data: ")
+    ]
+    message_delta = next(event for event in events if event["type"] == "message_delta")
+
+    assert message_delta["delta"] == {}
+    assert "end_turn" not in sse_text
 
 
 def test_response_to_sse_rejects_unknown_content_block() -> None: