headroom/tests/test_turn_hook_usage.py
Tejas Chopra 675d13f08d
fix(proxy/openai): run response hooks on Responses, and bill their re-drives (#2872)
The Responses path runs `run_request_hooks` but never
`run_response_hooks` —
only `handle_openai_chat` does. So a turn hook can shrink a Responses
turn and
then never be asked to resolve what the model did about it: the model's
injected
tool call goes straight to a client that has no such tool.

That asymmetry is why tool-belt deferral has to be disabled wholesale on
the
Responses API, which is the surface Codex uses.

## 1. Wire the response side

Mirrors the chat-completions block. **Buffered path only**, for the same
reason
CCR already forces `stream:false` when it needs to intercept: you cannot
re-drive a turn whose bytes are already flowing.

## 2. Honour `stream_safe_only` on the Responses request path

It was the one hook call site that ignored the flag. A re-driving hook
would run
its shrink on a streamed turn and then have no response side to finish
it —
latent until (1) lands, live afterwards.

`stream` is not a parameter of `_compress_openai_responses_payload`, but
the
payload it is compressing carries the flag. It is read **before** CCR
may force
`stream:false` further down, so this is the client's request rather than
the
effective one — conservative in the safe direction: at worst a
CCR-buffered turn
misses a saving, never a stranded tool call.

Fold-only hooks that declare `stream_safe = True` are unaffected.

## 3. Bill what the re-drives cost

Both handlers read usage from the **final** upstream response, so every
intermediate call a hook made was free as far as Headroom was concerned.

For a token-saving feature that is not a rounding error. A tool-search
reload is
a whole extra model call; counting only the last one lets the feature
hide its
own overhead behind the saving it is claiming, and the numbers come out
better
than the truth.

`TurnHookUsage` accumulates input/output/cached across re-drives; both
HTTP
paths fold it into their totals. The two surfaces report the same three
quantities under different names (`prompt_tokens` vs `input_tokens`), so
the key
pair is passed in.

Expect measured cost to go **up** and savings percentage to go **down**
on any
deployment running a re-driving hook. That is the correction, not a
regression.

## Also: restore the body after the hooks

A re-drive rewrites `body[input]` / `body[messages]` / `body[tools]` so
the next
upstream call carries the hook's turn. Everything downstream — CCR's
`_responses_input_to_items(body["input"])`, usage accounting,
observability — is
describing the request the *client* made, not the proxy's internal
detour.

Without the restore, a turn that both reloaded a tool and hit CCR
retrieval
hands CCR the proxy's synthetic items. The chat path had the same leak
(`body["messages"]` stayed rewritten); both are fixed the same way.

## Known gap

A re-drive on the custom backend path (`send_openai_message`) is still
not
folded into that request's accounting — its usage is recorded elsewhere.
Commented at the call site rather than silently skipped.

## Blast radius

**Inert unless a turn hook is registered**, so no behaviour change for a
stock
OSS proxy. `TurnHookUsage` starts at zero and stays there on every path
that
does not re-drive.

## Verification

- `tests/test_turn_hook_usage.py` — 5 new tests: per-surface key names,
accumulation across rounds, negative counts floored not subtracted, and
that
an unreadable shape still counts the call (a silent zero there looks
exactly
  like "the hook cost nothing")
- 434 passing across `turn_hook`, `extension`, `tool_search`,
`responses` and
  `openai_chat` suites
- `ruff check` + `ruff format` clean

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

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-09 10:09:24 -07:00

316 lines
11 KiB
Python

"""A turn hook's re-drives are billed calls and must reach token accounting.
A hook that resolves an injected tool call re-drives the model. Both OpenAI
handlers read usage from exactly ONE response — the original, or whichever the
hook returned in its place, because the handler swaps `response` for it. Every
other upstream call on that turn is spend nothing else records.
Getting that wrong is not a rounding error for a token-saving feature: it lets
the feature hide its own overhead behind the saving it claims. The first version
of this recorded only the re-drives and added them unconditionally, so a single
re-drive billed `B + B` and dropped the original `A` entirely. The handler tests
at the bottom are what catch that class of mistake; the unit tests above them
cannot, because the bug lives in how the accumulator composes with the response
swap rather than in the accumulator itself.
"""
from __future__ import annotations
import json
from typing import Any
import httpx
import pytest
import respx
from headroom.proxy.handlers.openai import (
CHAT_USAGE_KEYS,
RESPONSES_USAGE_KEYS,
TurnHookUsage,
)
pytest.importorskip("fastapi")
from fastapi.testclient import TestClient # noqa: E402
from headroom.proxy.loopback_guard import require_loopback # noqa: E402
from headroom.proxy.server import ProxyConfig, create_app # noqa: E402
from headroom.proxy.turn_hooks import clear_turn_hooks, register_turn_hook # noqa: E402
# --- unit: the accumulator -----------------------------------------------
def _chat(prompt: int, completion: int, cached: int = 0) -> dict[str, Any]:
return {
"usage": {
"prompt_tokens": prompt,
"completion_tokens": completion,
"prompt_tokens_details": {"cached_tokens": cached},
}
}
def test_no_redrive_adds_nothing() -> None:
"""The common path: one upstream call, which the usage block reads itself."""
u = TurnHookUsage()
original = _chat(100, 10)
u.record(original, **CHAT_USAGE_KEYS)
u.settle(original)
assert u.extra_calls == 0
assert (u.input_tokens, u.output_tokens, u.cache_read_tokens) == (0, 0, 0)
def test_one_redrive_leaves_the_original_to_add() -> None:
"""A + B billed; the block will read B; so A is the delta."""
u = TurnHookUsage()
a, b = _chat(100, 10, 60), _chat(150, 20, 90)
u.record(a, **CHAT_USAGE_KEYS)
u.record(b, **CHAT_USAGE_KEYS)
u.settle(b)
assert u.extra_calls == 1
assert (u.input_tokens, u.output_tokens, u.cache_read_tokens) == (100, 10, 60)
def test_two_redrives_leave_the_original_and_the_middle() -> None:
u = TurnHookUsage()
a, b, c = _chat(100, 10), _chat(150, 20), _chat(200, 30)
for r in (a, b, c):
u.record(r, **CHAT_USAGE_KEYS)
u.settle(c)
assert u.extra_calls == 2
assert (u.input_tokens, u.output_tokens) == (250, 30)
def test_hook_that_keeps_the_original_still_pays_for_the_redrive() -> None:
"""Re-drove, then returned the original anyway. B was still billed."""
u = TurnHookUsage()
a, b = _chat(100, 10), _chat(150, 20)
u.record(a, **CHAT_USAGE_KEYS)
u.record(b, **CHAT_USAGE_KEYS)
u.settle(a)
assert u.extra_calls == 1
assert (u.input_tokens, u.output_tokens) == (150, 20)
def test_synthesised_response_matches_nothing_and_over_counts() -> None:
"""Nothing is subtracted when the hook invents a response. Over-counting is
the safe direction for a bill; under-counting is the bug this file exists
for."""
u = TurnHookUsage()
a, b = _chat(100, 10), _chat(150, 20)
u.record(a, **CHAT_USAGE_KEYS)
u.record(b, **CHAT_USAGE_KEYS)
u.settle({"usage": {"prompt_tokens": 999}})
assert u.extra_calls == 2
assert u.input_tokens == 250
def test_responses_shape_uses_its_own_key_names() -> None:
u = TurnHookUsage()
a = {
"usage": {
"input_tokens": 400,
"output_tokens": 40,
"input_tokens_details": {"cached_tokens": 300},
}
}
b = {"usage": {"input_tokens": 500, "output_tokens": 50}}
u.record(a, **RESPONSES_USAGE_KEYS)
u.record(b, **RESPONSES_USAGE_KEYS)
u.settle(b)
assert (u.input_tokens, u.output_tokens, u.cache_read_tokens) == (400, 40, 300)
# Chat keys must not read a Responses payload: a silent 0 looks exactly like
# "the hook cost nothing".
v = TurnHookUsage()
v.record(a, **CHAT_USAGE_KEYS)
v.record(b, **CHAT_USAGE_KEYS)
v.settle(b)
assert v.input_tokens == 0
assert v.extra_calls == 1, "the call still happened even if its shape was unreadable"
def test_never_raises_on_a_shape_it_does_not_recognise() -> None:
"""A hook must not be able to 500 a request by returning something odd."""
u = TurnHookUsage()
for payload in (
None,
{},
[],
"not a dict",
{"usage": None},
{"usage": "nope"},
{"usage": {"prompt_tokens": None, "completion_tokens": "x"}},
{"usage": {"prompt_tokens": -5, "prompt_tokens_details": "nope"}},
):
u.record(payload, **CHAT_USAGE_KEYS)
u.settle(object())
assert u.extra_calls == 8
assert (u.input_tokens, u.output_tokens, u.cache_read_tokens) == (0, 0, 0)
# --- handler level: what the unit tests above structurally cannot see -----
class _RedriveOnce:
"""Minimal hook: re-drive the model exactly once, return the new response."""
name = "test_redrive"
stream_safe = False
def __init__(self) -> None:
self.calls = 0
def on_request(self, ctx: Any) -> None: # pragma: no cover - nothing to do
return None
async def on_response(self, ctx: Any, response: Any, call_model: Any) -> Any:
if self.calls:
return None
self.calls += 1
return await call_model(ctx.messages)
@pytest.fixture
def _no_hooks():
clear_turn_hooks()
yield
clear_turn_hooks()
def _app_and_outcomes(monkeypatch):
"""App with a spy on the outcome record, which is where the billed token
counts land (`provider_input_tokens` / `output_tokens`)."""
app = create_app(
ProxyConfig(
optimize=False,
cache_enabled=False,
rate_limit_enabled=False,
cost_tracking_enabled=False,
log_requests=False,
)
)
app.dependency_overrides[require_loopback] = lambda: None
outcomes: list[Any] = []
proxy = app.state.proxy
# Patched on the type, so the bound-call self arrives as the first argument.
async def _spy(_self, outcome, *a, **kw):
outcomes.append(outcome)
monkeypatch.setattr(type(proxy), "_record_request_outcome", _spy, raising=True)
return app, outcomes
@respx.mock
def test_chat_bills_the_original_plus_the_redrive(monkeypatch, _no_hooks) -> None:
"""A=100/10, B=150/20 -> 250 in / 30 out.
The bug this pins reported 300/40 (B twice, A dropped).
"""
register_turn_hook(_RedriveOnce())
app, outcomes = _app_and_outcomes(monkeypatch)
bodies = [
{
"id": "a",
"choices": [
{"message": {"role": "assistant", "content": "A"}, "finish_reason": "stop"}
],
"usage": {"prompt_tokens": 100, "completion_tokens": 10},
},
{
"id": "b",
"choices": [
{"message": {"role": "assistant", "content": "B"}, "finish_reason": "stop"}
],
"usage": {"prompt_tokens": 150, "completion_tokens": 20},
},
]
sent = iter(bodies)
respx.post("https://api.openai.com/v1/chat/completions").mock(
side_effect=lambda request: httpx.Response(200, json=next(sent))
)
with TestClient(app) as client:
r = client.post(
"/v1/chat/completions",
json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]},
headers={"authorization": "Bearer sk-test"},
)
assert r.status_code == 200
assert json.loads(r.content)["id"] == "b", "the hook's response is what the client gets"
assert outcomes, "an outcome must be recorded"
o = outcomes[-1]
assert o.provider_input_tokens == 250, f"want A+B=250, got {o.provider_input_tokens}"
assert o.output_tokens == 30, f"want A+B=30, got {o.output_tokens}"
@respx.mock
def test_responses_bills_the_original_plus_the_redrive(monkeypatch, _no_hooks) -> None:
"""Same arithmetic on /v1/responses, whose usage keys differ."""
register_turn_hook(_RedriveOnce())
app, outcomes = _app_and_outcomes(monkeypatch)
bodies = [
{
"id": "a",
"output": [{"type": "message", "role": "assistant", "content": []}],
"usage": {"input_tokens": 400, "output_tokens": 40},
},
{
"id": "b",
"output": [{"type": "message", "role": "assistant", "content": []}],
"usage": {"input_tokens": 500, "output_tokens": 50},
},
]
sent = iter(bodies)
respx.post("https://api.openai.com/v1/responses").mock(
side_effect=lambda request: httpx.Response(200, json=next(sent))
)
with TestClient(app) as client:
r = client.post(
"/v1/responses",
json={
"model": "gpt-4o",
"input": [{"type": "message", "role": "user", "content": []}],
"stream": False,
},
headers={"authorization": "Bearer sk-test"},
)
assert r.status_code == 200
assert outcomes, "an outcome must be recorded"
o = outcomes[-1]
assert o.provider_input_tokens == 900, f"want A+B=900, got {o.provider_input_tokens}"
assert o.output_tokens == 90, f"want A+B=90, got {o.output_tokens}"
@respx.mock
def test_no_hook_registered_bills_exactly_the_one_call(monkeypatch, _no_hooks) -> None:
"""The regression guard in the other direction: with no hook, accounting must
be untouched — this whole mechanism has to be inert on a stock proxy."""
app, outcomes = _app_and_outcomes(monkeypatch)
respx.post("https://api.openai.com/v1/chat/completions").mock(
return_value=httpx.Response(
200,
json={
"id": "a",
"choices": [
{"message": {"role": "assistant", "content": "A"}, "finish_reason": "stop"}
],
"usage": {"prompt_tokens": 100, "completion_tokens": 10},
},
)
)
with TestClient(app) as client:
r = client.post(
"/v1/chat/completions",
json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]},
headers={"authorization": "Bearer sk-test"},
)
assert r.status_code == 200
o = outcomes[-1]
assert o.provider_input_tokens == 100
assert o.output_tokens == 10