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

fix(proxy/openai): run response hooks on Responses, and bill their re-drives (#2872)

Browse Source 
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>
This commit is contained in:
Tejas Chopra
2026-08-09 10:09:24 -07:00
committed by GitHub
parent 91d6bf33cd
commit 675d13f08d
2 changed files with 590 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
headroom/proxy/handlers/openai.py
+274 -5
View File
@@ -187,6 +187,101 @@ def _header_get(headers: dict[str, str], name: str) -> str | None:
return None
#: Usage field names per OpenAI surface. Same three quantities, two spellings.
CHAT_USAGE_KEYS = {
"input_key": "prompt_tokens",
"output_key": "completion_tokens",
"details_key": "prompt_tokens_details",
}
RESPONSES_USAGE_KEYS = {
"input_key": "input_tokens",
"output_key": "output_tokens",
"details_key": "input_tokens_details",
}
class TurnHookUsage:
"""Upstream calls a turn hook caused that nothing else will account for.
A hook that re-drives the model (``call_model``) makes real, billed requests.
The handler's usage block reads 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 no
surface records.
The protocol is therefore two-sided, and both halves are required:
* :meth:`record` every upstream response as it arrives, original included.
* :meth:`settle` with the response the usage block will read.
``settle`` removes that one response's contribution, so what remains is
exactly the delta the handler must add. Recording only the re-drives and
adding them unconditionally — the first version of this — double-counted the
re-drive the handler had just promoted to ``response`` and dropped the
original entirely: one re-drive billed ``B + B`` instead of ``A + B``.
Matching is by object identity, because the handler hands back the very
object it recorded. A hook that synthesises a brand new response matches
nothing and nothing is subtracted, which over-counts rather than under —
the safe direction for a bill.
"""
__slots__ = ("_seen", "input_tokens", "output_tokens", "cache_read_tokens", "extra_calls")
def __init__(self) -> None:
self._seen: list[tuple[int, Any, int, int, int]] = []
self.input_tokens = 0
self.output_tokens = 0
self.cache_read_tokens = 0
self.extra_calls = 0
def record(
self,
payload: Any,
*,
input_key: str,
output_key: str,
details_key: str,
) -> None:
"""Note one upstream response. Never raises: a hook must not be able to
500 a request by returning an odd shape."""
usage = payload.get("usage") if isinstance(payload, dict) else None
def _int(value: Any) -> int:
try:
return max(int(value), 0)
except (TypeError, ValueError):
return 0
if isinstance(usage, dict):
details = usage.get(details_key)
cached = _int(details.get("cached_tokens")) if isinstance(details, dict) else 0
entry = (
id(payload),
payload,
_int(usage.get(input_key)),
_int(usage.get(output_key)),
cached,
)
else:
entry = (id(payload), payload, 0, 0, 0)
self._seen.append(entry)
def settle(self, final: Any) -> None:
"""Total everything except ``final``, which the usage block will read."""
self.input_tokens = self.output_tokens = self.cache_read_tokens = 0
self.extra_calls = 0
dropped = False
for _ident, payload, tin, tout, cached in self._seen:
if not dropped and payload is final:
dropped = True
continue
self.extra_calls += 1
self.input_tokens += tin
self.output_tokens += tout
self.cache_read_tokens += cached
def _sanitize_forwarded_response_headers(
headers: httpx.Headers | dict[str, str],
*extra_names: str,
@@ -2404,7 +2499,19 @@ class OpenAIHandlerMixin:
tools=working.get("tools"),
config=getattr(self, "config", None),
)
run_request_hooks(_req_ctx)
# Streaming turns get fold-only hooks, same rule as the
# chat-completions path. A hook that defers work to `on_response`
# cannot run here: the response side is wired on the buffered branch
# only, so on a real stream the shrink would land with no reload and
# the model's injected tool call would be streamed straight to a
# client that has no such tool. `stream` is not a parameter of this
# method, but the payload it is compressing carries the flag.
#
# 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 buffered-by-CCR turn misses a
# saving, never a stranded tool call.
run_request_hooks(_req_ctx, stream_safe_only=bool(payload.get("stream")))
if _req_ctx.tools is not working.get("tools"):
working["tools"] = _req_ctx.tools
# A hook may also fold the messages (replace or in-place). Write back a
@@ -3917,6 +4024,14 @@ class OpenAIHandlerMixin:
# Turn hooks (opt-in extensions) may inspect the turn
# or re-drive the model before we hand back the
# response. Inert when no hook is registered.
#
# Known gap: unlike the two HTTP paths, a re-drive
# here is NOT folded into this request's token
# accounting — `api_call_fn` goes through the custom
# backend, whose usage is recorded elsewhere. Wire a
# TurnHookUsage through `send_openai_message` before
# relying on cost numbers from a backend deployment
# that runs re-driving hooks.
from headroom.proxy.turn_hooks import (
TurnContext,
run_response_hooks,
@@ -4138,12 +4253,20 @@ class OpenAIHandlerMixin:
run_response_hooks,
)
# Tokens the hook's own re-drives cost. Stays at zero unless a
# hook actually calls the model again.
_hook_usage = TurnHookUsage()
if _registered_turn_hooks() and response.status_code == 200:
try:
_hook_resp_json = response.json()
except (ValueError, json.JSONDecodeError):
_hook_resp_json = None
if isinstance(_hook_resp_json, dict):
# The call we already made counts too. If the hook
# replaces the response, this original is the one nobody
# else will read.
_hook_usage.record(_hook_resp_json, **CHAT_USAGE_KEYS)
_hook_ctx = _TurnContext(
provider="openai",
model=str(model),
@@ -4154,12 +4277,31 @@ class OpenAIHandlerMixin:
async def _hook_call_model(_msgs):
body["messages"] = _msgs
if _hook_ctx.tools is not None:
body["tools"] = _hook_ctx.tools
_r = await self._retry_request("POST", url, headers, body)
return _r.json()
_r_json = _r.json()
_hook_usage.record(_r_json, **CHAT_USAGE_KEYS)
return _r_json
_hook_final = await run_response_hooks(
_hook_ctx, _hook_resp_json, _hook_call_model
)
# Same restore as the Responses path: the re-drive rewrote
# body so the next upstream call carried the hook's turn,
# but everything below is accounting for the request the
# client made, not the proxy's internal detour.
_hook_body_messages = body.get("messages")
_hook_body_tools = body.get("tools")
try:
_hook_final = await run_response_hooks(
_hook_ctx, _hook_resp_json, _hook_call_model
)
finally:
if _hook_body_messages is not None:
body["messages"] = _hook_body_messages
if _hook_body_tools is not None:
body["tools"] = _hook_body_tools
# Drop whichever response the usage block below reads;
# what is left is the spend nothing else records.
_hook_usage.settle(_hook_final)
if _hook_final is not _hook_resp_json:
response = httpx.Response(
status_code=200,
@@ -4320,6 +4462,20 @@ class OpenAIHandlerMixin:
f"[{request_id}] Failed to extract cached tokens from OpenAI response: {e}"
)
# Add what the hook's re-drives cost — see the matching block on
# the Responses path. 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.
if _hook_usage.extra_calls:
total_input_tokens += _hook_usage.input_tokens
output_tokens += _hook_usage.output_tokens
cache_read_tokens += _hook_usage.cache_read_tokens
logger.debug(
f"[{request_id}] turn hook: {_hook_usage.extra_calls} unaccounted call(s): "
f"+{_hook_usage.input_tokens} in / "
f"+{_hook_usage.output_tokens} out"
)
# Update prefix cache tracker for next turn
cache_write_tokens = _infer_openai_cache_write_tokens(
total_input_tokens,
@@ -5226,6 +5382,104 @@ class OpenAIHandlerMixin:
status_code=response.status_code,
metadata={"stream": stream, "auth_mode": auth_mode.value},
)
# Turn hooks, response side. `_compress_openai_responses_payload`
# already runs `run_request_hooks` for this surface, so without
# this block a hook could shrink a Responses turn and then never
# be asked to resolve what the model did about it — a deferral
# with no reload, which strands the model's call at a client
# that has no such tool. Mirrors the chat-completions wiring in
# `handle_openai_chat`; buffered path only, for the same reason
# CCR forces `stream:false` above: you cannot re-drive a turn
# whose bytes are already flowing.
from headroom.proxy.turn_hooks import (
TurnContext as _RespTurnContext,
)
from headroom.proxy.turn_hooks import (
registered_turn_hooks as _resp_registered_hooks,
)
from headroom.proxy.turn_hooks import (
run_response_hooks as _run_resp_hooks,
)
# Tokens the hook's own re-drives cost. Stays at zero unless a
# hook actually calls the model again.
_resp_hook_usage = TurnHookUsage()
if _resp_registered_hooks() and response.status_code == 200:
try:
_resp_hook_json = response.json()
except (ValueError, json.JSONDecodeError):
_resp_hook_json = None
if isinstance(_resp_hook_json, dict):
# The Responses API names the turn's items `input`;
# fall back to `messages` so an OpenAI-compatible
# upstream that accepts either still round-trips.
_resp_hook_usage.record(_resp_hook_json, **RESPONSES_USAGE_KEYS)
_resp_key = "input" if body.get("input") is not None else "messages"
_resp_hook_ctx = _RespTurnContext(
provider="openai",
model=str(model),
messages=body.get(_resp_key) or [],
tools=body.get("tools"),
config=self.config,
)
async def _resp_hook_call_model(
_items: list[dict[str, Any]],
_key: str = _resp_key,
) -> dict[str, Any]:
# Re-drive with the hook's items. Tools may also
# have grown (a reload makes resolved schemas
# callable), so send those back too.
body[_key] = _items
if _resp_hook_ctx.tools is not None:
body["tools"] = _resp_hook_ctx.tools
_rr = await self._retry_request(
"POST",
url,
headers,
body,
request_id=request_id,
forwarder_name="openai_responses_turn_hook",
path_for_log=url,
)
_rr_json = _rr.json()
_resp_hook_usage.record(_rr_json, **RESPONSES_USAGE_KEYS)
return _rr_json
# A re-drive rewrites body[input]/body[tools] so the
# next upstream call carries the hook's items. Put
# the client's turn back afterwards: everything below
# — CCR's `_responses_input_to_items(body["input"])`,
# usage accounting, observability — is describing the
# request the client actually made, not the proxy's
# internal detour. Without this, a turn that both
# reloaded a tool and hit CCR retrieval would hand
# CCR the hook's synthetic items.
_resp_body_input = body.get(_resp_key)
_resp_body_tools = body.get("tools")
try:
_resp_hook_final = await _run_resp_hooks(
_resp_hook_ctx, _resp_hook_json, _resp_hook_call_model
)
finally:
if _resp_body_input is not None:
body[_resp_key] = _resp_body_input
if _resp_body_tools is not None:
body["tools"] = _resp_body_tools
# Drop whichever response the usage block below reads.
_resp_hook_usage.settle(_resp_hook_final)
if _resp_hook_final is not _resp_hook_json:
response = httpx.Response(
status_code=200,
headers={
k: v
for k, v in response.headers.items()
if k.lower() not in ("content-encoding", "content-length")
},
content=json.dumps(_resp_hook_final).encode(),
)
total_latency = (time.time() - start_time) * 1000
total_input_tokens = original_tokens # fallback
@@ -5254,6 +5508,21 @@ class OpenAIHandlerMixin:
f"[{request_id}] Failed to extract cached tokens from OpenAI passthrough response: {e}"
)
# Add what the hook's re-drives cost. The usage read above
# describes the last upstream call; a hook that re-drove the
# model made earlier ones that were just as billed. Leaving
# them out lets a token-saving feature hide its own overhead,
# so cost and savings both read better than they are.
if _resp_hook_usage.extra_calls:
total_input_tokens += _resp_hook_usage.input_tokens
output_tokens += _resp_hook_usage.output_tokens
cache_read_tokens += _resp_hook_usage.cache_read_tokens
logger.debug(
f"[{request_id}] turn hook: {_resp_hook_usage.extra_calls} unaccounted call(s): "
f"+{_resp_hook_usage.input_tokens} in / "
f"+{_resp_hook_usage.output_tokens} out"
)
# CCR Response Handling: intercept headroom_retrieve tool
# calls server-side so a Responses API function_call the
# downstream caller can't resolve (e.g. Strands, or a
tests/test_turn_hook_usage.py
+316
View File
@@ -0,0 +1,316 @@
"""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