From b4a774ace8abb1dc18c72eaa4416a4c1227e6cb9 Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Wed, 23 Sep 2026 00:27:37 +0000 Subject: [PATCH] fix(llm): log cancelled attempts, capture native reply headers/request ids on every OpenAI client - RequestLoggingModel emits a terminal error event when an attempt is cancelled (stream idle timeout, shutdown) or the stream is closed early, instead of dropping it; LiteLLM-routed models get the same wrapper in abandoned-only mode since LiteLLM has no callback for that case. - An httpx response hook on the Agents SDK shared client records the reply's status and headers for the attempt in flight, so successful native attempts carry the real status, sanitized response headers and the provider request id. - The LLM_EXTRA_HEADERS OpenAI client now rides the shared observed client and the Codex client registers its own httpx client with the hook. --- strix/config/codex.py | 3 + strix/config/models.py | 30 +++- strix/llm/request_log.py | 252 +++++++++++++++++++++++-------- tests/test_llm_request_log.py | 270 +++++++++++++++++++++++++++++++++- 4 files changed, 486 insertions(+), 69 deletions(-) diff --git a/strix/config/codex.py b/strix/config/codex.py index 9f81ff6d..b7cf7ae5 100644 --- a/strix/config/codex.py +++ b/strix/config/codex.py @@ -356,6 +356,8 @@ def build_openai_client() -> AsyncOpenAI: import httpx from openai import AsyncOpenAI + from strix.llm import request_log + get_valid_token() # fail fast at configure time if the sign-in is dead async def _auth_hook(request: httpx.Request) -> None: @@ -367,6 +369,7 @@ def build_openai_client() -> AsyncOpenAI: timeout=httpx.Timeout(600.0, connect=30.0), event_hooks={"request": [_auth_hook]}, ) + request_log.observe_http_client(http_client) return AsyncOpenAI( api_key="strix-codex-oauth", # placeholder; the hook overwrites Authorization base_url=CODEX_BASE_URL, diff --git a/strix/config/models.py b/strix/config/models.py index f7882c33..27e47a37 100644 --- a/strix/config/models.py +++ b/strix/config/models.py @@ -538,10 +538,22 @@ class StrixProvider(MultiProvider): ) else: model = super().get_model(model_name) - if not _routes_via_litellm(model): + resolved_name = model_name or llm.model or "unknown" + if _routes_via_litellm(model): + # LiteLLM's callbacks log every reply; only a cancelled attempt + # (stream idle timeout, abandoned turn) escapes them. model = request_log.RequestLoggingModel( model, - model_name=model_name or llm.model or "unknown", + model_name=resolved_name, + provider=_litellm_provider(resolved_name), + base_url=self._override_base_url or llm.api_base, + route="litellm", + abandoned_only=True, + ) + else: + model = request_log.RequestLoggingModel( + model, + model_name=resolved_name, provider="openai", base_url=self._override_base_url or llm.api_base, ) @@ -565,6 +577,16 @@ def _routes_via_litellm(model: Model) -> bool: return isinstance(model, LitellmModel) +def _litellm_provider(model_name: str) -> str | None: + """The provider LiteLLM will route ``model_name`` to, if it can tell.""" + try: + import litellm + + return str(litellm.get_llm_provider(model_name)[1]) + except Exception: # noqa: BLE001 - unknown model ids are the provider's problem, not the log's + return None + + DEFAULT_MODEL_RETRY = ModelRetrySettings( max_retries=5, backoff=ModelRetryBackoffSettings( @@ -794,12 +816,16 @@ def _merge_litellm_headers(headers: dict[str, str]) -> None: def _register_openai_client_with_headers(llm: LlmSettings, headers: dict[str, str]) -> None: from agents import set_default_openai_client + from agents.models.openai_provider import shared_http_client from openai import AsyncOpenAI client = AsyncOpenAI( api_key=llm.api_key or "not-needed", base_url=llm.api_base, default_headers=dict(headers), + # The SDK's shared client is the one the request log observes for + # reply status, headers and provider request ids. + http_client=shared_http_client(), ) set_default_openai_client(client, use_for_tracing=False) diff --git a/strix/llm/request_log.py b/strix/llm/request_log.py index f998dd52..84e0d113 100644 --- a/strix/llm/request_log.py +++ b/strix/llm/request_log.py @@ -55,6 +55,7 @@ if TYPE_CHECKING: from agents.models.interface import ModelTracing from agents.retry import ModelRetryAdvice, ModelRetryAdviceRequest from agents.tool import Tool + from httpx import AsyncClient, Response from openai.types.responses import ResponsePromptParam @@ -148,6 +149,45 @@ def set_retry_attempt(attempt: int) -> None: _call_context.set(replace(current_call_context(), retry_attempt=attempt)) +@dataclass +class HttpReply: + """Status and headers of the last provider reply received during one attempt. + + The OpenAI SDK hands its callers a parsed body only; the reply's status and + headers (and with them ``x-request-id``) exist solely on the wire. An + httpx response hook records them here for the attempt awaiting them. + """ + + status_code: int | None = None + headers: dict[str, str] | None = None + + +_http_reply: ContextVar[HttpReply | None] = ContextVar("strix_llm_http_reply", default=None) + + +async def record_http_reply(response: Response) -> None: + """httpx ``response`` event hook: remember the reply for the attempt in flight. + + Hooks run on the task that awaits the request, so the holder set by that + task's :class:`RequestLoggingModel` is the one in scope. Replies to + requests made outside a logged attempt are ignored. + """ + reply = _http_reply.get() + if reply is None: + return + reply.status_code = response.status_code + reply.headers = dict(response.headers.items()) + + +def observe_http_client(client: AsyncClient) -> None: + """Make every reply ``client`` receives visible to the attempt that awaits it.""" + hooks = client.event_hooks + response_hooks = list(hooks.get("response", [])) + if record_http_reply in response_hooks: + return + client.event_hooks = {**hooks, "response": [*response_hooks, record_http_reply]} + + def current_call_context() -> LlmCallContext: return _call_context.get(_DEFAULT_CALL_CONTEXT) @@ -883,10 +923,12 @@ _litellm_logger: Any | None = None def install() -> None: - """Attach the LiteLLM capture (idempotent) and the default log-line sink.""" + """Attach the LiteLLM capture (idempotent), the native-route reply capture + and the default log-line sink.""" global _litellm_logger # noqa: PLW0603 if _log_line_sink not in _sinks: _sinks.insert(0, _log_line_sink) + _observe_sdk_shared_http_client() if _litellm_logger is not None: return import litellm @@ -899,6 +941,16 @@ def install() -> None: callbacks.append(capture) # pyright: ignore[reportUnknownMemberType] +def _observe_sdk_shared_http_client() -> None: + """The agents SDK's default OpenAI client rides one shared httpx client.""" + try: + from agents.models.openai_provider import shared_http_client + + observe_http_client(shared_http_client()) + except Exception: # noqa: BLE001 - a missing hook degrades to inferred status/no id + logger.warning("could not observe the SDK's shared HTTP client", exc_info=True) + + def _log_line_sink(event: LlmRequestEvent) -> None: level = logging.DEBUG if event.outcome == "success" else logging.WARNING logger.log( @@ -972,19 +1024,38 @@ def _openai_usage(response: ModelResponse) -> dict[str, int | None]: class RequestLoggingModel(Model): - """Wrap a Model whose calls do not pass through LiteLLM. + """Record one event per attempt around a Model's ``get_response`` and + ``stream_response``. - The native OpenAI Responses / Chat Completions routes (``openai/…``, the - ChatGPT subscription backend) never hit ``litellm.acompletion``, so the - LiteLLM capture does not see them. This wrapper records the same event - around each ``get_response`` and ``stream_response``. + On the native OpenAI Responses / Chat Completions routes (``openai/…``, the + ChatGPT subscription backend) nothing passes through ``litellm.acompletion``, + so this wrapper is the capture: it emits every outcome and reads the + reply's status and headers from the :class:`HttpReply` the httpx hook + fills in. + + On the LiteLLM route LiteLLM's own callbacks report successes and + failures, but LiteLLM has no callback for an attempt that is cancelled + (the stream idle timeout, an abandoned turn, shutdown); with + ``abandoned_only=True`` the wrapper fills exactly that gap and stays + silent otherwise. """ - def __init__(self, inner: Model, *, model_name: str, provider: str, base_url: str | None): + def __init__( + self, + inner: Model, + *, + model_name: str, + provider: str | None, + base_url: str | None, + route: Route = "openai", + abandoned_only: bool = False, + ): self._inner = inner self._model_name = model_name self._provider = provider self._host = api_host(base_url) or ("api.openai.com" if provider == "openai" else None) + self._route: Route = route + self._abandoned_only = abandoned_only @property def model(self) -> str: @@ -1005,6 +1076,7 @@ class RequestLoggingModel(Model): response: ModelResponse | None, exc: BaseException | None, request: _OpenAiRequest, + reply: HttpReply, first_event_mono: float | None = None, finish_reason: str | None = None, raw_response: object = None, @@ -1019,14 +1091,14 @@ class RequestLoggingModel(Model): context = current_call_context() base = LlmRequestEvent( call_id=str(uuid.uuid4()), - route="openai", + route=self._route, provider=self._provider, model=self._model_name, api_host=self._host, streaming=streaming, outcome="success", - status_code=200, - provider_request_id=None, + status_code=reply.status_code or 200, + provider_request_id=request_id_from_headers(reply.headers), response_id=None, error_type=None, error_message=None, @@ -1039,6 +1111,7 @@ class RequestLoggingModel(Model): request_bytes=request.size, time_to_first_token_ms=ttft_ms, finish_reason=finish_reason, + response_headers=headers_from_response(reply.headers), ) if exc is None: usage = _openai_usage(response) if response is not None else {} @@ -1062,15 +1135,18 @@ class RequestLoggingModel(Model): ), ) status, request_id = _openai_error_fields(exc) + error_headers = _openai_error_headers(exc) return replace( base, outcome="error", - status_code=status, - provider_request_id=request_id, + # A reply the hook saw but the SDK turned into an exception (a + # cancel mid-stream, an unparsable 200) keeps its wire status. + status_code=status if status is not None else reply.status_code, + provider_request_id=request_id or base.provider_request_id, error_type=type(exc).__name__, - error_message=clean_error_message(exc), + error_message=_abandonment_message(exc) or clean_error_message(exc), response_bytes=_exception_body_size(exc), - response_headers=headers_from_response(_openai_error_headers(exc)), + response_headers=headers_from_response(error_headers) or base.response_headers, details=merge_details( ("request", request.details), ("error", _exception_details(exc)), @@ -1115,6 +1191,9 @@ class RequestLoggingModel(Model): details["conversation_id"] = conversation_id return _OpenAiRequest(size=json_size(body), details=details) + def _should_emit(self, exc: BaseException | None) -> bool: + return not self._abandoned_only or _is_abandonment(exc) + async def get_response( self, system_instructions: str | None, @@ -1130,14 +1209,22 @@ class RequestLoggingModel(Model): prompt: ResponsePromptParam | None, ) -> ModelResponse: started, started_mono = datetime.now(UTC), time.monotonic() - request = self._request( - system_instructions, - input, - model_settings, - tools, - previous_response_id=previous_response_id, - conversation_id=conversation_id, - ) + + def request() -> _OpenAiRequest: + return self._request( + system_instructions, + input, + model_settings, + tools, + previous_response_id=previous_response_id, + conversation_id=conversation_id, + ) + + # The LiteLLM route serializes the request only for the rare abandoned + # attempt; the native route sizes it before the SDK sees it. + eager = None if self._abandoned_only else request() + reply = HttpReply() + token = _http_reply.set(reply) try: response = await self._inner.get_response( system_instructions, @@ -1151,30 +1238,34 @@ class RequestLoggingModel(Model): conversation_id=conversation_id, prompt=prompt, ) - except asyncio.CancelledError: - raise except BaseException as exc: + if self._should_emit(exc): + emit( + self._event( + started=started, + started_mono=started_mono, + streaming=False, + response=None, + exc=exc, + request=eager or request(), + reply=reply, + ) + ) + raise + finally: + _reset_http_reply(token) + if self._should_emit(None): emit( self._event( started=started, started_mono=started_mono, streaming=False, - response=None, - exc=exc, - request=request, + response=response, + exc=None, + request=eager or request(), + reply=reply, ) ) - raise - emit( - self._event( - started=started, - started_mono=started_mono, - streaming=False, - response=response, - exc=None, - request=request, - ) - ) return response async def stream_response( @@ -1192,18 +1283,24 @@ class RequestLoggingModel(Model): prompt: ResponsePromptParam | None, ) -> AsyncIterator[TResponseStreamEvent]: started, started_mono = datetime.now(UTC), time.monotonic() - request = self._request( - system_instructions, - input, - model_settings, - tools, - previous_response_id=previous_response_id, - conversation_id=conversation_id, - ) + + def request() -> _OpenAiRequest: + return self._request( + system_instructions, + input, + model_settings, + tools, + previous_response_id=previous_response_id, + conversation_id=conversation_id, + ) + + eager = None if self._abandoned_only else request() completed: ModelResponse | None = None raw_response: object = None first_event_mono: float | None = None finish_reason: str | None = None + reply = HttpReply() + token = _http_reply.set(reply) try: async for event in self._inner.stream_response( system_instructions, @@ -1235,34 +1332,59 @@ class RequestLoggingModel(Model): completed = ModelResponse(output=[], usage=usage, response_id=event.response.id) raw_response = event.response yield event - except asyncio.CancelledError: - raise except BaseException as exc: + if self._should_emit(exc): + emit( + self._event( + started=started, + started_mono=started_mono, + streaming=True, + response=None, + exc=exc, + request=eager or request(), + reply=reply, + first_event_mono=first_event_mono, + ) + ) + raise + finally: + _reset_http_reply(token) + if self._should_emit(None): emit( self._event( started=started, started_mono=started_mono, streaming=True, - response=None, - exc=exc, - request=request, + response=completed, + exc=None, + request=eager or request(), + reply=reply, first_event_mono=first_event_mono, + finish_reason=finish_reason, + raw_response=raw_response, ) ) - raise - emit( - self._event( - started=started, - started_mono=started_mono, - streaming=True, - response=completed, - exc=None, - request=request, - first_event_mono=first_event_mono, - finish_reason=finish_reason, - raw_response=raw_response, - ) - ) + + +def _reset_http_reply(token: Token[HttpReply | None]) -> None: + # An async generator finalized by the event loop's shutdown hook runs in a + # different context than the one that started it; the token is void there. + with contextlib.suppress(ValueError): + _http_reply.reset(token) + + +def _is_abandonment(exc: BaseException | None) -> bool: + """The caller gave up on the attempt: cancelled (stream idle timeout, + shutdown) or closed the stream before it finished.""" + return isinstance(exc, asyncio.CancelledError | GeneratorExit) + + +def _abandonment_message(exc: BaseException) -> str | None: + if isinstance(exc, asyncio.CancelledError): + return "attempt cancelled before the reply was consumed (stream idle timeout or shutdown)" + if isinstance(exc, GeneratorExit): + return "stream closed by the caller before it finished" + return None @dataclass(frozen=True) diff --git a/tests/test_llm_request_log.py b/tests/test_llm_request_log.py index 5ae56432..907fc4f0 100644 --- a/tests/test_llm_request_log.py +++ b/tests/test_llm_request_log.py @@ -4,13 +4,16 @@ import asyncio 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.interface import Model +from agents.models.openai_provider import shared_http_client from agents.tool import FunctionTool from agents.usage import Usage from litellm.llms.anthropic.common_utils import AnthropicError @@ -18,11 +21,12 @@ 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 AsyncIterator, Iterator + from collections.abc import AsyncGenerator, AsyncIterator, Iterator from strix.llm.request_log import LlmRequestEvent @@ -854,14 +858,257 @@ async def test_openai_route_timeout_event(captured: list[LlmRequestEvent]) -> No @pytest.mark.asyncio -async def test_openai_route_cancellation_is_not_logged(captured: list[LlmRequestEvent]) -> None: +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": "session=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 "set-cookie" not in event.response_headers + + +@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: @@ -1090,3 +1337,22 @@ def test_failure_text_uses_litellm_exception_headers_and_redacts() -> None: 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"]