fix(proxy): preserve signed Anthropic thinking blocks on outbound re-serialize (#2254)

## Description

When multi-turn Anthropic requests include signed `thinking` or
`redacted_thinking` blocks in conversation history, the proxy
re-serializes the body through `serialize_body_canonical` whenever
`body_mutated` is true. That re-encode changes the byte representation
of signed blocks and upstream rejects the turn with 400 "blocks cannot
be modified".

This detects those content blocks and, when original request bytes are
available, forwards them byte-for-byte instead of re-encoding. That
matches the preferred option from the issue and mirrors the existing
Agno skip for thinking-bearing histories.

Closes #2251

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

- Add `has_signed_thinking_blocks()` in
`headroom/proxy/body_forwarding.py`
- Prefer original-byte passthrough in `select_outbound_body` when signed
thinking blocks are present and original bytes exist
- Unit tests for thinking and redacted_thinking passthrough,
missing-original canonical fallback, legacy override, and unchanged
non-thinking behavior

## 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 pytest tests/test_proxy_byte_faithful_forwarding.py -q --tb=short
# 43 passed, 1 skipped

uv run ruff check / format on touched files
# passed
```

## Real Behavior Proof

- Environment: unit-level body forwarding with multi-turn
Anthropic-shaped payloads containing signed `thinking` /
`redacted_thinking` blocks
- Exact command / steps: focused pytest suite above
- Observed result: with `body_mutated=True` and original bytes present,
outbound source is `passthrough` and content equals original bytes;
without original bytes, behavior remains canonical; non-thinking mutated
bodies still use canonical
- Not tested: full `headroom wrap claude` multi-turn session against
Anthropic / Claude Code (no live Claude credentials here)

## 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
- [ ] 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
- [ ] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A

## Additional Notes

- Requests without thinking blocks keep existing
passthrough/canonical/legacy selection
- When original bytes are unavailable, signed-thinking requests still
re-serialize

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

headroom/proxy/body_forwarding.py

@@ -4,6 +4,7 @@ This module owns the small algebra used by Python proxy forwarders to decide
 which bytes leave Headroom:
 
 * unmutated body with original bytes -> byte-for-byte passthrough
+* signed thinking blocks with original bytes -> byte-for-byte passthrough
 * mutated body or missing original bytes -> canonical JSON bytes
 * explicit rollback mode -> legacy httpx-style JSON bytes
 """
@@ -50,6 +51,27 @@ def serialize_body_canonical(body: dict[str, Any]) -> bytes:
     return json.dumps(body, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
 
 
+def has_signed_thinking_blocks(body: dict[str, Any]) -> bool:
+    """Return whether message history contains Anthropic signed thinking blocks."""
+    messages = body.get("messages")
+    if not isinstance(messages, list):
+        return False
+
+    for message in messages:
+        if not isinstance(message, dict):
+            continue
+        content = message.get("content")
+        if not isinstance(content, list):
+            continue
+        for block in content:
+            if isinstance(block, dict) and block.get("type") in {
+                "thinking",
+                "redacted_thinking",
+            }:
+                return True
+    return False
+
+
 class BodyMutationTracker:
     """Records whether a request body was mutated and why."""
 
@@ -85,6 +107,9 @@ def select_outbound_body(
 ) -> OutboundBody:
     """Select the exact bytes to forward upstream."""
     mode = forwarder_mode if forwarder_mode is not None else get_python_forwarder_mode()
+    if original_body_bytes is not None and has_signed_thinking_blocks(body):
+        return OutboundBody(content=original_body_bytes, source="passthrough")
+
     if mode == "legacy_json_kwarg":
         content = json.dumps(body, separators=(", ", ": "), ensure_ascii=True).encode("utf-8")
         return OutboundBody(content=content, source="legacy")

tests/test_proxy_byte_faithful_forwarding.py

@@ -172,6 +172,89 @@ def test_prepare_outbound_mutated_uses_canonical() -> None:
     assert source == "canonical"
 
 
+@pytest.mark.parametrize("block_type", ["thinking", "redacted_thinking"])
+def test_signed_thinking_history_with_original_bytes_uses_passthrough(
+    block_type: str,
+) -> None:
+    body = {
+        "model": "claude-sonnet-4-5",
+        "messages": [
+            {"role": "user", "content": "Solve this"},
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": block_type,
+                        "thinking": "private reasoning",
+                        "signature": "sig123",
+                    },
+                    {"type": "text", "text": "The answer is 42."},
+                ],
+            },
+            {"role": "user", "content": "Continue"},
+        ],
+    }
+    original = json.dumps(body, indent=2).encode("utf-8")
+
+    outbound = select_outbound_body(
+        body=body,
+        original_body_bytes=original,
+        body_mutated=True,
+        forwarder_mode="byte_faithful",
+    )
+
+    assert outbound.source == "passthrough"
+    assert outbound.content == original
+
+
+def test_signed_thinking_history_without_original_bytes_uses_canonical() -> None:
+    body = {
+        "messages": [
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "thinking",
+                        "thinking": "private reasoning",
+                        "signature": "sig123",
+                    }
+                ],
+            }
+        ]
+    }
+
+    outbound = select_outbound_body(
+        body=body,
+        original_body_bytes=None,
+        body_mutated=True,
+        forwarder_mode="byte_faithful",
+    )
+
+    assert outbound.source == "canonical"
+    assert outbound.content == serialize_body_canonical(body)
+
+
+def test_signed_thinking_history_overrides_legacy_encoder() -> None:
+    body = {
+        "messages": [
+            {
+                "role": "assistant",
+                "content": [{"type": "thinking", "signature": "sig123"}],
+            }
+        ]
+    }
+    original = json.dumps(body, indent=2).encode("utf-8")
+
+    outbound = select_outbound_body(
+        body=body,
+        original_body_bytes=original,
+        body_mutated=True,
+        forwarder_mode="legacy_json_kwarg",
+    )
+
+    assert outbound == OutboundBody(content=original, source="passthrough")
+
+
 def test_prepare_outbound_no_original_bytes_uses_canonical() -> None:
     out, source = prepare_outbound_body_bytes(
         body={"a": 1},