from __future__ import annotations import asyncio import json import logging from dataclasses import dataclass from datetime import UTC, datetime, timedelta from types import SimpleNamespace from typing import TYPE_CHECKING, Any, ClassVar, cast import httpx import litellm import pytest from agents.items import ModelResponse from agents.models import _openai_shared from agents.models.fake_id import FAKE_RESPONSES_ID from agents.models.interface import Model from agents.models.openai_provider import shared_http_client from agents.tool import FunctionTool from agents.usage import Usage from litellm.exceptions import APIConnectionError as LiteLlmConnectionError from litellm.exceptions import APIError as LiteLlmApiError from litellm.exceptions import AuthenticationError as LiteLlmAuthenticationError from litellm.exceptions import Timeout as LiteLlmTimeout 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.config import codex, models from strix.llm import request_log if TYPE_CHECKING: from collections.abc import AsyncGenerator, AsyncIterator, Iterator from strix.llm.request_log import LlmRequestEvent ANTHROPIC_BLOCK_BODY = ( '{"type":"error","error":{"type":"invalid_request_error","message":"Output blocked by ' 'content filtering policy"},"request_id":"req_011CVBodyOnly00"}' ) @pytest.fixture def captured() -> Iterator[list[LlmRequestEvent]]: events: list[LlmRequestEvent] = [] request_log.register_sink(events.append) try: yield events finally: request_log.unregister_sink(events.append) @pytest.fixture(autouse=True) def _reset_context() -> Iterator[None]: token = request_log.bind_call_context(None, None) try: yield finally: request_log.reset_call_context(token) def _anthropic_kwargs( exc: BaseException | None, *, headers: dict[str, str] | None = None, stream: bool = False, error_message: str | None = None, ) -> dict[str, Any]: slo: dict[str, Any] = { "model": "claude-sonnet-4-5", "custom_llm_provider": "anthropic", "api_base": "https://api.anthropic.com", "stream": stream, "litellm_call_id": "call-1", "hidden_params": { "additional_headers": {f"llm_provider-{k}": v for k, v in (headers or {}).items()}, "response_cost": 0.0123, }, } if exc is not None: slo["error_information"] = { "error_code": str(getattr(exc, "status_code", "")), "error_class": type(exc).__name__, "llm_provider": "anthropic", "error_message": error_message or str(exc), } kwargs: dict[str, Any] = { "model": "anthropic/claude-sonnet-4-5", "custom_llm_provider": "anthropic", "litellm_call_id": "call-1", "stream": stream, "standard_logging_object": slo, "litellm_params": {"api_base": "https://api.anthropic.com", "api_key": "sk-ant-secret"}, "messages": [{"role": "user", "content": "SECRET PROMPT"}], } if exc is not None: kwargs["exception"] = exc return kwargs @dataclass class _FakeUsage: prompt_tokens: int = 120 completion_tokens: int = 30 total_tokens: int = 150 prompt_tokens_details: None = None cache_read_input_tokens: int = 100 class _FakeResponse: id = "msg_01abc" usage = _FakeUsage() _hidden_params: ClassVar[dict[str, Any]] = {} choices: ClassVar[list[Any]] = [{"message": {"content": "SECRET COMPLETION"}}] def _anthropic_error( status: int, body: str, headers: dict[str, str] | None = None ) -> AnthropicError: return AnthropicError(status, body, headers=httpx.Headers(headers or {})) # --------------------------------------------------------------------------- # # request-id extraction # # --------------------------------------------------------------------------- # def test_header_request_id_accepts_raw_and_litellm_prefixed_names() -> None: assert request_log.request_id_from_headers({"Request-Id": "req_a"}) == "req_a" assert request_log.request_id_from_headers({"llm_provider-request-id": "req_b"}) == "req_b" assert request_log.request_id_from_headers({"x-request-id": "req_c"}) == "req_c" assert request_log.request_id_from_headers({"x-amzn-requestid": "abc-123"}) == "abc-123" assert request_log.request_id_from_headers({"content-type": "json"}) is None assert request_log.request_id_from_headers({"request-id": " "}) is None assert request_log.request_id_from_headers(None) is None def test_header_request_id_prefers_provider_id_over_cdn_ray() -> None: headers = {"cf-ray": "8f0-FRA", "request-id": "req_real"} assert request_log.request_id_from_headers(headers) == "req_real" def test_body_request_id_extraction() -> None: assert request_log.request_id_from_text(ANTHROPIC_BLOCK_BODY) == "req_011CVBodyOnly00" assert request_log.request_id_from_text("request_id=req_abcdef") == "req_abcdef" assert request_log.request_id_from_text("no id here") is None assert request_log.request_id_from_text(None) is None def test_api_host_never_leaks_path_or_query() -> None: assert request_log.api_host("https://gw.corp.example/v1?key=abc") == "gw.corp.example" assert request_log.api_host("gw.corp.example:8443/v1") == "gw.corp.example" assert request_log.api_host("") is None assert request_log.api_host(None) is None # --------------------------------------------------------------------------- # # error message # # --------------------------------------------------------------------------- # def test_clean_error_message_truncates_and_keeps_the_text() -> None: exc = RuntimeError("x" * 5000) message = request_log.clean_error_message(exc) assert len(message) <= request_log.ERROR_MESSAGE_MAX_CHARS assert message.endswith("…") assert request_log.clean_error_message(RuntimeError("upstream 401")) == "upstream 401" assert request_log.clean_error_message(RuntimeError("")) == "RuntimeError" # --------------------------------------------------------------------------- # # LiteLLM route # # --------------------------------------------------------------------------- # def test_litellm_success_event_carries_header_request_id_usage_and_no_content() -> None: kwargs = _anthropic_kwargs(None, headers={"request-id": "req_ok_123456"}) kwargs["response_cost"] = 0.0123 start = datetime(2026, 9, 19, 12, 0, tzinfo=UTC) end = start + timedelta(milliseconds=850) event = request_log.event_from_litellm(kwargs, _FakeResponse(), start, end, outcome="success") assert event.route == "litellm" assert event.outcome == "success" assert event.status_code == 200 assert event.provider == "anthropic" assert event.model == "anthropic/claude-sonnet-4-5" assert event.api_host == "api.anthropic.com" assert event.provider_request_id == "req_ok_123456" assert event.response_id == "msg_01abc" assert event.call_id == "call-1" assert event.duration_ms == 850 assert (event.input_tokens, event.output_tokens, event.total_tokens) == (120, 30, 150) assert event.cached_input_tokens == 100 assert event.cost_usd == pytest.approx(0.0123) assert event.error_type is None and event.error_message is None serialized = str(event.to_dict()) assert "SECRET PROMPT" not in serialized assert "SECRET COMPLETION" not in serialized assert "sk-ant-secret" not in serialized def test_litellm_success_falls_back_to_response_hidden_headers() -> None: kwargs = _anthropic_kwargs(None) class _Response(_FakeResponse): _hidden_params: ClassVar[dict[str, Any]] = { "additional_headers": {"llm_provider-request-id": "req_hidden_1"} } event = request_log.event_from_litellm(kwargs, _Response(), None, None, outcome="success") assert event.provider_request_id == "req_hidden_1" def test_litellm_anthropic_block_prefers_header_id_over_body_id() -> None: exc = _anthropic_error(400, ANTHROPIC_BLOCK_BODY, {"request-id": "req_HeaderWins01"}) kwargs = _anthropic_kwargs(exc, headers={"request-id": "req_HeaderWins01"}) event = request_log.event_from_litellm(kwargs, None, None, None, outcome="error") assert event.outcome == "error" assert event.status_code == 400 assert event.provider_request_id == "req_HeaderWins01" assert event.error_type == "AnthropicError" assert event.error_message is not None assert "content filtering policy" in event.error_message assert "req_011CVBodyOnly00" in event.error_message def test_litellm_anthropic_block_falls_back_to_body_id() -> None: exc = _anthropic_error(400, ANTHROPIC_BLOCK_BODY) kwargs = _anthropic_kwargs(exc) event = request_log.event_from_litellm(kwargs, None, None, None, outcome="error") assert event.provider_request_id == "req_011CVBodyOnly00" assert event.status_code == 400 def test_litellm_failure_uses_exception_headers_when_slo_has_none() -> None: exc = _anthropic_error(529, "overloaded", {"request-id": "req_from_exc_hdr"}) kwargs = _anthropic_kwargs(exc) event = request_log.event_from_litellm(kwargs, None, None, None, outcome="error") assert event.provider_request_id == "req_from_exc_hdr" assert event.status_code == 529 def test_litellm_failure_without_any_request_id_is_none_not_fabricated() -> None: exc = _anthropic_error(500, "internal error") event = request_log.event_from_litellm( _anthropic_kwargs(exc), None, None, None, outcome="error" ) assert event.provider_request_id is None assert event.error_message == "internal error" def test_litellm_failure_without_exception_object_uses_error_information() -> None: kwargs = _anthropic_kwargs(None) kwargs["standard_logging_object"]["error_information"] = { "error_code": "429", "error_class": "RateLimitError", "llm_provider": "anthropic", "error_message": 'rate limited "request_id": "req_slo_only01"', } event = request_log.event_from_litellm(kwargs, None, None, None, outcome="error") assert event.status_code == 429 assert event.error_type == "RateLimitError" assert event.provider_request_id == "req_slo_only01" def test_litellm_timeout_is_an_error_event_with_no_status() -> 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.outcome == "error" assert event.status_code is None assert event.error_type == "APITimeoutError" assert event.provider_request_id is None def test_litellm_failure_message_is_the_providers_text() -> None: exc = _anthropic_error(401, "invalid x-api-key") event = request_log.event_from_litellm( _anthropic_kwargs(exc), None, None, None, outcome="error" ) assert event.error_message is not None assert "invalid x-api-key" in event.error_message def test_litellm_streaming_flag_and_call_id() -> None: kwargs = _anthropic_kwargs(None, headers={"request-id": "req_stream1"}, stream=True) event = request_log.event_from_litellm(kwargs, _FakeResponse(), None, None, outcome="success") assert event.streaming is True assert event.provider_request_id == "req_stream1" exc = _anthropic_error(400, ANTHROPIC_BLOCK_BODY) failed = request_log.event_from_litellm( _anthropic_kwargs(exc, stream=True), None, None, None, outcome="error" ) assert failed.streaming is True assert failed.provider_request_id == "req_011CVBodyOnly00" def test_litellm_event_carries_bound_agent_context_and_retry_attempt() -> None: token = request_log.bind_call_context("agent-7", "Recon") try: request_log.set_retry_attempt(2) event = request_log.event_from_litellm( _anthropic_kwargs(None), _FakeResponse(), None, None, outcome="success" ) finally: request_log.reset_call_context(token) assert (event.agent_id, event.agent_name, event.retry_attempt) == ("agent-7", "Recon", 2) assert request_log.current_call_context().agent_id is None @pytest.mark.asyncio async def test_call_context_is_isolated_between_tasks() -> None: seen: dict[str, str | None] = {} async def run(agent_id: str) -> None: token = request_log.bind_call_context(agent_id, None) try: await asyncio.sleep(0) seen[agent_id] = request_log.current_call_context().agent_id finally: request_log.reset_call_context(token) await asyncio.gather(run("a"), run("b")) assert seen == {"a": "a", "b": "b"} assert request_log.current_call_context().agent_id is None @pytest.mark.asyncio async def test_litellm_logger_dispatches_and_isolates_sink_failures( captured: list[LlmRequestEvent], caplog: pytest.LogCaptureFixture ) -> None: def boom(_event: LlmRequestEvent) -> None: raise RuntimeError("sink down") request_log.register_sink(boom) try: logger = request_log._build_litellm_logger() with caplog.at_level(logging.ERROR, logger="strix.llm.request_log"): await logger.async_log_success_event( _anthropic_kwargs(None, headers={"request-id": "req_dispatch"}), _FakeResponse(), None, None, ) await logger.async_log_failure_event( _anthropic_kwargs(_anthropic_error(400, ANTHROPIC_BLOCK_BODY)), None, None, None ) finally: request_log.unregister_sink(boom) assert [e.outcome for e in captured] == ["success", "error"] assert captured[0].provider_request_id == "req_dispatch" assert captured[1].provider_request_id == "req_011CVBodyOnly00" assert sum("sink down" in r.getMessage() or "failed" in r.getMessage() for r in caplog.records) def test_dispatch_swallows_malformed_callback_payloads(captured: list[LlmRequestEvent]) -> None: request_log._dispatch(cast("Any", None), None, None, None, outcome="success") assert captured == [] def test_install_is_idempotent_and_registers_one_litellm_callback() -> None: request_log.install() request_log.install() ours = [cb for cb in litellm.callbacks if type(cb).__name__ == "_StrixRequestLogger"] assert len(ours) == 1 def test_log_line_sink_formats_without_content(caplog: pytest.LogCaptureFixture) -> None: exc = _anthropic_error(400, ANTHROPIC_BLOCK_BODY, {"request-id": "req_line01"}) event = request_log.event_from_litellm( _anthropic_kwargs(exc, headers={"request-id": "req_line01"}), None, None, None, outcome="error", ) with caplog.at_level(logging.DEBUG, logger="strix.llm.request_log"): request_log._log_line_sink(event) line = caplog.records[-1].getMessage() assert "request_id=req_line01" in line assert "status=400" in line assert "provider=anthropic" in line assert "SECRET PROMPT" not in line # --------------------------------------------------------------------------- # # sizes, timing, finish reason, free-form headers and details # # --------------------------------------------------------------------------- # def test_response_headers_keep_everything_as_sent() -> None: picked = request_log.headers_from_response( { "llm_provider-anthropic-ratelimit-requests-remaining": "49", "Anthropic-RateLimit-Tokens-Reset": "2026-09-19T12:00:00Z", "x-ratelimit-limit-requests": 5000, "Retry-After": "12", "request-id": "req_x", "content-type": "application/json", "cf-ray": "8f0-FRA", "x-should-retry": "false", "openai-processing-ms": "812", "WWW-Authenticate": "Bearer realm=x", "set-cookie": "__cf_bm=abc", "x-vendor-new-header": "kept without a code change", "x-empty": " ", "x-object": {"not": "a string"}, } ) assert picked == { "anthropic-ratelimit-requests-remaining": "49", "anthropic-ratelimit-tokens-reset": "2026-09-19T12:00:00Z", "x-ratelimit-limit-requests": "5000", "retry-after": "12", "request-id": "req_x", "content-type": "application/json", "cf-ray": "8f0-FRA", "x-should-retry": "false", "openai-processing-ms": "812", "www-authenticate": "Bearer realm=x", "set-cookie": "__cf_bm=abc", "x-vendor-new-header": "kept without a code change", } assert request_log.headers_from_response({}) is None assert request_log.headers_from_response(None) is None def test_response_headers_are_bounded() -> None: headers = {f"x-h{i}": "v" * 5000 for i in range(200)} picked = request_log.headers_from_response(headers) assert picked is not None assert len(picked) == request_log.HEADERS_MAX_COUNT assert all(len(v) <= request_log.HEADER_VALUE_MAX_CHARS for v in picked.values()) def test_details_keep_everything_as_written_and_bound_it() -> None: details = request_log.bound_details( { "max_tokens": 4096, "temperature": 0, "thinking": {"type": "enabled", "budget_tokens": 1024}, "tools": [{"name": "t", "input_schema": {}}], "extra_headers": {"x-title": "strix"}, "api_base": "https://gw.example/v1?tenant=abc", "usage": { "input_tokens": 10, "cache_creation_input_tokens": 3, "server_tool_use": {"web_search_requests": 1}, }, "note": "kept as written", "when": datetime(2026, 9, 19, tzinfo=UTC), "nested": {"a": {"b": {"c": {"d": {"e": {"f": {"g": 1}}}}}}}, "empty": {}, "none": None, } ) assert details == { "max_tokens": 4096, "temperature": 0, "thinking": {"type": "enabled", "budget_tokens": 1024}, "tools": [{"name": "t"}], "extra_headers": {"x-title": "strix"}, "api_base": "https://gw.example/v1?tenant=abc", "usage": { "input_tokens": 10, "cache_creation_input_tokens": 3, "server_tool_use": {"web_search_requests": 1}, }, "note": "kept as written", "when": "2026-09-19T00:00:00+00:00", "nested": {"a": {"b": {"c": {"d": {"e": "…"}}}}}, } assert request_log.bound_details({}) is None assert request_log.bound_details("not a mapping") is None assert request_log.bound_details({"messages": []}) is None def test_details_are_bounded_by_size_and_name_the_dropped_keys() -> None: big = {"small": 1, "huge": ["x" * 200] * 32, "medium": {"k": "y" * 200}} request_log.DETAILS_MAX_BYTES, saved = 1024, request_log.DETAILS_MAX_BYTES try: details = request_log.bound_details(big) finally: request_log.DETAILS_MAX_BYTES = saved assert details is not None assert details["small"] == 1 assert details["_dropped"] == ["huge"] assert "huge" not in details assert (request_log.json_size(details) or 0) <= 1024 def test_details_list_and_string_bounds() -> None: details = request_log.bound_details( {"items": list(range(100)), "long": "z" * 1000, "keys": {str(i): i for i in range(100)}} ) assert details is not None assert len(details["items"]) == request_log.DETAILS_MAX_ITEMS assert len(details["long"]) == request_log.DETAILS_MAX_STRING assert len(details["keys"]) == request_log.DETAILS_MAX_ITEMS 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_headers_and_details() -> None: kwargs = _anthropic_kwargs( None, headers={ "request-id": "req_ok", "anthropic-ratelimit-requests-remaining": "49", "anthropic-ratelimit-tokens-remaining": "39000", "anthropic-organization-id": "org-123", }, ) kwargs["optional_params"] = { "max_tokens": 4096, "temperature": 0, "tools": [{"name": "a"}, {"name": "b"}], "extra_headers": {"authorization": "Bearer x"}, } kwargs["standard_logging_object"]["hidden_params"]["model_id"] = "m-1" kwargs["standard_logging_object"]["cache_hit"] = False class _Choice: finish_reason = "tool_calls" provider_specific_fields: ClassVar[dict[str, Any]] = {"stop_sequence": None} class _Response(_FakeResponse): choices: ClassVar[list[Any]] = [_Choice()] system_fingerprint = "fp_1" 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, "tools": [{"name": "a"}, {"name": "b"}], "extra_headers": {"authorization": "Bearer x"}, } ) 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.response_headers == { "request-id": "req_ok", "anthropic-ratelimit-requests-remaining": "49", "anthropic-ratelimit-tokens-remaining": "39000", "anthropic-organization-id": "org-123", } assert event.details is not None assert event.details["request"] == { "max_tokens": 4096, "temperature": 0, "tools": [{"name": "a"}, {"name": "b"}], "extra_headers": {"authorization": "Bearer x"}, "message_count": 1, "tool_count": 2, } assert event.details["response"]["id"] == "msg_01abc" assert event.details["response"]["system_fingerprint"] == "fp_1" assert event.details["response"]["usage"]["cache_read_input_tokens"] == 100 assert event.details["response"]["choice"] == {"finish_reason": "tool_calls"} assert event.details["response"]["choice_count"] == 1 assert event.details["litellm"] == { "response_cost": 0.0123, "model_id": "m-1", "cache_hit": False, } assert event.time_to_first_token_ms is None assert "SECRET PROMPT" not in str(event.to_dict()) assert "SECRET COMPLETION" not in str(event.to_dict()) def test_litellm_request_size_prefers_the_provider_payload_litellm_built() -> None: kwargs = _anthropic_kwargs(None) payload = {"model": "claude-sonnet-4-5", "system": "S", "messages": [], "max_tokens": 1} kwargs["additional_args"] = { "complete_input_dict": payload, "headers": {"x-api-key": "sk-ant-secret"}, } event = request_log.event_from_litellm(kwargs, _FakeResponse(), None, None, outcome="success") assert event.request_bytes == request_log.json_size(payload) assert "sk-ant-secret" not in str(event.to_dict()) def test_litellm_request_size_accepts_the_serialized_payload_streaming_adapters_record() -> None: kwargs = _anthropic_kwargs(None, stream=True) payload = '{"model":"claude-sonnet-4-5","messages":[],"max_tokens":1,"stream":true}' kwargs["additional_args"] = {"complete_input_dict": payload} event = request_log.event_from_litellm(kwargs, _FakeResponse(), None, None, outcome="success") assert event.request_bytes == len(payload.encode()) def test_openrouter_generation_id_is_the_request_id_and_the_cloudflare_ray_is_not() -> None: headers = {"cf-ray": "a3f5d8f23fde88dc-PDX", "x-generation-id": "gen-1790127690-M8vfqPTAr"} assert request_log.request_id_from_reply(headers) == "gen-1790127690-M8vfqPTAr" assert request_log.request_id_from_headers({"cf-ray": "a3f5-PDX"}) is None assert request_log.request_id_from_reply({"cf-ray": "a3f5-PDX"}) is None def test_request_id_in_the_error_body_when_the_gateway_strips_the_header() -> None: """A gateway that strips ``request-id`` still forwards Anthropic's body.""" body = '{"type":"error","error":{"type":"not_found_error"},"request_id":"req_body1"}' exc = _anthropic_error(404, body, {"cf-ray": "a3f5-PDX", "content-type": "application/json"}) event = request_log.event_from_litellm( _anthropic_kwargs(exc), None, None, None, outcome="error" ) assert event.provider_request_id == "req_body1" assert request_log.request_id_from_reply({"cf-ray": "a3f5-PDX"}, body) == "req_body1" assert request_log.request_id_from_reply({"cf-ray": "a3f5-PDX"}, "no id here") is None 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_headers_and_error_details() -> 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", "content-type": "application/json", }, ) 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.response_headers == { "request-id": "req_429", "retry-after": "7", "anthropic-ratelimit-requests-remaining": "0", "content-type": "application/json", } assert event.details is not None assert event.details["request"] == {"max_tokens": 10, "message_count": 1} assert event.details["error"]["llm_provider"] == "anthropic" assert event.finish_reason is None assert "sk-ant-secret" not in str(event.to_dict()) def test_litellm_failure_without_a_reply_has_no_status_headers_or_size() -> 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.status_code is None assert event.response_headers is None assert event.response_bytes is None assert event.error_type == "APITimeoutError" def test_litellm_connection_failure_drops_the_synthetic_500() -> None: exc = LiteLlmConnectionError( message="Connection refused", llm_provider="openrouter", model="openrouter/x" ) kwargs = _anthropic_kwargs(exc) event = request_log.event_from_litellm(kwargs, None, None, None, outcome="error") assert exc.status_code == 500 assert event.status_code is None assert event.response_bytes is None assert event.provider_request_id is None def test_litellm_timeout_drops_the_synthetic_408() -> None: exc = LiteLlmTimeout(message="Request timed out.", model="x", llm_provider="openrouter") kwargs = _anthropic_kwargs(exc) event = request_log.event_from_litellm(kwargs, None, None, None, outcome="error") assert exc.status_code == 408 assert event.status_code is None assert event.response_bytes is None def test_litellm_api_error_with_headers_keeps_its_status() -> None: body = '{"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}' exc = LiteLlmApiError( status_code=529, message=f"AnthropicException - b'{body}'", llm_provider="anthropic", model="x", ) # LiteLLM's exception mapping attaches the reply's headers this way. exc.litellm_response_headers = {"request-id": "req_529"} # type: ignore[attr-defined] kwargs = _anthropic_kwargs(exc) event = request_log.event_from_litellm(kwargs, None, None, None, outcome="error") assert event.status_code == 529 assert event.provider_request_id == "req_529" assert event.response_bytes == len(body.encode()) assert event.details is not None assert event.details["error"]["body"] == { "type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}, } def test_litellm_mapped_error_without_a_json_body_has_no_size_but_keeps_its_status() -> None: exc = LiteLlmApiError( status_code=502, message="OpenrouterException - bad gateway", llm_provider="openrouter", model="x", ) exc.litellm_response_headers = {"cf-ray": "a3f5-PDX"} # type: ignore[attr-defined] event = request_log.event_from_litellm( _anthropic_kwargs(exc), None, None, None, outcome="error" ) assert event.status_code == 502 assert event.response_bytes is None assert event.provider_request_id is None assert event.response_headers == {"cf-ray": "a3f5-PDX"} assert event.details is not None assert "body" not in event.details["error"] def test_litellm_mapped_error_sizes_the_body_it_kept_not_its_empty_stand_in_response() -> None: # LiteLLM's mapped exceptions subclass openai.APIStatusError and carry an # httpx.Response with no content; the reply body lives on ``body``/``message``. body = {"error": {"message": "User not found.", "code": 401}} exc = LiteLlmAuthenticationError( message="OpenrouterException - " + json.dumps(body, separators=(",", ":")), llm_provider="openrouter", model="x", ) assert isinstance(exc, APIStatusError) assert exc.response.content == b"" exc.litellm_response_headers = {"cf-ray": "a3f5-PDX"} # type: ignore[attr-defined] event = request_log.event_from_litellm( _anthropic_kwargs(exc), None, None, None, outcome="error" ) assert event.status_code == 401 assert event.response_bytes == len(json.dumps(body, separators=(",", ":")).encode()) def test_litellm_host_is_the_configured_endpoint_not_the_provider_default() -> None: # On a non-streamed call through a gateway, LiteLLM's hidden api_base can be # the provider default (api.openai.com) while the call went to the gateway. kwargs = _anthropic_kwargs(None) kwargs["litellm_params"] = {"api_base": "https://openrouter.ai/api/v1/"} kwargs["standard_logging_object"]["api_base"] = "https://openrouter.ai/api/v1" kwargs["standard_logging_object"]["hidden_params"]["api_base"] = "https://api.openai.com" event = request_log.event_from_litellm(kwargs, _FakeResponse(), None, None, outcome="success") assert event.api_host == "openrouter.ai" 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", "response_headers", "details", ): assert key in data assert "rate_limit" not in data assert data["time_to_first_token_ms"] == 100 # --------------------------------------------------------------------------- # # Native OpenAI route # # --------------------------------------------------------------------------- # def _openai_response(response_id: str, *, usage: Usage | None = None) -> ModelResponse: return ModelResponse( output=[], usage=usage or Usage(requests=1, input_tokens=10, output_tokens=5, total_tokens=15), response_id=response_id, ) def _completed_event(response_id: str) -> ResponseCompletedEvent: response = Response( id=response_id, created_at=0, model="gpt-5", object="response", output=[], parallel_tool_calls=False, tool_choice="auto", tools=[], ) return ResponseCompletedEvent(response=response, sequence_number=1, type="response.completed") class _Inner(Model): def __init__( self, *, response: ModelResponse | None = None, exc: BaseException | None = None, stream_events: list[Any] | None = None, fail_after: int | None = None, ) -> None: self._response = response self._exc = exc self._stream_events = stream_events or [] self._fail_after = fail_after self.closed = False async def get_response(self, *_args: Any, **_kwargs: Any) -> ModelResponse: if self._exc is not None: raise self._exc assert self._response is not None return self._response async def stream_response(self, *_args: Any, **_kwargs: Any) -> AsyncIterator[Any]: for index, event in enumerate(self._stream_events): if self._fail_after is not None and index == self._fail_after: assert self._exc is not None raise self._exc yield event async def close(self) -> None: self.closed = True _CALL_ARGS: tuple[Any, ...] = (None, "hi", None, [], None, [], None) _CALL_KWARGS: dict[str, Any] = { "previous_response_id": None, "conversation_id": None, "prompt": None, } def _openai_status_error(status: int, request_id: str | None, body: str = "") -> APIStatusError: request = httpx.Request("POST", "https://api.openai.com/v1/responses") headers = {"x-request-id": request_id} if request_id else {} response = httpx.Response(status, request=request, headers=headers, text=body) if status == 403: return PermissionDeniedError(body or "denied", response=response, body=None) return APIStatusError(body or f"status {status}", response=response, body=None) @pytest.mark.asyncio async def test_openai_route_success_event(captured: list[LlmRequestEvent]) -> None: model = request_log.RequestLoggingModel( _Inner(response=_openai_response("resp_123")), model_name="gpt-5", provider="openai", base_url=None, ) result = await model.get_response(*_CALL_ARGS, **_CALL_KWARGS) assert result.response_id == "resp_123" assert len(captured) == 1 event = captured[0] assert event.route == "openai" assert event.provider == "openai" assert event.api_host == "api.openai.com" assert event.outcome == "success" assert event.status_code == 200 assert event.response_id == "resp_123" assert event.provider_request_id is None assert (event.input_tokens, event.output_tokens, event.total_tokens) == (10, 5, 15) assert event.streaming is False assert model.model == "gpt-5" @pytest.mark.asyncio async def test_openai_route_drops_the_sdk_placeholder_response_id( captured: list[LlmRequestEvent], ) -> None: """Chat-completions backends get ``__fake_id__`` from the SDK, not a provider id.""" model = request_log.RequestLoggingModel( _Inner(response=_openai_response(FAKE_RESPONSES_ID)), model_name="anthropic/claude-sonnet-4-5", provider="openai", base_url="https://openrouter.ai/api/v1", ) await model.get_response(*_CALL_ARGS, **_CALL_KWARGS) assert len(captured) == 1 assert captured[0].response_id is None assert FAKE_RESPONSES_ID not in str(captured[0].to_dict()) @pytest.mark.asyncio async def test_openai_route_streaming_drops_the_sdk_placeholder_response_id( captured: list[LlmRequestEvent], ) -> None: model = request_log.RequestLoggingModel( _Inner(stream_events=[_completed_event(FAKE_RESPONSES_ID)]), model_name="anthropic/claude-sonnet-4-5", provider="openai", base_url="https://openrouter.ai/api/v1", ) async for _ in model.stream_response(*_CALL_ARGS, **_CALL_KWARGS): pass assert len(captured) == 1 assert captured[0].response_id is None assert FAKE_RESPONSES_ID not in str(captured[0].to_dict()) @pytest.mark.asyncio async def test_openai_route_blocked_request_keeps_header_request_id( captured: list[LlmRequestEvent], ) -> None: exc = _openai_status_error(403, "req_openai_blocked", "content policy violation") model = request_log.RequestLoggingModel( _Inner(exc=exc), model_name="gpt-5", provider="openai", base_url="https://gateway.corp.example/v1?token=abc", ) with pytest.raises(PermissionDeniedError): await model.get_response(*_CALL_ARGS, **_CALL_KWARGS) event = captured[0] assert event.outcome == "error" assert event.status_code == 403 assert event.provider_request_id == "req_openai_blocked" assert event.error_type == "PermissionDeniedError" assert event.api_host == "gateway.corp.example" assert "token=abc" not in str(event.to_dict()) @pytest.mark.asyncio async def test_openai_route_timeout_event(captured: list[LlmRequestEvent]) -> None: exc = APITimeoutError(httpx.Request("POST", "https://api.openai.com/v1/responses")) model = request_log.RequestLoggingModel( _Inner(exc=exc), model_name="gpt-5", provider="openai", base_url=None ) with pytest.raises(APITimeoutError): await model.get_response(*_CALL_ARGS, **_CALL_KWARGS) assert captured[0].outcome == "error" assert captured[0].status_code is None assert captured[0].error_type == "APITimeoutError" assert captured[0].provider_request_id is None @pytest.mark.asyncio async def test_openai_route_cancellation_is_logged_as_an_abandoned_attempt( captured: list[LlmRequestEvent], ) -> None: model = request_log.RequestLoggingModel( _Inner(exc=asyncio.CancelledError()), model_name="gpt-5", provider="openai", base_url=None ) with pytest.raises(asyncio.CancelledError): await model.get_response(*_CALL_ARGS, **_CALL_KWARGS) assert len(captured) == 1 event = captured[0] assert event.outcome == "error" assert event.error_type == "CancelledError" assert event.status_code is None assert "cancelled" in (event.error_message or "") assert event.request_bytes is not None class _ReplyingInner(_Inner): """Behaves like the OpenAI SDK: the wire reply is only visible to the httpx hook.""" def __init__(self, *, reply: httpx.Response, **kwargs: Any) -> None: super().__init__(**kwargs) self._reply = reply async def get_response(self, *args: Any, **kwargs: Any) -> ModelResponse: await request_log.record_http_reply(self._reply) return await super().get_response(*args, **kwargs) async def stream_response(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any]: await request_log.record_http_reply(self._reply) async for event in super().stream_response(*args, **kwargs): yield event def _wire_reply(status: int, headers: dict[str, str]) -> httpx.Response: return httpx.Response( status, request=httpx.Request("POST", "https://api.openai.com/v1/responses"), headers=headers, ) @pytest.mark.asyncio async def test_openai_route_success_reads_request_id_and_headers_from_the_wire( captured: list[LlmRequestEvent], ) -> None: reply = _wire_reply( 200, { "x-request-id": "req_wire_ok", "openai-processing-ms": "812", "x-ratelimit-remaining-tokens": "999", "Set-Cookie": "__cf_bm=abc", }, ) model = request_log.RequestLoggingModel( _ReplyingInner(reply=reply, response=_openai_response("resp_wire")), model_name="gpt-5", provider="openai", base_url=None, ) await model.get_response(*_CALL_ARGS, **_CALL_KWARGS) event = captured[0] assert event.outcome == "success" assert event.status_code == 200 assert event.provider_request_id == "req_wire_ok" assert event.response_id == "resp_wire" assert event.response_headers is not None assert event.response_headers["openai-processing-ms"] == "812" assert event.response_headers["x-ratelimit-remaining-tokens"] == "999" assert event.response_headers["set-cookie"] == "__cf_bm=abc" @pytest.mark.asyncio async def test_openai_route_streaming_success_reads_request_id_from_the_wire( captured: list[LlmRequestEvent], ) -> None: reply = _wire_reply(200, {"x-request-id": "req_wire_stream"}) model = request_log.RequestLoggingModel( _ReplyingInner(reply=reply, stream_events=[_completed_event("resp_s")]), model_name="gpt-5", provider="openai", base_url=None, ) _ = [e async for e in model.stream_response(*_CALL_ARGS, **_CALL_KWARGS)] assert captured[0].provider_request_id == "req_wire_stream" assert captured[0].response_id == "resp_s" assert captured[0].response_headers == {"x-request-id": "req_wire_stream"} @pytest.mark.asyncio async def test_openai_route_cancel_midstream_keeps_the_wire_status_and_request_id( captured: list[LlmRequestEvent], ) -> None: reply = _wire_reply(200, {"x-request-id": "req_wire_cancel"}) created = ResponseCreatedEvent( response=_completed_event("resp_c").response, sequence_number=0, type="response.created" ) model = request_log.RequestLoggingModel( _ReplyingInner( reply=reply, stream_events=[created, created], exc=asyncio.CancelledError(), fail_after=1, ), model_name="gpt-5", provider="openai", base_url=None, ) with pytest.raises(asyncio.CancelledError): async for _ in model.stream_response(*_CALL_ARGS, **_CALL_KWARGS): pass event = captured[0] assert event.outcome == "error" assert event.error_type == "CancelledError" assert event.status_code == 200 assert event.provider_request_id == "req_wire_cancel" assert event.streaming is True assert event.time_to_first_token_ms is not None @pytest.mark.asyncio async def test_openai_route_error_prefers_exception_fields_over_the_wire_reply( captured: list[LlmRequestEvent], ) -> None: exc = _openai_status_error(429, "req_exc") model = request_log.RequestLoggingModel( _ReplyingInner(reply=_wire_reply(200, {"x-request-id": "req_stale"}), exc=exc), model_name="gpt-5", provider="openai", base_url=None, ) with pytest.raises(APIStatusError): await model.get_response(*_CALL_ARGS, **_CALL_KWARGS) assert captured[0].status_code == 429 assert captured[0].provider_request_id == "req_exc" @pytest.mark.asyncio async def test_http_reply_hook_ignores_requests_outside_an_attempt() -> None: await request_log.record_http_reply(_wire_reply(500, {"x-request-id": "req_nobody"})) assert request_log._http_reply.get() is None @pytest.mark.asyncio async def test_http_reply_is_scoped_to_the_attempt_that_awaits_it( captured: list[LlmRequestEvent], ) -> None: def _model(request_id: str, response_id: str) -> request_log.RequestLoggingModel: return request_log.RequestLoggingModel( _ReplyingInner( reply=_wire_reply(200, {"x-request-id": request_id}), response=_openai_response(response_id), ), model_name="gpt-5", provider="openai", base_url=None, ) await asyncio.gather( _model("req_a", "resp_a").get_response(*_CALL_ARGS, **_CALL_KWARGS), _model("req_b", "resp_b").get_response(*_CALL_ARGS, **_CALL_KWARGS), ) by_response = {e.response_id: e.provider_request_id for e in captured} assert by_response == {"resp_a": "req_a", "resp_b": "req_b"} assert request_log._http_reply.get() is None def test_observe_http_client_installs_the_hook_once() -> None: client = httpx.AsyncClient() request_log.observe_http_client(client) request_log.observe_http_client(client) assert client.event_hooks["response"].count(request_log.record_http_reply) == 1 def test_install_observes_the_sdk_shared_http_client() -> None: request_log.install() assert request_log.record_http_reply in shared_http_client().event_hooks["response"] @pytest.mark.asyncio async def test_litellm_route_wrapper_only_logs_abandoned_attempts( captured: list[LlmRequestEvent], ) -> None: ok = request_log.RequestLoggingModel( _Inner(response=_openai_response("resp_ok")), model_name="anthropic/claude-sonnet-4-5", provider="anthropic", base_url=None, route="litellm", abandoned_only=True, ) await ok.get_response(*_CALL_ARGS, **_CALL_KWARGS) failed = request_log.RequestLoggingModel( _Inner(exc=_openai_status_error(500, "req_litellm_handles_this")), model_name="anthropic/claude-sonnet-4-5", provider="anthropic", base_url=None, route="litellm", abandoned_only=True, ) with pytest.raises(APIStatusError): await failed.get_response(*_CALL_ARGS, **_CALL_KWARGS) assert captured == [] cancelled = request_log.RequestLoggingModel( _Inner(stream_events=[_completed_event("r")], exc=asyncio.CancelledError(), fail_after=0), model_name="anthropic/claude-sonnet-4-5", provider="anthropic", base_url="https://api.anthropic.com", route="litellm", abandoned_only=True, ) with pytest.raises(asyncio.CancelledError): async for _ in cancelled.stream_response(*_CALL_ARGS, **_CALL_KWARGS): pass assert len(captured) == 1 event = captured[0] assert event.route == "litellm" assert event.provider == "anthropic" assert event.api_host == "api.anthropic.com" assert event.outcome == "error" assert event.error_type == "CancelledError" assert event.streaming is True assert event.request_bytes is not None @pytest.mark.asyncio async def test_stream_closed_early_is_logged_as_abandoned(captured: list[LlmRequestEvent]) -> None: created = ResponseCreatedEvent( response=_completed_event("resp_e").response, sequence_number=0, type="response.created" ) model = request_log.RequestLoggingModel( _Inner(stream_events=[created, created, created]), model_name="gpt-5", provider="openai", base_url=None, ) stream = cast("AsyncGenerator[Any, None]", model.stream_response(*_CALL_ARGS, **_CALL_KWARGS)) await stream.__anext__() await stream.aclose() assert len(captured) == 1 assert captured[0].error_type == "GeneratorExit" assert "closed by the caller" in (captured[0].error_message or "") @pytest.mark.asyncio async def test_openai_route_streaming_success(captured: list[LlmRequestEvent]) -> None: created = ResponseCreatedEvent( response=_completed_event("resp_stream").response, sequence_number=0, type="response.created", ) model = request_log.RequestLoggingModel( _Inner(stream_events=[created, _completed_event("resp_stream")]), model_name="gpt-5", provider="openai-codex", base_url="https://chatgpt.com/backend-api", ) events = [e async for e in model.stream_response(*_CALL_ARGS, **_CALL_KWARGS)] assert len(events) == 2 assert len(captured) == 1 event = captured[0] assert event.streaming is True assert event.outcome == "success" assert event.response_id == "resp_stream" assert event.provider == "openai-codex" assert event.api_host == "chatgpt.com" @pytest.mark.asyncio async def test_openai_route_streaming_failure_midstream(captured: list[LlmRequestEvent]) -> None: exc = _openai_status_error(500, "req_mid_stream", "upstream reset") created = ResponseCreatedEvent( response=_completed_event("resp_x").response, sequence_number=0, type="response.created" ) model = request_log.RequestLoggingModel( _Inner(stream_events=[created, created], exc=exc, fail_after=1), model_name="gpt-5", provider="openai", base_url=None, ) received = 0 with pytest.raises(APIStatusError): async for _ in model.stream_response(*_CALL_ARGS, **_CALL_KWARGS): received += 1 assert received == 1 assert len(captured) == 1 assert captured[0].streaming is True 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.response_headers is None assert event.details is not None assert event.details["request"]["tool_count"] == 1 assert event.details["request"]["input_items"] == 1 assert event.details["response"]["usage"]["input_tokens"] == 10 assert "SYSTEM SECRET INSTRUCTIONS" not in str(event.to_dict()) @pytest.mark.asyncio async def test_openai_route_error_carries_response_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.response_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", "content-length": str(len(body.encode())), "content-type": "text/plain; charset=utf-8", } assert event.response_bytes is not None and event.response_bytes > 0 assert event.details is not None assert event.details["request"]["tool_count"] == 0 @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 async def test_openai_route_delegates_close() -> None: inner = _Inner(response=_openai_response("r")) model = request_log.RequestLoggingModel( inner, model_name="gpt-5", provider="openai", base_url=None ) await model.close() assert inner.closed is True # --------------------------------------------------------------------------- # # failure_text # # --------------------------------------------------------------------------- # def test_failure_text_appends_header_id_for_openai_errors() -> None: exc = _openai_status_error(403, "req_hdr_only", "blocked") text = request_log.failure_text(exc) assert text.endswith("[provider request id: req_hdr_only]") assert "blocked" in text def test_failure_text_does_not_duplicate_body_id() -> None: exc = _anthropic_error(400, ANTHROPIC_BLOCK_BODY) text = request_log.failure_text(exc) assert text.count("req_011CVBodyOnly00") == 1 assert "[provider request id" not in text def test_failure_text_uses_litellm_exception_headers() -> None: exc = _anthropic_error(401, "bad key", {"request-id": "req_exc_hdr"}) text = request_log.failure_text(exc) assert "bad key" in text assert text.endswith("[provider request id: req_exc_hdr]") def test_failure_text_plain_exception_unchanged() -> None: assert request_log.failure_text(RuntimeError("boom")) == "boom" assert request_log.failure_text(RuntimeError("")) == "RuntimeError" def test_extra_headers_openai_client_rides_the_observed_shared_client( monkeypatch: pytest.MonkeyPatch, ) -> None: request_log.install() monkeypatch.setattr(_openai_shared, "_default_openai_client", None) settings = SimpleNamespace(api_key="k", api_base="https://gateway.example/v1") models._register_openai_client_with_headers(cast("Any", settings), {"X-Gateway-Feature": "svc"}) client = _openai_shared.get_default_openai_client() assert client is not None assert client._client is shared_http_client() assert request_log.record_http_reply in client._client.event_hooks["response"] def test_codex_client_is_observed(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(codex, "get_valid_token", lambda: ("access", "acct")) client = codex.build_openai_client() assert request_log.record_http_reply in client._client.event_hooks["response"]