akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(proxy): log compressed messages alongside original request (#261)

Browse source 
## Description

Expose the post-compression message list that was actually sent upstream
as a new `compressed_messages` field on `RequestLog`, paired with the
existing (now consistently pre-compression) `request_messages`.
Consumers of `/transformations/feed` — dashboards and any downstream
observability — can now diff the two sides of a compression to see
exactly what the pipeline stripped, replaced, or kept. Turns an abstract
"saved N tokens" into a legible before/after.

Gated by the same `log_full_messages` flag as `request_messages` so the
two sides stay in sync; it's pointless to store one without the other.

Also fixes a latent correctness bug: today's `request_messages` field is
inconsistent across the four `RequestLog` construction sites — sometimes
it's the pre-compression snapshot, sometimes it's the mutated
`body["messages"]` (which is the compressed list, because the proxy
mutates `body` in place before the log call). After this change,
`request_messages` always means pre-compression and
`compressed_messages` always means what went upstream.

## Type of Change

- [ ] Bug fix (non-breaking change that fixes an issue)
- [x] 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)

Note on "breaking": strictly speaking this is a semantic correction of
an inconsistently-populated field, not a schema break. The field name
`request_messages` is unchanged and the JSON shape is unchanged; what
changes is that the field now consistently holds the pre-compression
list. Consumers that treated it as "whatever messages we have" continue
to work. Consumers that depended on the accidental post-compression
value (if any existed) would shift to `compressed_messages`.

## Changes Made

- **`headroom/proxy/models.py`**: `RequestLog` gains
`compressed_messages: list[dict] | None = None`. Doc comment explains
it's paired with `request_messages` and gated by the same
`log_full_messages` flag.
- **`headroom/proxy/handlers/anthropic.py`** (2 sites — Bedrock
non-streaming and main non-streaming): `request_messages` now
consistently sources from `original_messages` (the pre-compression
snapshot at line 724), `compressed_messages` sources from
`body["messages"]` (the compressed list after in-place mutation at line
1189). Both gated symmetrically.
- **`headroom/proxy/handlers/streaming.py`** (2 sites — main streaming
in `_finalize_stream_response`, Bedrock streaming in
`_stream_response_bedrock`): same treatment. `_stream_response_bedrock`
gains a new `original_messages: list[dict] | None = None` parameter so
it has access to the pre-compression snapshot; the sole caller in
`anthropic.py` now threads it through.
- **`headroom/proxy/server.py`**: `/transformations/feed` adds
`compressed_messages` to the JSON payload alongside the existing
`request_messages` / `response_content`. *Split into a separate
preceding commit is a one-time EOL normalization to LF — the file blob
in history carries CRLF but `.gitattributes` declares `*.py text
eol=lf`, so any contributor editing `server.py` triggers the same
whole-file renormalization. Separating the two commits keeps this
feature commit's diff at a single line.*
- **`headroom/proxy/request_logger.py`**: `compressed_messages` is
stripped from the JSONL file log and from `get_recent()` alongside the
existing `request_messages` / `response_content` stripping.
`get_memory_stats()` also counts it toward the deque's byte budget.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .` and `ruff format --check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed (via the Headroom Desktop client that
consumes `/transformations/feed` — confirmed both fields arrive and
render)

Test coverage added/extended:

- `tests/test_proxy/test_request_logger.py` (new file): round-trip unit
tests for `RequestLogger`. Confirms `get_recent` strips both sides (pre
+ post), `get_recent_with_messages` exposes both, and the JSONL file log
drops both when `log_full_messages=False`.
- `tests/test_proxy/test_transformations_feed.py`: extended to assert
`compressed_messages` appears in the endpoint payload alongside
`request_messages` / `response_content`.
- `tests/test_proxy_streaming_request_logger.py`: existing include/omit
tests updated to assert both sides populate when the flag is on and both
are `None` when it's off.

## Test Output

```
$ uv run ruff check headroom tests
All checks passed!
$ uv run ruff format --check headroom tests
614 files already formatted
$ uv run pytest tests/test_proxy/test_request_logger.py tests/test_proxy_streaming_request_logger.py tests/test_proxy/test_transformations_feed.py -v
...
tests/test_proxy/test_request_logger.py::test_get_recent_strips_compressed_messages_alongside_request_and_response PASSED
tests/test_proxy/test_request_logger.py::test_get_recent_with_messages_returns_compressed_messages PASSED
tests/test_proxy/test_request_logger.py::test_jsonl_file_strips_both_sides_when_log_full_messages_disabled PASSED
tests/test_proxy_streaming_request_logger.py::test_finalize_stream_response_logs_original_and_compressed_messages PASSED
tests/test_proxy_streaming_request_logger.py::test_finalize_stream_response_omits_messages_when_log_full_messages_disabled PASSED
tests/test_proxy/test_transformations_feed.py::test_transformations_feed_returns_messages PASSED
...
============================== 11 passed in 5.36s ==============================
```

## 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
(the two-sided gating at each log site, the `_stream_response_bedrock`
parameter addition, and the `get_memory_stats` accounting)
- [ ] 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

## Additional Notes

### Non-Anthropic backends

`handlers/openai.py` and `handlers/gemini.py` do not currently emit
`RequestLog` entries at all — only Anthropic and the shared streaming
paths do. This PR therefore only populates `compressed_messages` on
Anthropic traffic (which is what `/transformations/feed` shows today).
Wiring OpenAI and Gemini into `RequestLogger` end-to-end is a separate,
larger gap worth its own PR.

### `server.py` EOL normalization

The feature change in `server.py` is a single line. To keep the diff
readable, the preceding commit is a whitespace-only `chore(proxy):
normalize server.py to LF per .gitattributes` — the file blob was stored
with CRLF terminators but `.gitattributes` declares `*.py text eol=lf`.
Any contributor touching `server.py` triggers this renormalization;
isolating it here keeps the feature commit reviewable. Happy to rebase /
drop / reshape as preferred.

### Downstream desktop compatibility

The Headroom Desktop client I work on now consumes `compressed_messages`
and renders the pre/post pair side-by-side on the "Recent large
compression" card. The desktop was updated to handle both shapes:
proxies without the field render the legacy single "Request" block;
proxies with the field render "Request (original, N tokens)" + "Request
(compressed, M tokens)" where N/M come from `input_tokens_original` /
`input_tokens_optimized`. No changes needed downstream if this PR lands.

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

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
gglucass 2026-06-12 02:02:54 +02:00 • committed by GitHub
parent b4395993ae
commit 2269e40bde
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 197 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

22
headroom/proxy/handlers/anthropic.py
View file

@ -1736,6 +1736,7 @@ class AnthropicHandlerMixin:
tags,
optimization_latency,
pipeline_timing=pipeline_timing,
original_messages=original_client_messages,
)
else:
async with stage_timer.measure("upstream_connect"):
@ -1837,7 +1838,15 @@ class AnthropicHandlerMixin:
turn_id=compute_turn_id(
model, body.get("system"), body.get("messages")
),
request_messages=body.get("messages")
# `original_client_messages` is the deep-copied
# pre-compression snapshot; `body["messages"]`
# is the compressed list sent upstream. Both
# share the `log_full_messages` gate so the two
# sides stay symmetric.
request_messages=original_client_messages
if self.config.log_full_messages
else None,
compressed_messages=body.get("messages")
if self.config.log_full_messages
else None,
)
@ -2358,7 +2367,16 @@ class AnthropicHandlerMixin:
turn_id=compute_turn_id(
model, body.get("system"), body.get("messages")
),
request_messages=messages if self.config.log_full_messages else None,
# `original_client_messages` is the deep-copied
# pre-compression snapshot; `body["messages"]` is the
# compressed list sent upstream. Both gated by
# `log_full_messages`.
request_messages=original_client_messages
if self.config.log_full_messages
else None,
compressed_messages=body.get("messages")
if self.config.log_full_messages
else None,
)
)

3
headroom/proxy/handlers/streaming.py
View file

@ -785,6 +785,7 @@ class StreamingMixin:
uncached_input_tokens=uncached_input_tokens,
ttfb_ms=stream_state["ttfb_ms"] or total_latency,
pipeline_timing=pipeline_timing,
original_messages=original_messages,
)
await self._record_request_outcome(outcome)
@ -1349,6 +1350,7 @@ class StreamingMixin:
tags: dict[str, str],
optimization_latency: float,
pipeline_timing: dict[str, float] | None = None,
original_messages: list[dict] | None = None,
) -> StreamingResponse:
"""Stream response from Bedrock backend with metrics tracking.
@ -1455,6 +1457,7 @@ class StreamingMixin:
cache_write_1h_tokens=stream_state["cache_creation_ephemeral_1h_input_tokens"],
ttfb_ms=stream_state["ttfb_ms"] or 0,
pipeline_timing=pipeline_timing,
original_messages=original_messages,
)
await self._record_request_outcome(outcome)

5
headroom/proxy/models.py
View file

@ -48,6 +48,11 @@ class RequestLog:
# Request/Response (optional, for debugging)
request_messages: list[dict] | None = None
# Messages after compression, as actually sent upstream. Paired with
# `request_messages` (the pre-compression snapshot) so consumers can diff
# the two sides of the compression. Governed by the same
# `log_full_messages` gate as `request_messages`.
compressed_messages: list[dict] | None = None
response_content: str | None = None
error: str | None = None

30
headroom/proxy/outcome.py
View file

@ -127,6 +127,12 @@ class RequestOutcome:
num_messages: int = 0
turn_id: str | None = None
request_messages: list[dict[str, Any]] | None = None
# Post-compression messages actually sent upstream, paired with
# ``request_messages`` (pre-compression) so consumers can diff the two.
# Only populated when a caller threads in the pre-compression snapshot
# (``original_messages``); otherwise ``request_messages`` carries the sent
# body for backward compatibility and this stays ``None``.
compressed_messages: list[dict[str, Any]] | None = None
tags: dict[str, str] = field(default_factory=dict)
client: str | None = None
project: str | None = None
@ -201,6 +207,7 @@ class RequestOutcome:
ttfb_ms: float = 0.0,
pipeline_timing: dict[str, float] | None = None,
waste_signals: dict[str, int] | None = None,
original_messages: list[dict] | None = None,
) -> RequestOutcome:
"""Construct an outcome from the locals available at streaming
finalize. Three streaming finalizers
@ -246,6 +253,25 @@ class RequestOutcome:
if system is None:
system = body.get("systemInstruction")
# ``request_items`` is ``body["messages"]`` (or ``body["contents"]``
# for Gemini, falling back to ``[]``) — the post-compression list the
# caller already mutated in place before finalize. When a
# caller threads in ``original_messages`` (the pre-compression
# snapshot), log it as ``request_messages`` and the sent body as
# ``compressed_messages`` so the two sides stay diffable. Callers that
# don't thread it in (gemini ``contents``, OpenAI-via-backend) keep the
# prior behaviour: sent body under ``request_messages``, no compressed
# side. Both sides share the ``log_full_messages`` gate.
if not log_full_messages:
log_request_messages = None
log_compressed_messages = None
elif original_messages is not None:
log_request_messages = original_messages
log_compressed_messages = request_items
else:
log_request_messages = request_items
log_compressed_messages = None
return cls(
request_id=request_id,
provider=provider,
@ -271,7 +297,8 @@ class RequestOutcome:
turn_id=compute_turn_id(model, system, turn_messages),
tags=tags or {},
client=client,
request_messages=request_items if log_full_messages else None,
request_messages=log_request_messages,
compressed_messages=log_compressed_messages,
)
@ -376,6 +403,7 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None:
transforms_applied=list(outcome.transforms_applied),
waste_signals=outcome.waste_signals,
request_messages=outcome.request_messages,
compressed_messages=outcome.compressed_messages,
turn_id=outcome.turn_id,
)
)

15
headroom/proxy/request_logger.py
View file

@ -212,9 +212,9 @@ class RequestLogger:
"""Log a request. Oldest entries are automatically removed when limit reached.
Phase G PR-G3 (P4-45): base64-encoded image payloads in
``request_messages`` / ``response_content`` are redacted
before write. Redaction also applies to the in-memory deque
so the ``/stats/recent_requests`` endpoint never serves a
``request_messages`` / ``compressed_messages`` / ``response_content``
are redacted before write. Redaction also applies to the in-memory
deque so the ``/stats/recent_requests`` endpoint never serves a
multi-MB image either.
"""
# Redact image payloads in-place on the deque entry so memory
@ -223,6 +223,8 @@ class RequestLogger:
# ``get_recent_with_messages`` unchanged.
if entry.request_messages is not None:
entry.request_messages = redact_image_base64(entry.request_messages)
if entry.compressed_messages is not None:
entry.compressed_messages = redact_image_base64(entry.compressed_messages)
if entry.response_content is not None:
entry.response_content = redact_image_base64(entry.response_content)
@ -234,20 +236,21 @@ class RequestLogger:
log_dict = asdict(entry)
if not self.log_full_messages:
log_dict.pop("request_messages", None)
log_dict.pop("compressed_messages", None)
log_dict.pop("response_content", None)
f.write(json.dumps(log_dict) + "\n")
except OSError:
pass # Graceful degradation: memory-only logging continues
def get_recent(self, n: int = 100) -> list[dict]:
"""Get recent log entries (without request_messages and response_content)."""
"""Get recent log entries (without request/compressed messages and response_content)."""
# Convert deque to list for slicing (deque doesn't support slicing)
entries = list(self._logs)[-n:]
return [
{
k: v
for k, v in asdict(e).items()
if k not in ("request_messages", "response_content")
if k not in ("request_messages", "compressed_messages", "response_content")
}
for e in entries
]
@ -289,6 +292,8 @@ class RequestLogger:
# Messages and response can be large
if log_entry.request_messages:
size_bytes += sys.getsizeof(log_entry.request_messages)
if log_entry.compressed_messages:
size_bytes += sys.getsizeof(log_entry.compressed_messages)
if log_entry.response_content:
size_bytes += len(log_entry.response_content)

1
headroom/proxy/server.py
View file

@ -2533,6 +2533,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
"savings_percent": log.get("savings_percent"),
"transforms_applied": log.get("transforms_applied", []),
"request_messages": log.get("request_messages"),
"compressed_messages": log.get("compressed_messages"),
"response_content": log.get("response_content"),
"turn_id": log.get("turn_id"),
}

100
tests/test_proxy/test_request_logger.py Normal file
View file

@ -0,0 +1,100 @@
"""Tests for the in-memory request logger.
Covers the `log_full_messages` gate, which controls whether the
pre-compression (`request_messages`) and post-compression
(`compressed_messages`) payloads persist past the in-memory entry onto disk.
Both sides are governed by the same flag so the two sides of the compression
stay in sync - it's pointless to store one without the other.
"""
from __future__ import annotations
from headroom.proxy.models import RequestLog
from headroom.proxy.request_logger import RequestLogger
def _entry(**overrides) -> RequestLog:
base: dict = {
"request_id": "r1",
"timestamp": "2026-04-24T10:00:00Z",
"provider": "anthropic",
"model": "claude-sonnet-4-6",
"input_tokens_original": 100,
"input_tokens_optimized": 40,
"output_tokens": 10,
"tokens_saved": 60,
"savings_percent": 60.0,
"optimization_latency_ms": 1.0,
"total_latency_ms": 20.0,
"tags": {},
"cache_hit": False,
"transforms_applied": ["kompress:user:0.4"],
}
base.update(overrides)
return RequestLog(**base)
def test_get_recent_strips_compressed_messages_alongside_request_and_response():
logger = RequestLogger(log_file=None, log_full_messages=True)
logger.log(
_entry(
request_messages=[{"role": "user", "content": "pre"}],
compressed_messages=[{"role": "user", "content": "post"}],
response_content="ok",
)
)
recent = logger.get_recent(10)
assert len(recent) == 1
assert "request_messages" not in recent[0]
assert "compressed_messages" not in recent[0]
assert "response_content" not in recent[0]
def test_get_recent_with_messages_returns_compressed_messages():
logger = RequestLogger(log_file=None, log_full_messages=True)
logger.log(
_entry(
request_messages=[{"role": "user", "content": "pre"}],
compressed_messages=[{"role": "user", "content": "post"}],
)
)
recent = logger.get_recent_with_messages(10)
assert len(recent) == 1
assert recent[0]["request_messages"] == [{"role": "user", "content": "pre"}]
assert recent[0]["compressed_messages"] == [{"role": "user", "content": "post"}]
def test_jsonl_file_strips_both_sides_when_log_full_messages_disabled(tmp_path):
log_file = tmp_path / "requests.jsonl"
logger = RequestLogger(log_file=str(log_file), log_full_messages=False)
logger.log(
_entry(
request_messages=[{"role": "user", "content": "pre"}],
compressed_messages=[{"role": "user", "content": "post"}],
response_content="ok",
)
)
import json
lines = log_file.read_text().strip().splitlines()
assert len(lines) == 1
obj = json.loads(lines[0])
assert "request_messages" not in obj
assert "compressed_messages" not in obj
assert "response_content" not in obj
def test_get_memory_stats_accounts_for_compressed_messages():
logger = RequestLogger(log_file=None)
logger.log(
_entry(
compressed_messages=[{"role": "user", "content": "post"}],
)
)
stats = logger.get_memory_stats()
assert stats.entry_count == 1
assert stats.size_bytes > 0

15
tests/test_proxy/test_transformations_feed.py
View file

@ -30,15 +30,24 @@ async def test_transformations_feed_endpoint_returns_list(app):
@pytest.mark.asyncio
async def test_transformations_feed_returns_messages(app):
"""Each transformation should include request_messages and response_content."""
"""Each transformation exposes both the original request and the
post-compression form that was actually sent upstream, plus the response.
The pre/post pair is what makes compression legible: consumers can diff
the two to see what the pipeline stripped, replaced, or kept.
"""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
response = await client.get("/transformations/feed")
data = response.json()
transformations = data["transformations"]
for t in transformations:
assert "request_messages" in t or t.get("request_messages") is None
assert "response_content" in t or t.get("response_content") is None
assert "request_messages" in t
assert t["request_messages"] is None or isinstance(t["request_messages"], list)
assert "compressed_messages" in t
assert t["compressed_messages"] is None or isinstance(t["compressed_messages"], list)
assert "response_content" in t
assert t["response_content"] is None or isinstance(t["response_content"], str)
@pytest.mark.asyncio

20
tests/test_proxy_streaming_request_logger.py
View file

@ -114,9 +114,17 @@ async def test_finalize_stream_response_logs_request_for_feed():
@pytest.mark.asyncio
async def test_finalize_stream_response_includes_messages_when_log_full_messages_enabled():
async def test_finalize_stream_response_logs_original_and_compressed_messages():
"""With log_full_messages enabled, both sides of the compression are
recorded: `request_messages` is the pre-compression snapshot the caller
threads in via `original_messages`, `compressed_messages` is what was
actually sent upstream (i.e. `body["messages"]` after in-place mutation)."""
proxy = _build_proxy_with_real_logger(log_full_messages=True)
body = {"messages": [{"role": "user", "content": "hello"}]}
# `body["messages"]` models the post-compression list - the proxy mutates
# `body` in place before calling `_finalize_stream_response`, so this is
# already what was shipped over the wire.
body = {"messages": [{"role": "user", "content": "[compressed]"}]}
original = [{"role": "user", "content": "[original, pre-compression]"}]
await proxy._finalize_stream_response(
body=body,
@ -130,11 +138,13 @@ async def test_finalize_stream_response_includes_messages_when_log_full_messages
optimization_latency=1.0,
stream_state=_stream_state(output_tokens=5),
start_time=0.0,
original_messages=original,
)
entries = proxy.logger.get_recent_with_messages(10)
assert len(entries) == 1
assert entries[0]["request_messages"] == body["messages"]
assert entries[0]["request_messages"] == original
assert entries[0]["compressed_messages"] == body["messages"]
@pytest.mark.asyncio
@ -153,11 +163,15 @@ async def test_finalize_stream_response_omits_messages_when_log_full_messages_di
optimization_latency=1.0,
stream_state=_stream_state(output_tokens=5),
start_time=0.0,
original_messages=[{"role": "user", "content": "dropped"}],
)
entries = proxy.logger.get_recent_with_messages(10)
assert len(entries) == 1
# Both sides share the same gate - neither leaks when log_full_messages
# is off.
assert entries[0]["request_messages"] is None
assert entries[0]["compressed_messages"] is None
@pytest.mark.asyncio