headroom/tests/test_proxy_anthropic_compression_diagnostics.py
chopratejas f0dcc02775 fix: A3 — byte-faithful Python forwarders; serialize canonical only when mutated
Eliminates P0-2 universally. Every Python forwarder (server.py
`_retry_request`, handlers/streaming.py `_stream_response`,
handlers/openai.py `_ws_http_fallback`, handlers/batch.py `_batch_passthrough`
+ batch-create + Google batch passthrough, handlers/anthropic.py CCR
continuation + batch endpoint) now switches from `httpx ... json=body` to
`httpx ... content=raw_bytes`. The default httpx JSON encoder was
re-serializing every request with `, `/`: ` separators and `\\uXXXX` ASCII
escapes — collapsing Anthropic prompt-cache hit-rate.

Forwarder strategy:
  - unmutated body → forward `await request.body()` verbatim;
  - mutated body  → re-serialize once via the new
    `serialize_body_canonical(body) -> bytes` helper (compact separators,
    `ensure_ascii=False`, dict insertion order preserved).

`HEADROOM_PROXY_PYTHON_FORWARDER_MODE` env var configures the mode:
  - `byte_faithful` (default) — the new behavior;
  - `legacy_json_kwarg` — explicit operator opt-in for emergency rollback.
Documented in `docs/content/docs/configuration.mdx`. NOT a fallback —
unknown values raise loudly per build constraint #4.

`BodyMutationTracker` accompanies each request through the handler so
transform sites mark the tracker (`memory_injection`,
`image_compression`, `compression_*`, `batch_compression`,
`ccr_continuation`, etc.). At forwarder dispatch we additionally compare
the final body dict against the parsed original bytes as a structural
safety net — any silent mutation we missed still triggers canonical
re-serialization.

A2 follow-up: `handlers/openai.py:534-540` (Chat Completions memory
injection) was prepending a system message; replaced with
`append_text_to_latest_user_chat_message`, the OpenAI Chat Completions
analog of `_append_context_to_latest_non_frozen_user_turn`. The cache
hot zone (system messages) is now sacrosanct on /v1/chat/completions
too. Honors `HEADROOM_MEMORY_INJECTION_MODE=disabled`.

Structured logging: every forwarder emits an `event=outbound_request`
log line with `forwarder`, `path`, `body_bytes`, `body_mutated`,
`mutation_reasons`, `source` (passthrough|canonical|legacy),
`request_id`. Never logs Authorization or full body.

`_read_request_json` factored to share `_read_request_body_bytes` with
new `read_request_json_with_bytes` so the anthropic handler can capture
both the parsed dict and the original (decompressed) bytes.

Tests:
  - `tests/test_proxy_byte_faithful_forwarding.py` (28 tests):
    SHA-256 byte-equality on /v1/messages and streaming, unicode
    preservation, numeric precision, mutation-tracker invariants,
    canonical-serializer properties, legacy-mode rollback, OpenAI
    Chat memory routing.
  - Existing test mocks updated to accept the new `**kwargs` on
    `_retry_request` (no behavior change).
  - `tests/test_proxy_handlers_batch.py` updated to read the captured
    `content=` bytes (formerly `json=`).
  - One A2 test corrected (`test_anthropic_tool_sort_and_context_append_helpers`)
    to match the live-zone-tail semantics introduced by A2.

Constraints satisfied: configurable env var; no new regex / hardcodes;
no silent fallback (`legacy_json_kwarg` is operator opt-in);
performant (`prepare_outbound_body_bytes` is O(1) for passthrough);
elegant single-responsibility helpers; structured tracing logs.
2026-05-02 09:02:10 -07:00

153 lines
5.1 KiB
Python

"""Diagnostics for Anthropic compression-stage observability (issue #296).
These tests verify two diagnostic improvements that let bug reports
distinguish a real pipeline failure from a thread-pool starvation timeout:
1. ``request_id`` is plumbed into ``pipeline.apply`` so its log lines
("Pipeline: freezing first ...", "Pipeline complete: ...") can be
correlated with a specific request rather than guessed at from
interleaved concurrent logs.
2. When ``compression_first_stage`` raises, the warning includes the
exception type — ``str(asyncio.TimeoutError())`` is empty, which is
why issue #296 shows ``Optimization failed:`` with nothing after the
colon.
"""
from __future__ import annotations
import asyncio
from types import SimpleNamespace
from unittest.mock import patch
import httpx
import pytest
pytest.importorskip("fastapi")
from fastapi.testclient import TestClient
from headroom.proxy.server import ProxyConfig, create_app
def _make_proxy_client() -> TestClient:
config = ProxyConfig(
optimize=True,
mode="token",
cache_enabled=False,
rate_limit_enabled=False,
cost_tracking_enabled=False,
log_requests=False,
ccr_inject_tool=False,
ccr_handle_responses=False,
ccr_context_tracking=False,
image_optimize=False,
)
app = create_app(config)
return TestClient(app)
def _ok_response(msg_id: str) -> httpx.Response:
return httpx.Response(
200,
json={
"id": msg_id,
"type": "message",
"role": "assistant",
"content": [{"type": "text", "text": "ok"}],
"usage": {
"input_tokens": 10,
"output_tokens": 3,
"cache_read_input_tokens": 0,
"cache_creation_input_tokens": 0,
},
},
)
def test_request_id_plumbed_to_pipeline_apply() -> None:
"""The handler must pass request_id into pipeline.apply so the
pipeline's log lines can be correlated with a specific request."""
captured: dict[str, object] = {}
with _make_proxy_client() as client:
proxy = client.app.state.proxy
def _fake_apply(**kwargs):
captured["request_id"] = kwargs.get("request_id")
return SimpleNamespace(
messages=kwargs["messages"],
transforms_applied=[],
timing={},
tokens_before=10,
tokens_after=10,
waste_signals=None,
)
proxy.anthropic_pipeline.apply = _fake_apply
async def _fake_retry(method, url, headers, body, stream=False, **kwargs): # noqa: ANN001
return _ok_response("msg_diag_1")
proxy._retry_request = _fake_retry
response = client.post(
"/v1/messages",
headers={"x-api-key": "test-key", "anthropic-version": "2023-06-01"},
json={
"model": "claude-sonnet-4-6",
"max_tokens": 64,
"messages": [{"role": "user", "content": "hello"}],
},
)
assert response.status_code == 200
assert isinstance(captured["request_id"], str)
assert captured["request_id"] # non-empty
def test_optimization_failure_logs_exception_type() -> None:
"""When pipeline.apply raises, the warning must include the
exception type — issue #296 reported ``Optimization failed:`` with
an empty message because asyncio.TimeoutError has no str repr.
We patch the handler module's ``logger.warning`` directly rather than
relying on logging propagation: the headroom logger sets
``propagate=False`` (see proxy/helpers.py) and per-test mutations of
handler chains have proven brittle in CI.
"""
from headroom.proxy.handlers import anthropic as anth_handler
with _make_proxy_client() as client:
proxy = client.app.state.proxy
def _raise_timeout(**kwargs):
raise asyncio.TimeoutError()
proxy.anthropic_pipeline.apply = _raise_timeout
async def _fake_retry(method, url, headers, body, stream=False, **kwargs): # noqa: ANN001
return _ok_response("msg_diag_2")
proxy._retry_request = _fake_retry
with patch.object(anth_handler.logger, "warning") as mock_warning:
response = client.post(
"/v1/messages",
headers={"x-api-key": "test-key", "anthropic-version": "2023-06-01"},
json={
"model": "claude-sonnet-4-6",
"max_tokens": 64,
"messages": [{"role": "user", "content": "hello"}],
},
)
assert response.status_code == 200, response.text
warning_msgs = [
call.args[0]
for call in mock_warning.call_args_list
if call.args and "Optimization failed" in str(call.args[0])
]
assert warning_msgs, (
f"expected an 'Optimization failed' warning, got calls: {mock_warning.call_args_list!r}"
)
msg = warning_msgs[0]
assert "TimeoutError" in msg, f"expected exception type in warning, got: {msg!r}"