mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
fix(proxy): reassemble server_tool_use.input from streamed partial_json (#2449)
## Description
Under `--target-ratio 0.4` a session died mid-run with a fatal Anthropic
400:
```
messages.13.content.0.server_tool_use.input: Input should be an object
```
Root cause is not compression of the request: the request path passes
structured blocks through byte-for-byte. It is **SSE stream
reconstruction**. When the proxy rebuilds a full Anthropic message from
the streamed response (non-stream retry, buffered, and CCR round-trip
paths), the `content_block_stop` handler parsed the accumulated
`_partial_json` into `input` only for blocks whose type was exactly
`tool_use`. A `server_tool_use` block streams its input identically via
`input_json_delta`, so its input was never reassembled: the block kept
the empty start-event `input: {}` and leaked the internal
`_partial_json` scratch key. That reconstructed block becomes assistant
history, and on the next turn the client replays it, so Anthropic
rejects `server_tool_use.input`. `--target-ratio` only makes the
buffered/reconstructed path more likely; it does not itself rewrite the
block.
Refs #2438 (Finding 2). Findings 1 (prompt-cache regression) and 3
(compression not engaging) are architectural and tracked separately.
## 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
- `headroom/proxy/handlers/streaming.py` (`_parse_sse_to_response`):
gate the `content_block_stop` `_partial_json` → `input` parse on the
presence of `_partial_json`, not `type == "tool_use"`, so
`server_tool_use` (and any future tool-ish block) is reassembled. Always
strip the scratch key; `input` is always a parsed object (`{}` on
malformed/empty JSON).
- `headroom/ccr/response_handler.py`
(`StreamingCCRHandler._reconstruct_anthropic_response`): same
stop-handler fix, and relax the `input_json_delta` accumulator that was
likewise gated on `type == "tool_use"` so server_tool_use partial JSON
is accumulated at all.
- Regression tests in `tests/test_sse_thinking_blocks.py` and
`tests/test_ccr_response_handler_extra.py`: a `server_tool_use` whose
input arrives via `input_json_delta` must reconstruct to the parsed
object with no `_partial_json` leak.
- Leave `CHANGELOG.md` untouched, release-please generates it.
## Testing
- [x] Unit tests pass (`python -m pytest
tests/test_sse_thinking_blocks.py
tests/test_ccr_response_handler_extra.py -q`)
- [x] Linting passes (`ruff check`, `ruff format --check` on the four
changed files)
- [x] Type checking passes (`mypy headroom/proxy/handlers/streaming.py
headroom/ccr/response_handler.py --ignore-missing-imports`)
- [x] New tests added for new functionality
### Test Output
```text
$ python -m pytest tests/test_sse_thinking_blocks.py tests/test_ccr_response_handler_extra.py -q
26 passed in 3.36s
$ ruff check <changed files>
All checks passed!
```
## Real Behavior Proof
- Environment: Windows 11, Python 3.13, local dev checkout on a branch
off upstream/main
- Exact command / steps: Fed a synthetic Anthropic SSE stream with a
`server_tool_use` block whose `input` arrives as `input_json_delta`
partial JSON through both reconstructors (`_parse_sse_to_response`,
`_reconstruct_anthropic_response`); then temporarily restored the `type
== "tool_use"` guard and re-ran.
- Observed result: With the fix, the reconstructed block has `input ==
{"query": ...}` and no `_partial_json` key. With the old guard the test
fails, `input` stays `{}` and the scratch key leaks, reproducing the
malformed block that Anthropic rejects on replay.
- Not tested: End-to-end multi-turn `--target-ratio` session against the
live Anthropic API from this environment, reproduced at the
reconstruction seam instead; the reporter observed the 400 on real
traffic.
## Review Readiness
- [x] I have performed a self-review
- [x] This PR is ready for human review
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
bec4cce8a9
commit
8c8fae0d0b
4 changed files with 102 additions and 14 deletions
|
|
@ -811,9 +811,11 @@ class StreamingCCRHandler:
|
|||
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
|
||||
# Accumulate for any block streaming input (tool_use AND
|
||||
# server_tool_use); the stop handler parses it into `input`
|
||||
# (#2438).
|
||||
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", ""
|
||||
|
|
@ -830,13 +832,17 @@ class StreamingCCRHandler:
|
|||
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"] = {}
|
||||
# Parse streamed `_partial_json` into `input` for any block
|
||||
# that carried input_json_delta — tool_use AND
|
||||
# server_tool_use — not just tool_use. The narrow type gate
|
||||
# left server_tool_use.input malformed and leaked the scratch
|
||||
# key into replayed history (#2438). Always strip the key.
|
||||
if "_partial_json" in target:
|
||||
partial = target.pop("_partial_json")
|
||||
try:
|
||||
target["input"] = json.loads(partial) if partial else {}
|
||||
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"]:
|
||||
|
|
|
|||
|
|
@ -494,13 +494,20 @@ class StreamingMixin:
|
|||
idx = 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 for tool_use blocks.
|
||||
if target.get("type") == "tool_use" and "_partial_json" in target:
|
||||
# Parse accumulated JSON into `input` for any block that
|
||||
# streamed `input_json_delta` — tool_use AND server_tool_use
|
||||
# (and future tool-ish blocks). Gating on the block type
|
||||
# missed server_tool_use, leaving its `input` at the empty
|
||||
# start-event value and leaking the `_partial_json` scratch
|
||||
# key into replayed assistant history, which Anthropic then
|
||||
# rejects with `server_tool_use.input: Input should be an
|
||||
# object` (#2438). Always strip the scratch key.
|
||||
if "_partial_json" in target:
|
||||
raw = target.pop("_partial_json")
|
||||
try:
|
||||
target["input"] = json.loads(target["_partial_json"])
|
||||
target["input"] = json.loads(raw) if raw else {}
|
||||
except json.JSONDecodeError:
|
||||
target["input"] = {}
|
||||
del target["_partial_json"]
|
||||
# Materialize the thinking buffer into the
|
||||
# canonical `thinking` field expected by the
|
||||
# Anthropic API.
|
||||
|
|
|
|||
|
|
@ -425,3 +425,36 @@ async def test_response_to_sse_preserves_anthropic_shape() -> None:
|
|||
assert parsed["content"][1]["data"] == "ENC:abc"
|
||||
assert parsed["stop_reason"] == "refusal"
|
||||
assert parsed["stop_details"] == stop_details
|
||||
|
||||
|
||||
def test_reconstruct_server_tool_use_input_from_partial_json() -> None:
|
||||
# StreamingCCRHandler._reconstruct_anthropic_response must parse streamed
|
||||
# input_json_delta into `input` for server_tool_use, not only tool_use.
|
||||
# The narrow type gate left server_tool_use.input malformed and leaked the
|
||||
# `_partial_json` scratch key into replayed assistant history → Anthropic
|
||||
# 400 `server_tool_use.input: Input should be an object` (#2438).
|
||||
handler = StreamingCCRHandler(CCRResponseHandler(), provider="anthropic")
|
||||
events = [
|
||||
{"type": "message_start", "message": {"id": "msg_1", "model": "claude-opus-4"}},
|
||||
{
|
||||
"type": "content_block_start",
|
||||
"index": 0,
|
||||
"content_block": {
|
||||
"type": "server_tool_use",
|
||||
"id": "srvtoolu_1",
|
||||
"name": "web_search",
|
||||
"input": {},
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "content_block_delta",
|
||||
"index": 0,
|
||||
"delta": {"type": "input_json_delta", "partial_json": '{"query": "x"}'},
|
||||
},
|
||||
{"type": "content_block_stop", "index": 0},
|
||||
]
|
||||
response = handler._reconstruct_anthropic_response(events)
|
||||
block = response["content"][0]
|
||||
assert block["type"] == "server_tool_use"
|
||||
assert block["input"] == {"query": "x"}
|
||||
assert "_partial_json" not in block
|
||||
|
|
|
|||
|
|
@ -499,3 +499,45 @@ def test_buffered_ccr_extended_thinking_round_trip_preserves_all_blocks() -> Non
|
|||
)
|
||||
assert round_tripped["content"][2]["type"] == "tool_use"
|
||||
assert round_tripped["content"][2]["input"] == {"y": 2}
|
||||
|
||||
|
||||
def test_server_tool_use_input_reassembled_from_partial_json() -> None:
|
||||
# server_tool_use streams its input via input_json_delta exactly like
|
||||
# tool_use: the content_block_start carries an empty input, the real args
|
||||
# arrive as partial_json, and content_block_stop must parse them into an
|
||||
# object. Gating the parse on type == "tool_use" left server_tool_use.input
|
||||
# empty and leaked the `_partial_json` scratch key into replayed assistant
|
||||
# history, which Anthropic rejects on the next turn (#2438).
|
||||
parser = _Parser()
|
||||
events = [
|
||||
{"type": "message_start", "message": {"id": "msg_1", "model": "claude-opus-4"}},
|
||||
{
|
||||
"type": "content_block_start",
|
||||
"index": 0,
|
||||
"content_block": {
|
||||
"type": "server_tool_use",
|
||||
"id": "srvtoolu_1",
|
||||
"name": "web_search",
|
||||
"input": {},
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "content_block_delta",
|
||||
"index": 0,
|
||||
"delta": {"type": "input_json_delta", "partial_json": '{"query": "hea'},
|
||||
},
|
||||
{
|
||||
"type": "content_block_delta",
|
||||
"index": 0,
|
||||
"delta": {"type": "input_json_delta", "partial_json": 'droom proxy"}'},
|
||||
},
|
||||
{"type": "content_block_stop", "index": 0},
|
||||
]
|
||||
sse = _build_sse(events)
|
||||
response = parser._parse_sse_to_response(sse, "anthropic")
|
||||
assert response is not None
|
||||
block = response["content"][0]
|
||||
assert block["type"] == "server_tool_use"
|
||||
assert block["input"] == {"query": "headroom proxy"}
|
||||
# Scratch key must never leak into a block that gets replayed as history.
|
||||
assert "_partial_json" not in block
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue