scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-10-15 04:18:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

feat(llm): record request/response sizes, time to first token, finish reason and rate-limit headers per attempt

Browse source
This commit is contained in:
Ahmed Allam 2026-09-22 20:47:52 +00:00
parent 432edf0df6
commit bb4955be76
2 changed files with 548 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

305
strix/llm/request_log.py
View file

@ -19,6 +19,7 @@ from __future__ import annotations
import asyncio
import contextlib
import json
import logging
import re
import time
@ -32,9 +33,11 @@ from urllib.parse import urlsplit
from agents.items import ModelResponse
from agents.models.interface import Model
from agents.tool import FunctionTool
from agents.usage import Usage
from openai import APIError, APIStatusError
from openai.types.responses import ResponseCompletedEvent
from pydantic import BaseModel
if TYPE_CHECKING:
@ -92,6 +95,14 @@ class LlmRequestEvent:
agent_id: str | None = None
agent_name: str | None = None
retry_attempt: int = 0
# Serialized JSON body sizes, not on-the-wire bytes.
request_bytes: int | None = None
response_bytes: int | None = None
# Streamed calls only.
time_to_first_token_ms: int | None = None
finish_reason: str | None = None
# ``x-ratelimit-*`` / ``anthropic-ratelimit-*`` / ``retry-after`` only, never other headers.
rate_limit: dict[str, str] | None = None
def to_dict(self) -> dict[str, Any]:
data = asdict(self)
@ -234,6 +245,52 @@ def request_id_from_text(text: str | None) -> str | None:
return match.group(1) if match else None
# Rate-limit telemetry a provider puts in response headers. Names are
# normalized to lowercase with hyphens before matching; LiteLLM stores them
# both as ``x_ratelimit_remaining_tokens`` and ``llm_provider-…`` keys.
_RATE_LIMIT_HEADER_PREFIXES: tuple[str, ...] = (
"x-ratelimit-",
"anthropic-ratelimit-",
"retry-after",
)
RATE_LIMIT_MAX_HEADERS = 24
FINISH_REASON_MAX_CHARS = 128
def rate_limit_from_headers(headers: Mapping[str, Any] | None) -> dict[str, str] | None:
"""Only the rate-limit headers, name-normalized. Never the whole header block."""
if not headers:
return None
picked: dict[str, str] = {}
for key, value in headers.items():
if isinstance(value, bool) or not isinstance(value, str | int | float):
continue
text = str(value).strip()
if not text:
continue
name = str(key).lower().removeprefix(_LITELLM_HEADER_PREFIX).replace("_", "-")
if not name.startswith(_RATE_LIMIT_HEADER_PREFIXES):
continue
picked.setdefault(name, text[:64])
if len(picked) >= RATE_LIMIT_MAX_HEADERS:
break
return picked or None
def json_size(value: object) -> int | None:
"""Byte length of ``value`` serialized as compact JSON; None when it cannot be serialized."""
if value is None:
return None
try:
if isinstance(value, BaseModel):
text = value.model_dump_json(exclude_none=True)
else:
text = json.dumps(value, default=str, separators=(",", ":"), ensure_ascii=False)
except Exception: # noqa: BLE001 - size is best-effort telemetry
return None
return len(text.encode("utf-8"))
def api_host(url: str | None) -> str | None:
"""Hostname of an endpoint URL. Never the path or query, which may hold keys."""
if not isinstance(url, str) or not url.strip():
@ -303,12 +360,38 @@ def _litellm_usage(response: object) -> dict[str, int | None]:
}
def _finish_reason(value: object) -> str | None:
text = _str_or_none(value)
return text[:FINISH_REASON_MAX_CHARS] if text else None
def _litellm_finish_reason(response: object) -> str | None:
choices = getattr(response, "choices", None)
if not isinstance(choices, list) or not choices:
return None
first: object = cast("list[object]", choices)[0]
return _finish_reason(getattr(first, "finish_reason", None))
def _litellm_request_size(kwargs: Mapping[str, Any]) -> int | None:
"""Size of the request as LiteLLM sent it: model, messages and the provider params."""
body: dict[str, Any] = {"model": kwargs.get("model")}
messages = kwargs.get("messages")
if messages is not None:
body["messages"] = messages
optional = _mapping(kwargs.get("optional_params"))
if optional:
body.update(optional)
return json_size(body)
def _exception_headers(exc: BaseException | None) -> Mapping[str, Any] | None:
"""Provider response headers LiteLLM keeps on a mapped exception.
Where they live depends on the provider adapter: ``litellm_response_headers``
for the HTTP-handler adapters (Anthropic), ``headers`` for the OpenAI SDK
adapters, and the wrapped ``response`` otherwise.
adapters, and the wrapped ``response`` otherwise. The first block that
names a request id wins; failing that, the first non-empty block.
"""
if exc is None:
return None
@ -317,11 +400,25 @@ def _exception_headers(exc: BaseException | None) -> Mapping[str, Any] | None:
getattr(exc, "headers", None),
getattr(getattr(exc, "response", None), "headers", None),
)
first: Mapping[str, Any] | None = None
for raw in candidates:
headers = _headers_mapping(raw)
if headers and request_id_from_headers(headers):
if not headers:
continue
if request_id_from_headers(headers):
return headers
return None
first = first or headers
return first
def _exception_body_size(exc: BaseException) -> int | None:
body = getattr(exc, "body", None)
size = json_size(body) if body is not None else None
if size is not None:
return size
text = getattr(exc, "message", None)
text = text if isinstance(text, str) and text else str(exc)
return len(text.encode("utf-8")) if text else None
def _headers_mapping(headers: object) -> Mapping[str, Any] | None:
@ -368,45 +465,23 @@ def event_from_litellm(
response_headers = _mapping(hidden.get("additional_headers"))
response_hidden = _mapping(getattr(response, "_hidden_params", None)) or {}
hidden_headers = _mapping(response_hidden.get("additional_headers"))
request_id = request_id_from_headers(response_headers) or request_id_from_headers(
_mapping(response_hidden.get("additional_headers"))
hidden_headers
)
rate_limit = rate_limit_from_headers(response_headers) or rate_limit_from_headers(
hidden_headers
)
status_code: int | None = None
error_type: str | None = None
error_message: str | None = None
response_id: str | None = None
usage: dict[str, int | None] = {}
cost: float | None = None
if outcome == "success":
status_code = 200
response_id = _str_or_none(getattr(response, "id", None))
usage = _litellm_usage(response)
cost = _float_or_none(kwargs.get("response_cost"))
if cost is None:
cost = _float_or_none(hidden.get("response_cost"))
else:
if exc is not None:
status_code = _int_or_none(getattr(exc, "status_code", None))
error_type = type(exc).__name__
error_message = clean_error_message(exc)
request_id = request_id or request_id_from_headers(_exception_headers(exc))
if status_code is None:
code = error_info.get("error_code")
status_code = _int_or_none(code)
if status_code is None and isinstance(code, str) and code.isdigit():
status_code = int(code)
error_type = error_type or _str_or_none(error_info.get("error_class")) or "Exception"
if error_message is None:
raw = _str_or_none(error_info.get("error_message")) or _str_or_none(
slo.get("error_str")
)
error_message = redact(raw)[:ERROR_MESSAGE_MAX_CHARS] if raw else error_type
request_id = request_id or request_id_from_text(error_message)
ttft_ms: int | None = None
if streaming:
first = kwargs.get("completion_start_time")
if first is not None:
first_at = _to_datetime(first, started)
ttft_ms = max(0, int((first_at - started).total_seconds() * 1000))
context = current_call_context()
return LlmRequestEvent(
base = LlmRequestEvent(
call_id=call_id,
route="litellm",
provider=provider,
@ -414,22 +489,85 @@ def event_from_litellm(
api_host=host,
streaming=streaming,
outcome=outcome,
status_code=status_code,
status_code=None,
provider_request_id=request_id,
response_id=response_id,
error_type=error_type,
error_message=error_message,
response_id=None,
error_type=None,
error_message=None,
started_at=started,
finished_at=finished,
duration_ms=max(0, int((finished - started).total_seconds() * 1000)),
agent_id=context.agent_id,
agent_name=context.agent_name,
retry_attempt=context.retry_attempt,
request_bytes=_litellm_request_size(kwargs),
time_to_first_token_ms=ttft_ms,
rate_limit=rate_limit,
)
if outcome == "success":
return _litellm_success(base, kwargs, response, hidden)
return _litellm_failure(base, exc, error_info, slo)
def _litellm_success(
base: LlmRequestEvent, kwargs: Mapping[str, Any], response: object, hidden: Mapping[str, Any]
) -> LlmRequestEvent:
usage = _litellm_usage(response)
cost = _float_or_none(kwargs.get("response_cost"))
if cost is None:
cost = _float_or_none(hidden.get("response_cost"))
return replace(
base,
status_code=200,
response_id=_str_or_none(getattr(response, "id", None)),
input_tokens=usage.get("input_tokens"),
output_tokens=usage.get("output_tokens"),
cached_input_tokens=usage.get("cached_input_tokens"),
total_tokens=usage.get("total_tokens"),
cost_usd=cost,
agent_id=context.agent_id,
agent_name=context.agent_name,
retry_attempt=context.retry_attempt,
response_bytes=json_size(response),
finish_reason=_litellm_finish_reason(response),
)
def _litellm_failure(
base: LlmRequestEvent,
exc: BaseException | None,
error_info: Mapping[str, Any],
slo: Mapping[str, Any],
) -> LlmRequestEvent:
status_code: int | None = None
error_type: str | None = None
error_message: str | None = None
request_id = base.provider_request_id
rate_limit = base.rate_limit
response_bytes: int | None = None
if exc is not None:
status_code = _int_or_none(getattr(exc, "status_code", None))
error_type = type(exc).__name__
error_message = clean_error_message(exc)
exc_headers = _exception_headers(exc)
request_id = request_id or request_id_from_headers(exc_headers)
rate_limit = rate_limit or rate_limit_from_headers(exc_headers)
response_bytes = _exception_body_size(exc)
if status_code is None:
code = error_info.get("error_code")
status_code = _int_or_none(code)
if status_code is None and isinstance(code, str) and code.isdigit():
status_code = int(code)
error_type = error_type or _str_or_none(error_info.get("error_class")) or "Exception"
if error_message is None:
raw = _str_or_none(error_info.get("error_message")) or _str_or_none(slo.get("error_str"))
error_message = redact(raw)[:ERROR_MESSAGE_MAX_CHARS] if raw else error_type
request_id = request_id or request_id_from_text(error_message)
return replace(
base,
status_code=status_code,
provider_request_id=request_id,
error_type=error_type,
error_message=error_message,
response_bytes=response_bytes,
rate_limit=rate_limit,
)
@ -494,7 +632,8 @@ def _log_line_sink(event: LlmRequestEvent) -> None:
logger.log(
level,
"llm_request route=%s provider=%s model=%s host=%s outcome=%s status=%s "
"request_id=%s response_id=%s stream=%s duration_ms=%d "
"request_id=%s response_id=%s stream=%s duration_ms=%d ttft_ms=%s "
"req_bytes=%s res_bytes=%s finish=%s "
"in=%s out=%s cached=%s cost=%s agent=%s attempt=%d%s",
event.route,
event.provider or "-",
@ -506,6 +645,10 @@ def _log_line_sink(event: LlmRequestEvent) -> None:
event.response_id or "-",
"1" if event.streaming else "0",
event.duration_ms,
event.time_to_first_token_ms if event.time_to_first_token_ms is not None else "-",
event.request_bytes if event.request_bytes is not None else "-",
event.response_bytes if event.response_bytes is not None else "-",
event.finish_reason or "-",
event.input_tokens if event.input_tokens is not None else "-",
event.output_tokens if event.output_tokens is not None else "-",
event.cached_input_tokens if event.cached_input_tokens is not None else "-",
@ -529,12 +672,22 @@ def _openai_error_fields(exc: BaseException) -> tuple[int | None, str | None]:
status = _int_or_none(exc.status_code)
request_id = _str_or_none(exc.request_id)
if request_id is None:
request_id = request_id_from_headers(_headers_mapping(exc.response.headers))
request_id = request_id_from_headers(_openai_error_headers(exc))
if request_id is None:
request_id = request_id_from_text(str(exc))
return status, request_id
def _openai_error_headers(exc: BaseException) -> Mapping[str, Any] | None:
if not isinstance(exc, APIStatusError):
return None
headers = exc.response.headers
try:
return dict(headers.items())
except Exception: # noqa: BLE001 - httpx.Headers, or a test double without items()
return _headers_mapping(headers)
def _openai_usage(response: ModelResponse) -> dict[str, int | None]:
usage = response.usage
details = usage.input_tokens_details
@ -579,9 +732,17 @@ class RequestLoggingModel(Model):
streaming: bool,
response: ModelResponse | None,
exc: BaseException | None,
request_bytes: int | None = None,
first_event_mono: float | None = None,
finish_reason: str | None = None,
) -> LlmRequestEvent:
finished = datetime.now(UTC)
duration_ms = max(0, int((time.monotonic() - started_mono) * 1000))
ttft_ms = (
max(0, int((first_event_mono - started_mono) * 1000))
if streaming and first_event_mono is not None
else None
)
context = current_call_context()
base = LlmRequestEvent(
call_id=str(uuid.uuid4()),
@ -602,6 +763,9 @@ class RequestLoggingModel(Model):
agent_id=context.agent_id,
agent_name=context.agent_name,
retry_attempt=context.retry_attempt,
request_bytes=request_bytes,
time_to_first_token_ms=ttft_ms,
finish_reason=finish_reason,
)
if exc is None:
usage = _openai_usage(response) if response is not None else {}
@ -613,6 +777,7 @@ class RequestLoggingModel(Model):
output_tokens=usage.get("output_tokens"),
cached_input_tokens=usage.get("cached_input_tokens"),
total_tokens=usage.get("total_tokens"),
response_bytes=json_size(response.output) if response is not None else None,
)
status, request_id = _openai_error_fields(exc)
return replace(
@ -622,6 +787,28 @@ class RequestLoggingModel(Model):
provider_request_id=request_id,
error_type=type(exc).__name__,
error_message=clean_error_message(exc),
response_bytes=_exception_body_size(exc),
rate_limit=rate_limit_from_headers(_openai_error_headers(exc)),
)
@staticmethod
def _request_size(
system_instructions: str | None,
input: object, # noqa: A002
tools: list[Tool],
) -> int | None:
serialized_tools = [
{
"name": tool.name,
"description": tool.description,
"parameters": tool.params_json_schema,
}
if isinstance(tool, FunctionTool)
else {"name": tool.name}
for tool in tools
]
return json_size(
{"instructions": system_instructions, "input": input, "tools": serialized_tools}
)
async def get_response(
@ -639,6 +826,7 @@ class RequestLoggingModel(Model):
prompt: ResponsePromptParam | None,
) -> ModelResponse:
started, started_mono = datetime.now(UTC), time.monotonic()
request_bytes = self._request_size(system_instructions, input, tools)
try:
response = await self._inner.get_response(
system_instructions,
@ -662,6 +850,7 @@ class RequestLoggingModel(Model):
streaming=False,
response=None,
exc=exc,
request_bytes=request_bytes,
)
)
raise
@ -672,6 +861,7 @@ class RequestLoggingModel(Model):
streaming=False,
response=response,
exc=None,
request_bytes=request_bytes,
)
)
return response
@ -691,7 +881,10 @@ class RequestLoggingModel(Model):
prompt: ResponsePromptParam | None,
) -> AsyncIterator[TResponseStreamEvent]:
started, started_mono = datetime.now(UTC), time.monotonic()
request_bytes = self._request_size(system_instructions, input, tools)
completed: ModelResponse | None = None
first_event_mono: float | None = None
finish_reason: str | None = None
try:
async for event in self._inner.stream_response(
system_instructions,
@ -705,7 +898,10 @@ class RequestLoggingModel(Model):
conversation_id=conversation_id,
prompt=prompt,
):
if first_event_mono is None:
first_event_mono = time.monotonic()
if isinstance(event, ResponseCompletedEvent):
finish_reason = _openai_finish_reason(event)
raw_usage = event.response.usage
usage = Usage()
if raw_usage is not None:
@ -729,6 +925,8 @@ class RequestLoggingModel(Model):
streaming=True,
response=None,
exc=exc,
request_bytes=request_bytes,
first_event_mono=first_event_mono,
)
)
raise
@ -739,10 +937,23 @@ class RequestLoggingModel(Model):
streaming=True,
response=completed,
exc=None,
request_bytes=request_bytes,
first_event_mono=first_event_mono,
finish_reason=finish_reason,
)
)
def _openai_finish_reason(event: ResponseCompletedEvent) -> str | None:
"""``completed`` / ``incomplete:<reason>`` from a Responses API terminal event."""
status = _str_or_none(event.response.status)
details = event.response.incomplete_details
reason = details.reason if details is not None else None
if status == "incomplete" and reason:
return _finish_reason(f"incomplete:{reason}")
return _finish_reason(status)
def failure_text(exc: BaseException) -> str:
"""The message stored as an agent's failure reason, with the provider request id appended.
@ -774,6 +985,8 @@ __all__ = [
"event_from_litellm",
"failure_text",
"install",
"json_size",
"rate_limit_from_headers",
"redact",
"register_sink",
"request_id_from_headers",

289
tests/test_llm_request_log.py
View file

@ -10,10 +10,12 @@ import litellm
import pytest
from agents.items import ModelResponse
from agents.models.interface import Model
from agents.tool import FunctionTool
from agents.usage import Usage
from litellm.llms.anthropic.common_utils import AnthropicError
from openai import APIStatusError, APITimeoutError, PermissionDeniedError
from openai.types.responses import Response, ResponseCompletedEvent, ResponseCreatedEvent
from openai.types.responses.response import IncompleteDetails
from strix.llm import request_log
@ -412,6 +414,166 @@ def test_log_line_sink_formats_without_content(caplog: pytest.LogCaptureFixture)
assert "SECRET PROMPT" not in line
# --------------------------------------------------------------------------- #
# sizes, timing, finish reason, rate-limit headers #
# --------------------------------------------------------------------------- #
def test_rate_limit_headers_keep_only_rate_limit_names() -> None:
picked = request_log.rate_limit_from_headers(
{
"llm_provider-anthropic-ratelimit-requests-remaining": "49",
"Anthropic-RateLimit-Tokens-Reset": "2026-09-19T12:00:00Z",
"x-ratelimit-limit-requests": 5000,
"Retry-After": "12",
"authorization": "Bearer sk-ant-secret",
"x-api-key": "sk-ant-secret",
"set-cookie": "session=abc",
"request-id": "req_x",
"content-type": "application/json",
}
)
assert picked == {
"anthropic-ratelimit-requests-remaining": "49",
"anthropic-ratelimit-tokens-reset": "2026-09-19T12:00:00Z",
"x-ratelimit-limit-requests": "5000",
"retry-after": "12",
}
assert "secret" not in str(picked)
assert request_log.rate_limit_from_headers({"content-type": "json"}) is None
assert request_log.rate_limit_from_headers(None) is None
def test_rate_limit_headers_are_bounded() -> None:
headers = {f"x-ratelimit-h{i}": "v" * 500 for i in range(100)}
picked = request_log.rate_limit_from_headers(headers)
assert picked is not None
assert len(picked) == request_log.RATE_LIMIT_MAX_HEADERS
assert all(len(v) <= 64 for v in picked.values())
def test_json_size_counts_utf8_bytes_of_compact_json() -> None:
assert request_log.json_size({"a": "é"}) == len('{"a":"é"}'.encode())
assert request_log.json_size(None) is None
assert request_log.json_size(_openai_response("r")) is not None
assert request_log.json_size(object()) is not None # default=str fallback
def test_litellm_success_carries_sizes_finish_reason_and_rate_limit() -> None:
kwargs = _anthropic_kwargs(
None,
headers={
"request-id": "req_ok",
"anthropic-ratelimit-requests-remaining": "49",
"anthropic-ratelimit-tokens-remaining": "39000",
"x-api-key": "sk-ant-secret",
},
)
kwargs["optional_params"] = {"max_tokens": 4096, "temperature": 0}
class _Choice:
finish_reason = "tool_calls"
class _Response(_FakeResponse):
choices: ClassVar[list[Any]] = [_Choice()]
event = request_log.event_from_litellm(kwargs, _Response(), None, None, outcome="success")
expected_request = request_log.json_size(
{
"model": "anthropic/claude-sonnet-4-5",
"messages": [{"role": "user", "content": "SECRET PROMPT"}],
"max_tokens": 4096,
"temperature": 0,
}
)
assert event.request_bytes == expected_request
assert event.response_bytes is not None and event.response_bytes > 0
assert event.finish_reason == "tool_calls"
assert event.rate_limit == {
"anthropic-ratelimit-requests-remaining": "49",
"anthropic-ratelimit-tokens-remaining": "39000",
}
assert event.time_to_first_token_ms is None
assert "sk-ant-secret" not in str(event.to_dict())
assert "SECRET PROMPT" not in str(event.to_dict())
def test_litellm_streaming_time_to_first_token_from_completion_start() -> None:
kwargs = _anthropic_kwargs(None, stream=True)
start = datetime(2026, 9, 19, 12, 0, tzinfo=UTC)
kwargs["completion_start_time"] = start + timedelta(milliseconds=420)
end = start + timedelta(seconds=3)
event = request_log.event_from_litellm(kwargs, _FakeResponse(), start, end, outcome="success")
assert event.streaming is True
assert event.time_to_first_token_ms == 420
assert event.duration_ms == 3000
no_first = request_log.event_from_litellm(
_anthropic_kwargs(None, stream=True), _FakeResponse(), start, end, outcome="success"
)
assert no_first.time_to_first_token_ms is None
def test_litellm_failure_carries_status_body_size_and_rate_limit_from_exception() -> None:
body = '{"type":"error","error":{"type":"rate_limit_error","message":"slow down"}}'
exc = _anthropic_error(
429,
body,
{
"request-id": "req_429",
"retry-after": "7",
"anthropic-ratelimit-requests-remaining": "0",
"x-api-key": "sk-ant-secret",
},
)
kwargs = _anthropic_kwargs(exc)
kwargs["optional_params"] = {"max_tokens": 10}
event = request_log.event_from_litellm(kwargs, None, None, None, outcome="error")
assert event.status_code == 429
assert event.provider_request_id == "req_429"
assert event.request_bytes is not None and event.request_bytes > 0
assert event.response_bytes == len(body.encode())
assert event.rate_limit == {
"retry-after": "7",
"anthropic-ratelimit-requests-remaining": "0",
}
assert event.finish_reason is None
assert "sk-ant-secret" not in str(event.to_dict())
def test_litellm_failure_without_headers_has_no_rate_limit() -> None:
exc = APITimeoutError(httpx.Request("POST", "https://api.anthropic.com/v1/messages"))
kwargs = _anthropic_kwargs(exc)
kwargs["standard_logging_object"]["error_information"]["error_code"] = ""
event = request_log.event_from_litellm(kwargs, None, None, None, outcome="error")
assert event.rate_limit is None
assert event.response_bytes is not None # size of the timeout message text
def test_to_dict_and_log_line_include_new_fields() -> None:
kwargs = _anthropic_kwargs(None, headers={"request-id": "req_ok"}, stream=True)
start = datetime(2026, 9, 19, 12, 0, tzinfo=UTC)
kwargs["completion_start_time"] = start + timedelta(milliseconds=100)
event = request_log.event_from_litellm(
kwargs, _FakeResponse(), start, start + timedelta(seconds=1), outcome="success"
)
data = event.to_dict()
for key in (
"request_bytes",
"response_bytes",
"time_to_first_token_ms",
"finish_reason",
"rate_limit",
):
assert key in data
assert data["time_to_first_token_ms"] == 100
# --------------------------------------------------------------------------- #
# Native OpenAI route #
# --------------------------------------------------------------------------- #
@ -608,6 +770,133 @@ async def test_openai_route_streaming_failure_midstream(captured: list[LlmReques
assert captured[0].outcome == "error"
assert captured[0].status_code == 500
assert captured[0].provider_request_id == "req_mid_stream"
assert captured[0].time_to_first_token_ms is not None
assert captured[0].response_bytes == len(b"upstream reset")
@pytest.mark.asyncio
async def test_openai_route_success_carries_request_and_response_sizes(
captured: list[LlmRequestEvent],
) -> None:
tool = FunctionTool(
name="lookup",
description="Look something up",
params_json_schema={"type": "object", "properties": {"q": {"type": "string"}}},
on_invoke_tool=_noop_tool,
)
model = request_log.RequestLoggingModel(
_Inner(response=_openai_response("resp_sized")),
model_name="gpt-5",
provider="openai",
base_url=None,
)
args = list(_CALL_ARGS)
args[0] = "SYSTEM SECRET INSTRUCTIONS"
args[3] = [tool]
await model.get_response(*args, **_CALL_KWARGS)
event = captured[0]
expected = request_log.json_size(
{
"instructions": "SYSTEM SECRET INSTRUCTIONS",
"input": "hi",
"tools": [
{
"name": "lookup",
"description": "Look something up",
"parameters": tool.params_json_schema,
}
],
}
)
assert event.request_bytes == expected
assert event.response_bytes == request_log.json_size([])
assert event.time_to_first_token_ms is None
assert event.finish_reason is None
assert event.rate_limit is None
assert "SYSTEM SECRET INSTRUCTIONS" not in str(event.to_dict())
@pytest.mark.asyncio
async def test_openai_route_error_carries_rate_limit_headers_and_body_size(
captured: list[LlmRequestEvent],
) -> None:
request = httpx.Request("POST", "https://api.openai.com/v1/responses")
body = '{"error":{"message":"Rate limit reached","type":"tokens"}}'
response = httpx.Response(
429,
request=request,
headers={
"x-request-id": "req_429_openai",
"x-ratelimit-limit-tokens": "30000",
"x-ratelimit-remaining-tokens": "0",
"x-ratelimit-reset-tokens": "6ms",
"retry-after": "1",
"openai-organization": "org-secret",
},
text=body,
)
exc = APIStatusError("rate limited", response=response, body=None)
model = request_log.RequestLoggingModel(
_Inner(exc=exc), model_name="gpt-5", provider="openai", base_url=None
)
with pytest.raises(APIStatusError):
await model.get_response(*_CALL_ARGS, **_CALL_KWARGS)
event = captured[0]
assert event.status_code == 429
assert event.provider_request_id == "req_429_openai"
assert event.rate_limit == {
"x-ratelimit-limit-tokens": "30000",
"x-ratelimit-remaining-tokens": "0",
"x-ratelimit-reset-tokens": "6ms",
"retry-after": "1",
}
assert event.response_bytes is not None and event.response_bytes > 0
assert "org-secret" not in str(event.to_dict())
@pytest.mark.asyncio
async def test_openai_route_streaming_ttft_and_finish_reason(
captured: list[LlmRequestEvent],
) -> None:
created = ResponseCreatedEvent(
response=_completed_event("resp_fin").response,
sequence_number=0,
type="response.created",
)
completed = _completed_event("resp_fin")
completed.response.status = "incomplete"
completed.response.incomplete_details = IncompleteDetails(reason="max_output_tokens")
model = request_log.RequestLoggingModel(
_Inner(stream_events=[created, completed]),
model_name="gpt-5",
provider="openai",
base_url=None,
)
async for _ in model.stream_response(*_CALL_ARGS, **_CALL_KWARGS):
pass
event = captured[0]
assert event.streaming is True
assert event.time_to_first_token_ms is not None
assert event.time_to_first_token_ms <= event.duration_ms
assert event.finish_reason == "incomplete:max_output_tokens"
captured.clear()
plain = request_log.RequestLoggingModel(
_Inner(stream_events=[created, _completed_event("resp_done")]),
model_name="gpt-5",
provider="openai",
base_url=None,
)
async for _ in plain.stream_response(*_CALL_ARGS, **_CALL_KWARGS):
pass
assert captured[0].finish_reason is None or captured[0].finish_reason == "completed"
async def _noop_tool(_ctx: Any, _args: str) -> str:
return ""
@pytest.mark.asyncio