github-featured/chopratejas--headroom
2
0
Fork 0
Code Issues Pull Requests Actions 16 Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(ccr): relay a successful upstream turn when post-processing fails (#3094)

Browse Source 
## Description

Closes #3088

The buffered CCR path flips a streaming turn to `stream: false` so a
`headroom_retrieve` call can be resolved server-side. Everything it does
*after* the provider answers — retrieval, memory tool calls, turn hooks,
usage accounting, caching, SSE resynthesis — is post-processing layered
on a turn that already succeeded and was already billed.

When any of that raised, the entire turn surfaced to the client as a
generic `api_error`. In the reported capture the provider returned a
complete **69,351-byte** answer in 1.9s and the client received **1,841
bytes**: keepalives, then a synthesized failure. A paid-for response was
discarded because a bookkeeping step downstream of it broke.

**On the reporter's stated root cause:** the "≈30s compression timeout"
inference does not hold. Their own log says *"[12 seconds later]"*,
which matches 49 pings × the 0.25s post-commit interval, not 30s. And
`COMPRESSION_TIMEOUT_SECONDS` guards `_count_offloaded`, which **fails
open** to estimation and cannot propagate. So that correlation is a
coincidence.

**What actually raises is still unidentified**, and that is the second
half of this report. The handler logged `f"Request failed:
{type(e).__name__}: {e}"` with no `exc_info`, which is exactly why the
reporter found "no visible traceback" — and why reading the entire
post-upstream path (memory tool calls, `run_response_hooks`, CCR
handling all catch internally) does not reveal it either.

## Type of Change

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

## Changes Made

- Capture the upstream response the moment it parses as a 200, before
any post-processing can touch it.
- Wrap the buffered operation so an unexpected raise relays that
captured response as SSE instead of a synthesized error.
- Log the exception with `exc_info=True`. Salvaging **without** this
would paper over the defect permanently; the goal is to stop losing user
turns while making the real bug diagnosable.
- Refuse to salvage a response the client cannot safely consume. A reply
still carrying an unresolved `headroom_retrieve` call is exactly the
case the handler already fails closed on — relaying it would hand the
client a tool call it is not expected to service and a marker nobody
expanded. The check reuses the existing `residual_ccr_status` /
`has_ccr_tool_calls` signals rather than inventing a second notion of
"safe".

**This is containment, not root cause.** It converts a hard failure on a
successful turn into a degraded success, and makes the underlying raise
visible so it can be fixed properly. I have said so in the commit
message too, so this is not mistaken for a full diagnosis later.

## Testing

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

New `tests/test_buffered_ccr_salvage.py` covers the reported shape
(thinking + text, and the captured `bash` tool_use turn with no retrieve
call), that the healthy path is untouched, and that an unresolved
retrieve call is never relayed.

### Test Output

```text
# BEFORE (main) — the same test file reproduces the report exactly:
E   AssertionError: {"type": "error", "error": {"type": "api_error",
    "message": "An error occurred while processing your request. Please try again."}}
E   assert 502 == 200

# AFTER (this branch):
$ pytest tests/test_buffered_ccr_salvage.py -q
8 passed, 1 warning in 2.94s

$ pytest tests/ -q
3 failed, 11168 passed, 581 skipped in 421.26s (0:07:01)

Same 3 failures as a clean-main baseline run on this machine:
  tests/test_graceful_shutdown.py::test_run_server_installs_cancelled_error_filter
  tests/test_learn/test_integration.py::TestCodexIntegration::test_full_pipeline
  tests/test_release_workflows.py::test_no_native_tls_in_wheel_build_tree

$ ruff check . && ruff format --check .
All checks passed!
```

## Real Behavior Proof

- Environment: this branch driven through the real FastAPI app with a
stubbed upstream returning a complete 200 turn; macOS arm64, Python
3.12.
- Exact command / steps: posted a buffered CCR turn, then forced a
post-upstream step to raise (`_record_request_outcome`), standing in for
whatever breaks in the field; ran the identical test file against `main`
and against this branch.
- Observed result: on `main` the client gets HTTP 502 with the report's
literal `api_error` string; on this branch the client gets HTTP 200
`text/event-stream` carrying the provider's own content
(`message_start`, thinking, text / `toolu_bash`) and no invented error.
- Not tested: the field defect itself. What raises in the reporter's
environment is still unknown — that is what the added traceback logging
exists to surface. A follow-up will need their logs on a build carrying
this change.

## Runtime Rollout Safety

- Rollout-managed feature(s): none — this guards an existing code path
and is not behind a rollout channel.
- Minimum rollout channel: n/a (ships to stable with the fix).
- Stable/default behavior changed: yes, and only in the failure case. A
buffered turn whose post-processing raises now returns the upstream's
answer instead of a 502 `api_error`. Successful turns are
byte-identical.
- Kill switch / disable path: no new switch. The guard only engages on
an exception that previously produced a hard failure, so disabling it
would restore the bug.
- Unsafe override required: no.
- Qualification impact: none — no qualification-gated surface is
touched.
- Rollback path: revert this commit; the previous behavior (synthesized
`api_error`) returns.

## 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
- [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 did **not** edit `CHANGELOG.md`

## Additional Notes

Documentation update is marked N/A: no user-facing flag or endpoint
changes. Type checking (`mypy headroom`) was not run separately; `ruff`
is the gate this repo's CI enforces.

Same family, still open: #3078, #3082, #3017, #2857, #2825. The added
traceback is the fastest route to whether they share this root cause.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Tejas Chopra <tejas@Tejass-MacBook-Pro.local>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Tejas Chopra
2026-08-17 18:06:59 -07:00
committed by GitHub
parent c16be9bbbe
commit 0ec73faa28
2 changed files with 321 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
headroom/proxy/handlers/anthropic.py
+93 -2
View File
@@ -318,6 +318,34 @@ class AnthropicHandlerMixin:
return True
return False
def _can_salvage_buffered_upstream(self, resp_json: Any) -> bool:
"""May this upstream response be relayed when post-processing failed?
Only when the client can actually consume it. The buffered path exists
because a ``headroom_retrieve`` call has to be resolved server-side, so
a response still carrying one is exactly the case the handler already
fails closed on: relaying it would hand the client a tool call naming an
endpoint it is not expected to reach, and a marker nobody expanded.
Anything else — an ordinary answer, a client tool call, a turn whose
retrieval already resolved — is a complete provider turn and is safer in
the client's hands than a synthesized error (#3088).
"""
if not isinstance(resp_json, dict):
return False
handler = getattr(self, "ccr_response_handler", None)
if handler is None:
return True
try:
from headroom.ccr.response_handler import RESIDUAL_CCR_ERROR
if handler.residual_ccr_status(resp_json, "anthropic") == RESIDUAL_CCR_ERROR:
return False
return not handler.has_ccr_tool_calls(resp_json, "anthropic")
except Exception: # pragma: no cover - defensive
logger.debug("CCR: salvage check failed; not salvaging", exc_info=True)
return False
@staticmethod
def _outgoing_body_has_redeemable_marker(body: Any) -> bool:
"""Does the body about to be sent carry a marker retrieval could expand?
@@ -3610,6 +3638,9 @@ class AnthropicHandlerMixin:
session_key=session_key,
)
else:
# Populated once the upstream answers 200 with parseable
# JSON, so the guard below can fall back to it (#3088).
_salvageable_upstream: dict[str, Any] = {}
async def _buffered_ccr_operation():
async with stage_timer.measure("upstream_connect"):
@@ -3781,6 +3812,16 @@ class AnthropicHandlerMixin:
resp_json = None
try:
resp_json = response.json()
if buffered_stream_ccr and response.status_code == 200 and resp_json:
# Remember the upstream's own answer before any
# post-processing touches it. Everything from here
# to the SSE resynthesis — retrieval, memory tool
# calls, turn hooks, usage accounting, caching — is
# work layered on top of a turn the provider has
# already produced and billed. If any of it raises
# unexpectedly, this is what the client should get
# instead of a synthesized error (#3088).
_salvageable_upstream["resp_json"] = resp_json
except (json.JSONDecodeError, ValueError) as e:
# DEBUG is right for the buffered non-stream path, where
# an unparseable body is just "no CCR handling". On the
@@ -4496,8 +4537,58 @@ class AnthropicHandlerMixin:
headers=response_headers,
)
async def _buffered_ccr_operation_salvaging():
"""Never trade a successful upstream turn for a synthesized error.
Everything the buffered path does after the provider answers
— server-side retrieval, memory tool calls, turn hooks, usage
accounting, caching, SSE resynthesis — is post-processing on
a turn that already succeeded and was already billed. When a
step raised unexpectedly the whole turn surfaced as a generic
``api_error``, so the client lost a complete 69KB answer the
provider had produced (#3088), and the cause was unlogged.
Relay the upstream's own answer instead. Two things are
deliberately preserved: the exception is logged with a
traceback so the real defect stays diagnosable rather than
being papered over, and a response the client cannot safely
consume is never salvaged — see ``_can_salvage``.
"""
try:
return await _buffered_ccr_operation()
except asyncio.CancelledError:
raise
except Exception:
salvaged = _salvageable_upstream.get("resp_json")
if salvaged is None or not self._can_salvage_buffered_upstream(salvaged):
raise
logger.error(
f"[{request_id}] CCR: buffered post-processing failed after a "
"successful upstream turn; relaying the upstream response "
"instead of failing the request (#3088)",
exc_info=True,
)
try:
events = self._response_to_sse(salvaged, "anthropic")
except Exception:
logger.error(
f"[{request_id}] CCR: could not resynthesize the salvaged "
"upstream response; failing the request",
exc_info=True,
)
raise
async def _salvaged_sse():
for event in events:
yield event
return StreamingResponse(
_salvaged_sse(),
media_type="text/event-stream",
)
if buffered_stream_ccr:
operation = asyncio.create_task(_buffered_ccr_operation())
operation = asyncio.create_task(_buffered_ccr_operation_salvaging())
# Holds out for the real status, then keeps the stream alive
# once waiting silently would risk the client's idle
@@ -4520,7 +4611,7 @@ class AnthropicHandlerMixin:
await _buffered_call(scope, receive, send)
return _BufferedCCRResponse(media_type="text/event-stream")
return await _buffered_ccr_operation()
return await _buffered_ccr_operation_salvaging()
except HTTPException:
# FastAPI HTTPException carries its own status code, headers,
# and client-facing message (e.g. 429 with Retry-After, 413 for
tests/test_buffered_ccr_salvage.py
+228
View File
@@ -0,0 +1,228 @@
"""A successful upstream turn must never become a synthesized error (#3088).
The buffered CCR path flips a streaming turn to ``stream: false`` so retrieval
can be resolved server-side. Everything it does *after* the provider answers —
retrieval, memory tool calls, turn hooks, usage accounting, caching, SSE
resynthesis — is post-processing layered on a turn that already succeeded and
was already billed.
When one of those steps raised, the whole turn surfaced to the client as:
event: error
data: {"type":"error","error":{"type":"api_error", ...}}
In the reported capture the provider had returned a complete 69,351-byte answer
in 1.9s; the client received 1,841 bytes of keepalives and that error. The
answer was paid for and thrown away, and no traceback was logged, so the real
defect stayed invisible.
These tests pin the two halves of the fix: relay the upstream's own answer
rather than inventing a failure, and refuse to relay a response the client
cannot safely consume.
"""
from __future__ import annotations
import json
import pytest
fastapi = pytest.importorskip("fastapi")
httpx = pytest.importorskip("httpx")
from fastapi.testclient import TestClient # noqa: E402
from headroom.cache.backends import InMemoryBackend # noqa: E402
from headroom.cache.compression_store import ( # noqa: E402
get_compression_store,
reset_compression_store,
)
from headroom.ccr.tool_injection import create_ccr_tool_definition # noqa: E402
from headroom.proxy.server import ProxyConfig, create_app # noqa: E402
def _config() -> ProxyConfig:
return ProxyConfig(
optimize=False,
cache_enabled=False,
rate_limit_enabled=False,
memory_enabled=False,
ccr_inject_tool=True,
ccr_handle_responses=True,
ccr_context_tracking=False,
image_optimize=False,
# Commit immediately, so a failure is exercised on the committed path
# too — the shape the report was filed against.
buffered_ccr_grace_seconds=5.0,
)
@pytest.fixture(autouse=True)
def _store():
reset_compression_store()
get_compression_store(backend=InMemoryBackend())
try:
yield
finally:
reset_compression_store()
def _marker() -> str:
return get_compression_store().store(
original=json.dumps({"earlier": "tool output"}),
compressed="{}",
original_item_count=1,
)
def _upstream(content: list[dict], stop_reason: str = "end_turn") -> dict:
return {
"id": "msg_upstream",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-6",
"content": content,
"stop_reason": stop_reason,
"usage": {
"input_tokens": 1200,
"output_tokens": 295,
"cache_read_input_tokens": 0,
"cache_creation_input_tokens": 0,
},
}
def _body() -> dict:
return {
"model": "claude-sonnet-4-6",
"max_tokens": 512,
"stream": True,
"tools": [create_ccr_tool_definition("anthropic")],
"messages": [{"role": "user", "content": f"go (earlier output at <<ccr:{_marker()}>>)"}],
}
def _headers() -> dict[str, str]:
return {"x-api-key": "test-key", "anthropic-version": "2023-06-01"}
def _run(upstream: dict, *, break_post_processing: bool):
"""Drive one buffered turn, optionally exploding after the upstream answers."""
app = create_app(_config())
with TestClient(app) as client:
proxy = client.app.state.proxy
async def _fake_retry(method, url, headers, body, stream=False, **kwargs): # noqa: ANN001
return httpx.Response(200, json=upstream)
proxy._retry_request = _fake_retry # type: ignore[assignment]
if break_post_processing:
# Stand in for any of the post-upstream steps failing. The point is
# that the provider already answered; what broke is ours.
real = proxy._record_request_outcome
async def _boom(*args, **kwargs): # noqa: ANN002, ANN003
raise RuntimeError("post-processing exploded")
proxy._record_request_outcome = _boom # type: ignore[assignment]
assert real is not None
return client.post("/v1/messages", json=_body(), headers=_headers())
# --------------------------------------------------------------------------- #
# The reported failure
# --------------------------------------------------------------------------- #
def test_a_successful_turn_survives_post_processing_blowing_up() -> None:
"""The whole point: the client gets the answer the provider produced."""
upstream = _upstream(
[
{"type": "thinking", "thinking": "reasoning", "signature": "sig-1"},
{"type": "text", "text": "here is the answer"},
]
)
resp = _run(upstream, break_post_processing=True)
assert resp.status_code == 200, resp.text
assert "text/event-stream" in resp.headers["content-type"]
# The provider's content reaches the client...
assert "here is the answer" in resp.text
assert "message_start" in resp.text
# ...and no invented failure does.
assert "api_error" not in resp.text
def test_a_client_tool_call_is_salvaged_too() -> None:
"""The captured failure was a `bash` tool_use turn with no retrieve call."""
upstream = _upstream(
[
{"type": "thinking", "thinking": "plan", "signature": "sig-2"},
{
"type": "tool_use",
"id": "toolu_bash",
"name": "bash",
"input": {"command": "ls"},
},
],
stop_reason="tool_use",
)
resp = _run(upstream, break_post_processing=True)
assert resp.status_code == 200, resp.text
assert "toolu_bash" in resp.text
assert "api_error" not in resp.text
def test_the_healthy_path_is_untouched() -> None:
"""Salvage must not change a turn that never failed."""
upstream = _upstream([{"type": "text", "text": "ordinary answer"}])
resp = _run(upstream, break_post_processing=False)
assert resp.status_code == 200, resp.text
assert "ordinary answer" in resp.text
assert "api_error" not in resp.text
# --------------------------------------------------------------------------- #
# What must never be salvaged
# --------------------------------------------------------------------------- #
def test_an_unresolved_retrieve_call_is_not_relayed() -> None:
"""Failing closed here is deliberate and stays that way.
The buffered path exists to resolve ``headroom_retrieve`` server-side. A
response still carrying one is precisely the case the handler already fails
closed on — relaying it would hand the client a tool call it is not expected
to service and a marker nobody expanded.
"""
app = create_app(_config())
with TestClient(app) as client:
proxy = client.app.state.proxy
unresolved = _upstream(
[
{
"type": "tool_use",
"id": "toolu_ccr",
"name": "headroom_retrieve",
"input": {"hash_key": "deadbeefcafe"},
}
],
stop_reason="tool_use",
)
assert proxy._can_salvage_buffered_upstream(unresolved) is False
# An ordinary turn is salvageable, so the guard is not simply off.
assert (
proxy._can_salvage_buffered_upstream(_upstream([{"type": "text", "text": "hi"}]))
is True
)
@pytest.mark.parametrize("bad", [None, "not-a-dict", 42, []])
def test_a_non_dict_response_is_never_salvaged(bad) -> None: # type: ignore[no-untyped-def]
app = create_app(_config())
with TestClient(app) as client:
assert client.app.state.proxy._can_salvage_buffered_upstream(bad) is False