diff --git a/pyproject.toml b/pyproject.toml index d63d84521..ed8d2cdce 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -308,6 +308,9 @@ ignore = [ "strix/report/dedupe.py" = ["PLC0415"] "strix/telemetry/logging.py" = ["PLC0415"] "strix/config/models.py" = ["PLC0415"] +# LiteLLM is imported lazily: the request log is wired at startup on every +# route, including the native OpenAI ones that never load LiteLLM. +"strix/llm/request_log.py" = ["PLC0415"] # Heavy inference deps (httpx, openai) imported lazily so auth-status checks # don't pull them in. "strix/config/codex.py" = ["PLC0415"] diff --git a/strix/config/codex.py b/strix/config/codex.py index 9f81ff6db..b7cf7ae55 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 d8444ee1f..27e47a371 100644 --- a/strix/config/models.py +++ b/strix/config/models.py @@ -40,6 +40,7 @@ from strix.config import codex from strix.config.loader import load_settings from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input from strix.config.tool_call_limits import TurnToolCallLimiter +from strix.llm import request_log if TYPE_CHECKING: @@ -525,13 +526,37 @@ class StrixProvider(MultiProvider): # The ChatGPT subscription backend is always streamed; it has no # non-streaming mode to fall back to, so LLM_DISABLE_STREAMING # does not apply here. - model: Model = _CodexResponsesModel( - slug, - codex.get_subscription_client(), - reasoning_effort=llm.reasoning_effort, + model: Model = request_log.RequestLoggingModel( + _CodexResponsesModel( + slug, + codex.get_subscription_client(), + reasoning_effort=llm.reasoning_effort, + ), + model_name=slug, + provider="openai-codex", + base_url=None, ) else: model = super().get_model(model_name) + 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=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, + ) if llm.disable_streaming: model = _NonStreamingModel(model) # The wrapper emits its single event only once the whole request @@ -545,6 +570,23 @@ class StrixProvider(MultiProvider): ) +def _routes_via_litellm(model: Model) -> bool: + """LiteLLM-backed models are captured by the LiteLLM callback, not the wrapper.""" + from agents.extensions.models.litellm_model import LitellmModel + + 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( @@ -621,6 +663,7 @@ def configure_sdk_model_defaults(settings: Settings) -> None: """Apply Strix config to SDK-native defaults.""" llm = settings.llm set_tracing_disabled(True) + request_log.install() if codex.subscription_model(llm.model): return _configure_litellm_compatibility() @@ -773,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/core/execution.py b/strix/core/execution.py index dfcd39fa2..d5d94eb57 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -33,6 +33,7 @@ from strix.core.sessions import ( seed_initial_input, strip_all_images_from_session, ) +from strix.llm import request_log from strix.llm.compaction import is_context_overflow, maybe_compact @@ -197,6 +198,44 @@ async def run_agent_loop( start_parked: bool = False, event_sink: StreamEventSink | None = None, hooks: RunHooks[dict[str, Any]] | None = None, +) -> RunResultBase | None: + agent_name = getattr(agent, "name", None) + token = request_log.bind_call_context( + agent_id, agent_name if isinstance(agent_name, str) else None + ) + try: + return await _run_agent_loop( + agent=agent, + initial_input=initial_input, + run_config=run_config, + context=context, + max_turns=max_turns, + coordinator=coordinator, + agent_id=agent_id, + interactive=interactive, + session=session, + start_parked=start_parked, + event_sink=event_sink, + hooks=hooks, + ) + finally: + request_log.reset_call_context(token) + + +async def _run_agent_loop( + *, + agent: Any, + initial_input: Any, + run_config: RunConfig, + context: dict[str, Any], + max_turns: int, + coordinator: AgentCoordinator, + agent_id: str, + interactive: bool, + session: Session | None, + start_parked: bool, + event_sink: StreamEventSink | None, + hooks: RunHooks[dict[str, Any]] | None, ) -> RunResultBase | None: await coordinator.attach_runtime( agent_id, @@ -636,7 +675,7 @@ async def _run_cycle_parked( raise except Exception as exc: logger.exception("error escaped the run cycle for %s; parking as failed", agent_id) - await coordinator.set_status(agent_id, "failed", error=str(exc) or type(exc).__name__) + await coordinator.set_status(agent_id, "failed", error=request_log.failure_text(exc)) await notify_parent_on_terminal(coordinator, agent_id, "failed") return None @@ -658,6 +697,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 image_strips = 0 compactions = 0 model_retries = 0 + request_log.set_retry_attempt(0) while True: stream: Any = None pre_run_items: list[Any] = [] @@ -785,6 +825,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 exc, ) await asyncio.sleep(delay) + request_log.set_retry_attempt(model_retries) if session is not None: input_data = [] continue @@ -792,7 +833,9 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 await _salvage_stream_to_session(session, pre_run_items, stream, agent_id) if isinstance(exc, ProviderRefusalError): logger.warning("agent %s refused by the model provider: %s", agent_id, exc) - await coordinator.set_status(agent_id, "failed", error=str(exc)) + await coordinator.set_status( + agent_id, "failed", error=request_log.failure_text(exc) + ) await notify_parent_on_terminal(coordinator, agent_id, "failed") return None if isinstance(exc, MaxTurnsExceeded): @@ -806,7 +849,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 # non-interactive agent's task: a child that dies still owes its parent a # report, and the parent would otherwise wait out its timeout on a message # the dead child can no longer send. - await coordinator.set_status(agent_id, status, error=str(exc) or type(exc).__name__) + await coordinator.set_status(agent_id, status, error=request_log.failure_text(exc)) await notify_parent_on_terminal(coordinator, agent_id, status) if not interactive: raise diff --git a/strix/llm/request_log.py b/strix/llm/request_log.py new file mode 100644 index 000000000..84dd890d4 --- /dev/null +++ b/strix/llm/request_log.py @@ -0,0 +1,1413 @@ +"""Structured record of every model provider call. + +One :class:`LlmRequestEvent` per HTTP attempt against a provider, whatever the +route (LiteLLM or the native OpenAI client) and whatever the outcome. Retries +are separate events; a streamed call is one event once the stream settles. +Events carry the provider's own request identifier when the provider returns +one (Anthropic ``request-id``, OpenAI ``x-request-id``), which is what a +provider's support team asks for when a call was blocked or misbehaved. + +The typed fields are the common ground every provider shares (status, ids, +timing, sizes, tokens, cost). Everything else a provider or SDK reports about +the attempt travels free-form: the response headers, and a ``details`` object +with the request parameters, the full usage object, the SDK's hidden +parameters and the error body. Every string, list and object is bounded. Raw +request and response bodies are never captured: prompts and completions are +only ever measured for their size. + +Sinks are plain callables. The built-in sink writes one log line per event; +deployments register their own (a database, a queue) with +:func:`register_sink`. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import dataclasses +import json +import logging +import re +import time +import uuid +from collections.abc import AsyncIterator, Callable, Mapping, Sequence +from contextvars import ContextVar, Token +from dataclasses import asdict, dataclass, replace +from datetime import UTC, date, datetime +from typing import TYPE_CHECKING, Any, Literal, cast +from urllib.parse import urlsplit + +from agents.items import ModelResponse +from agents.models.fake_id import FAKE_RESPONSES_ID +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: + from agents.agent_output import AgentOutputSchemaBase + from agents.handoffs import Handoff + from agents.items import TResponseInputItem, TResponseStreamEvent + from agents.model_settings import ModelSettings + 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 + + +logger = logging.getLogger(__name__) + +Outcome = Literal["success", "error"] +Route = Literal["litellm", "openai"] + +ERROR_MESSAGE_MAX_CHARS = 2000 + + +@dataclass(frozen=True) +class LlmCallContext: + """Who is making the call, and which replay of the turn this is.""" + + agent_id: str | None = None + agent_name: str | None = None + retry_attempt: int = 0 + + +@dataclass(frozen=True) +class LlmRequestEvent: + """One attempt against a model provider.""" + + call_id: str + route: Route + provider: str | None + model: str + api_host: str | None + streaming: bool + outcome: Outcome + status_code: int | None + provider_request_id: str | None + response_id: str | None + error_type: str | None + error_message: str | None + started_at: datetime + finished_at: datetime + duration_ms: int + input_tokens: int | None = None + output_tokens: int | None = None + cached_input_tokens: int | None = None + total_tokens: int | None = None + cost_usd: float | None = None + 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 + # Every response header the provider sent, name-normalized, values capped. + response_headers: dict[str, str] | None = None + # Free-form: request parameters, full usage object, SDK hidden params, + # error body ... sizes bounded. + details: dict[str, Any] | None = None + + def to_dict(self) -> dict[str, Any]: + data = asdict(self) + data["started_at"] = self.started_at.isoformat() + data["finished_at"] = self.finished_at.isoformat() + return data + + +LlmRequestSink = Callable[[LlmRequestEvent], None] + +_sinks: list[LlmRequestSink] = [] +_DEFAULT_CALL_CONTEXT = LlmCallContext() +_call_context: ContextVar[LlmCallContext] = ContextVar("strix_llm_call_context") + + +def bind_call_context(agent_id: str | None, agent_name: str | None) -> Token[LlmCallContext]: + """Attribute every provider call made from this task tree to one agent. + + Bind in the task that owns the agent's run loop, not in an SDK hook: the + SDK awaits hooks through ``asyncio.gather``, whose child tasks copy the + context and cannot write it back. + """ + return _call_context.set(LlmCallContext(agent_id=agent_id, agent_name=agent_name)) + + +def reset_call_context(token: Token[LlmCallContext]) -> None: + _call_context.reset(token) + + +def set_retry_attempt(attempt: int) -> None: + """Stamp subsequent calls with the turn-replay number (0 = first try).""" + _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) + + +def register_sink(sink: LlmRequestSink) -> None: + """Add a sink. Sinks must return quickly and never raise; a raising sink is logged and kept.""" + if sink not in _sinks: + _sinks.append(sink) + install() + + +def unregister_sink(sink: LlmRequestSink) -> None: + with contextlib.suppress(ValueError): + _sinks.remove(sink) + + +def emit(event: LlmRequestEvent) -> None: + for sink in tuple(_sinks): + try: + sink(event) + except Exception: + logger.exception("LLM request log sink %r failed", sink) + + +def clean_error_message(exc: BaseException) -> str: + message = str(exc).strip() or type(exc).__name__ + if len(message) > ERROR_MESSAGE_MAX_CHARS: + message = message[: ERROR_MESSAGE_MAX_CHARS - 1] + "…" + return message + + +# --------------------------------------------------------------------------- # +# Provider request-id extraction # +# --------------------------------------------------------------------------- # + +# Provider-side request identifiers, in the order they are trusted. Anthropic +# and most gateways: ``request-id``; OpenAI/Azure: ``x-request-id``; Bedrock: +# ``x-amzn-requestid``; Vertex/Gemini: ``x-goog-request-id``; OpenRouter: +# ``x-generation-id``. LiteLLM prefixes stored provider headers with +# ``llm_provider-``; both spellings are accepted. +_REQUEST_ID_HEADERS: tuple[str, ...] = ( + "request-id", + "x-request-id", + "x-amzn-requestid", + "x-amz-request-id", + "x-goog-request-id", + "x-generation-id", +) +_LITELLM_HEADER_PREFIX = "llm_provider-" + +# Anthropic error bodies carry the id even when a gateway strips the header. +_BODY_REQUEST_ID = re.compile(r"[\"']?request_id[\"']?\s*[:=]\s*[\"']?(req_[A-Za-z0-9_-]+)") + + +def _header_value(headers: Mapping[str, Any] | None, names: tuple[str, ...]) -> str | None: + if not headers: + return None + normalized: dict[str, str] = {} + for key, value in headers.items(): + if not isinstance(value, str) or not value.strip(): + continue + name = str(key).lower().removeprefix(_LITELLM_HEADER_PREFIX) + normalized.setdefault(name, value.strip()) + for name in names: + if name in normalized: + return normalized[name] + return None + + +def request_id_from_headers(headers: Mapping[str, Any] | None) -> str | None: + """The provider's own request id from a reply's headers, if it sent one.""" + return _header_value(headers, _REQUEST_ID_HEADERS) + + +def request_id_from_text(text: str | None) -> str | None: + if not text: + return None + match = _BODY_REQUEST_ID.search(text) + return match.group(1) if match else None + + +def request_id_from_reply(headers: Mapping[str, Any] | None, text: str | None = None) -> str | None: + """Provider header id, then the id in the error body.""" + return request_id_from_headers(headers) or request_id_from_text(text) + + +# --------------------------------------------------------------------------- # +# Free-form telemetry: response headers and details # +# --------------------------------------------------------------------------- # + +FINISH_REASON_MAX_CHARS = 128 + +HEADERS_MAX_COUNT = 64 +HEADER_VALUE_MAX_CHARS = 512 + +DETAILS_MAX_BYTES = 16 * 1024 +DETAILS_MAX_DEPTH = 6 +DETAILS_MAX_ITEMS = 32 +DETAILS_MAX_STRING = 256 + + +def _normalize_header_name(key: object) -> str: + return str(key).lower().removeprefix(_LITELLM_HEADER_PREFIX).replace("_", "-").strip() + + +def headers_from_response(headers: Mapping[str, Any] | None) -> dict[str, str] | None: + """The provider's response headers, name-normalized and value-capped.""" + 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 = _normalize_header_name(key) + if not name: + continue + picked.setdefault(name, text[:HEADER_VALUE_MAX_CHARS]) + if len(picked) >= HEADERS_MAX_COUNT: + break + return picked or None + + +def _bounded_value(value: object, depth: int) -> object: # noqa: PLR0911 + if value is None or isinstance(value, bool | int | float): + return value + if isinstance(value, str): + return value[:DETAILS_MAX_STRING] + if isinstance(value, datetime | date): + return value.isoformat() + if depth >= DETAILS_MAX_DEPTH: + return "…" + if isinstance(value, BaseModel): + return _bounded_value(value.model_dump(exclude_none=True), depth) + if dataclasses.is_dataclass(value) and not isinstance(value, type): + return _bounded_value(dataclasses.asdict(value), depth) + if isinstance(value, Mapping): + out: dict[str, object] = {} + for raw_key, item in cast("Mapping[object, object]", value).items(): + key = str(raw_key) + cleaned = _bounded_value(item, depth + 1) + if cleaned is None or cleaned in ({}, []): + continue + out[key[:DETAILS_MAX_STRING]] = cleaned + if len(out) >= DETAILS_MAX_ITEMS: + break + return out + if isinstance(value, Sequence | set | frozenset): + items = list(cast("Sequence[object]", value))[:DETAILS_MAX_ITEMS] + return [_bounded_value(item, depth + 1) for item in items] + return str(value)[:DETAILS_MAX_STRING] + + +def bound_details(value: object) -> dict[str, Any] | None: + """A bounded copy of ``value`` (a mapping), or None. + + Strings are capped; nesting, item counts and total size are bounded. When + the copy is still too large the biggest top-level entries go first and + their names are listed under ``_dropped``. + """ + cleaned = _bounded_value(value, 0) + if not isinstance(cleaned, dict): + return None + details = cast("dict[str, Any]", cleaned) + details = {k: v for k, v in details.items() if v not in (None, {}, [])} + if not details: + return None + size = json_size(details) or 0 + if size <= DETAILS_MAX_BYTES: + return details + dropped: list[str] = [] + by_size = sorted(details, key=lambda k: json_size(details[k]) or 0, reverse=True) + for key in by_size: + if size <= DETAILS_MAX_BYTES: + break + size -= json_size(details.pop(key)) or 0 + dropped.append(key) + details["_dropped"] = dropped + return details + + +def merge_details(*parts: tuple[str, object]) -> dict[str, Any] | None: + """``{name: value}`` for every non-empty part, bounded as one object.""" + merged: dict[str, Any] = {} + for name, value in parts: + if value is None: + continue + merged[name] = value + return bound_details(merged) + + +def json_text(value: object) -> str | None: + """``value`` as compact JSON text; None when it cannot be serialized.""" + if value is None: + return None + try: + if isinstance(value, BaseModel): + return value.model_dump_json(exclude_none=True) + if isinstance(value, str): + return value + if isinstance(value, bytes): + return value.decode("utf-8", errors="replace") + return json.dumps(value, default=str, separators=(",", ":"), ensure_ascii=False) + except Exception: # noqa: BLE001 - telemetry is best-effort + return None + + +def json_size(value: object) -> int | None: + """Byte length of ``value`` serialized as compact JSON; None when it cannot be serialized.""" + text = json_text(value) + return None if text is None else 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(): + return None + try: + host = urlsplit(url if "://" in url else f"//{url}").hostname + except ValueError: + return None + return host or None + + +def _int_or_none(value: object) -> int | None: + if isinstance(value, bool): + return None + if isinstance(value, int): + return value + if isinstance(value, float) and value.is_integer(): + return int(value) + return None + + +def _float_or_none(value: object) -> float | None: + if isinstance(value, bool): + return None + if isinstance(value, int | float): + return float(value) + return None + + +def _str_or_none(value: object) -> str | None: + return value if isinstance(value, str) and value else None + + +def _real_response_id(response: ModelResponse | None) -> str | None: + """The provider's response id; the SDK's chat-completions placeholder is not one.""" + if response is None or response.response_id == FAKE_RESPONSES_ID: + return None + return _str_or_none(response.response_id) + + +def _mapping(value: object) -> Mapping[str, Any] | None: + return cast("Mapping[str, Any]", value) if isinstance(value, Mapping) else None + + +def _to_datetime(value: object, fallback: datetime) -> datetime: + if isinstance(value, datetime): + return value if value.tzinfo else value.replace(tzinfo=UTC) + if isinstance(value, int | float): + return datetime.fromtimestamp(value, tz=UTC) + return fallback + + +# --------------------------------------------------------------------------- # +# LiteLLM route # +# --------------------------------------------------------------------------- # + + +def _litellm_usage(response: object) -> dict[str, int | None]: + usage = getattr(response, "usage", None) + if usage is None: + return {} + prompt = _int_or_none(getattr(usage, "prompt_tokens", None)) + completion = _int_or_none(getattr(usage, "completion_tokens", None)) + total = _int_or_none(getattr(usage, "total_tokens", None)) + details = getattr(usage, "prompt_tokens_details", None) + cached = _int_or_none(getattr(details, "cached_tokens", None)) if details else None + if cached is None: + cached = _int_or_none(getattr(usage, "cache_read_input_tokens", None)) + return { + "input_tokens": prompt, + "output_tokens": completion, + "cached_input_tokens": cached, + "total_tokens": total, + } + + +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_body(kwargs: Mapping[str, Any]) -> object: + """The request as LiteLLM sent it, for its size only. + + The provider payload LiteLLM built (``additional_args.complete_input_dict``) + when the adapter recorded it; otherwise model, messages and the provider + params, which is the same information before translation. + """ + additional = _mapping(kwargs.get("additional_args")) + payload = additional.get("complete_input_dict") if additional else None + if isinstance(payload, str | bytes) and payload: + return payload + if _mapping(payload): + return payload + 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 body + + +def _count(value: object) -> int | None: + return len(cast("Sequence[object]", value)) if isinstance(value, Sequence | Mapping) else None + + +def _litellm_request_details(kwargs: Mapping[str, Any]) -> dict[str, Any]: + optional = _mapping(kwargs.get("optional_params")) or {} + request: dict[str, Any] = dict(optional) + counts = { + "message_count": _count(kwargs.get("messages")), + "tool_count": _count(optional.get("tools") or optional.get("functions")), + } + request.update({k: v for k, v in counts.items() if v is not None}) + return request + + +def _exception_details(exc: BaseException | None) -> dict[str, Any] | None: + """What the SDK exception says beyond its message: the error body and its typed fields.""" + if exc is None: + return None + error: dict[str, Any] = {} + body = _error_body(exc) + if body is not None: + error["body"] = body + for name in ("code", "param", "type", "llm_provider", "max_retries", "num_retries"): + value = getattr(exc, name, None) + if isinstance(value, str | int) and not isinstance(value, bool): + error[name] = value + return error or None + + +def _error_body(exc: BaseException) -> object: + """The provider's error reply as the SDK kept it. + + The OpenAI SDK parses it onto ``body``. LiteLLM's mapped exceptions keep + only their message, which ends in the reply text (``AnthropicException - + {...}``); the JSON object in that text is the body. + """ + body = getattr(exc, "body", None) + if body is not None: + return body + text = getattr(exc, "message", None) + return _json_object_in(text if isinstance(text, str) else str(exc)) + + +def _json_object_in(text: str) -> Mapping[str, Any] | None: + start, end = text.find("{"), text.rfind("}") + if start < 0 or end <= start: + return None + try: + return _mapping(json.loads(text[start : end + 1])) + except ValueError: + return None + + +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. The first block that + names a request id wins; failing that, the first non-empty block. + """ + if exc is None: + return None + candidates = ( + getattr(exc, "litellm_response_headers", 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 not headers: + continue + if request_id_from_headers(headers): + return headers + first = first or headers + return first + + +def _exception_body_size(exc: BaseException, status_code: int | None) -> int | None: + """Size of the error reply's body: the bytes the OpenAI SDK read, else the + body LiteLLM kept. Unknown when no reply arrived or the SDK kept none of it. + """ + if status_code is None: + return None + if isinstance(exc, APIStatusError): + # LiteLLM's mapped exceptions subclass this with an empty stand-in response. + with contextlib.suppress(Exception): + content = exc.response.content + if content: + return len(content) + return json_size(_error_body(exc)) + + +def _litellm_reply_status(exc: BaseException, headers: Mapping[str, Any] | None) -> int | None: + """The status of the provider's reply, or None when no reply arrived. + + LiteLLM stamps a status on every exception, including 408 on timeouts and + 500 on connection failures that never reached the provider. Those are the + non-status ``APIError`` family and carry no response headers; a reply that + was received always does. + """ + if isinstance(exc, APIError) and not isinstance(exc, APIStatusError) and not headers: + return None + return _int_or_none(getattr(exc, "status_code", None)) + + +def _headers_mapping(headers: object) -> Mapping[str, Any] | None: + return cast("Mapping[str, Any]", headers) if isinstance(headers, Mapping) else None + + +def event_from_litellm( + kwargs: Mapping[str, Any], + response: object, + start_time: object, + end_time: object, + *, + outcome: Outcome, +) -> LlmRequestEvent: + """Build the event from a LiteLLM success/failure callback payload.""" + now = datetime.now(UTC) + started = _to_datetime(start_time, now) + finished = _to_datetime(end_time, now) + slo = _mapping(kwargs.get("standard_logging_object")) or {} + hidden = _mapping(slo.get("hidden_params")) or {} + litellm_params = _mapping(kwargs.get("litellm_params")) or {} + error_info = _mapping(slo.get("error_information")) or {} + exc = kwargs.get("exception") + exc = exc if isinstance(exc, BaseException) else None + + provider = ( + _str_or_none(kwargs.get("custom_llm_provider")) + or _str_or_none(slo.get("custom_llm_provider")) + or _str_or_none(error_info.get("llm_provider")) + or _str_or_none(litellm_params.get("custom_llm_provider")) + ) + model = _str_or_none(kwargs.get("model")) or _str_or_none(slo.get("model")) or "unknown" + # The endpoint the call was configured with. LiteLLM's hidden ``api_base`` + # can be the provider default (``api.openai.com``) on a non-streamed call + # that actually went to a gateway. + host = api_host( + _str_or_none(litellm_params.get("api_base")) + or _str_or_none(slo.get("api_base")) + or _str_or_none(hidden.get("api_base")) + ) + streaming = bool(kwargs.get("stream")) or bool(slo.get("stream")) + call_id = ( + _str_or_none(kwargs.get("litellm_call_id")) + or _str_or_none(slo.get("litellm_call_id")) + or str(uuid.uuid4()) + ) + + # LiteLLM stores the reply's headers on the logging object or on the + # response, depending on the adapter; the first non-empty block is the reply. + response_hidden = _mapping(getattr(response, "_hidden_params", None)) or {} + reply_headers = _mapping(hidden.get("additional_headers")) or _mapping( + response_hidden.get("additional_headers") + ) + + 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() + base = LlmRequestEvent( + call_id=call_id, + route="litellm", + provider=provider, + model=model, + api_host=host, + streaming=streaming, + outcome=outcome, + status_code=None, + provider_request_id=request_id_from_reply(reply_headers), + 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=json_size(_litellm_request_body(kwargs)), + time_to_first_token_ms=ttft_ms, + response_headers=headers_from_response(reply_headers), + ) + if outcome == "success": + return _litellm_success(base, kwargs, response, hidden, slo) + return _litellm_failure(base, kwargs, exc, error_info, slo, reply_headers) + + +def _litellm_response_details(response: object) -> dict[str, Any] | None: + """Everything LiteLLM's normalized response says about the call except the choices.""" + if isinstance(response, BaseModel): + summary: dict[str, Any] = response.model_dump(exclude={"choices"}, exclude_none=True) + else: + summary = {} + for name in ("id", "created", "model", "object", "system_fingerprint", "service_tier"): + value = getattr(response, name, None) + if value is not None: + summary[name] = value + usage = getattr(response, "usage", None) + if usage is not None: + summary["usage"] = usage + choices = getattr(response, "choices", None) + if isinstance(choices, list) and choices: + first: object = cast("list[object]", choices)[0] + summary["choice"] = { + "finish_reason": getattr(first, "finish_reason", None), + "provider_specific_fields": getattr(first, "provider_specific_fields", None), + } + summary["choice_count"] = len(cast("list[object]", choices)) + return summary or None + + +def _litellm_hidden_details(hidden: Mapping[str, Any], slo: Mapping[str, Any]) -> dict[str, Any]: + """LiteLLM's own bookkeeping for the call (model id, cache hit, overhead, cost basis).""" + litellm_info: dict[str, Any] = { + k: v for k, v in hidden.items() if k not in {"additional_headers", "api_base"} + } + for name in ("cache_hit", "saved_cache_cost", "model_group", "model_id"): + value = slo.get(name) + if value is not None: + litellm_info.setdefault(name, value) + return litellm_info + + +def _litellm_success( + base: LlmRequestEvent, + kwargs: Mapping[str, Any], + response: object, + hidden: Mapping[str, Any], + slo: 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, + response_bytes=json_size(response), + finish_reason=_litellm_finish_reason(response), + details=merge_details( + ("request", _litellm_request_details(kwargs)), + ("response", _litellm_response_details(response)), + ("litellm", _litellm_hidden_details(hidden, slo)), + ), + ) + + +def _litellm_failure( + base: LlmRequestEvent, + kwargs: Mapping[str, Any], + exc: BaseException | None, + error_info: Mapping[str, Any], + slo: Mapping[str, Any], + reply_headers: Mapping[str, Any] | None, +) -> LlmRequestEvent: + status_code: int | None = None + error_type: str | None = None + error_message: str | None = None + response_bytes: int | None = None + error: dict[str, Any] = {} + if exc is not None: + reply_headers = reply_headers or _exception_headers(exc) + status_code = _litellm_reply_status(exc, reply_headers) + error_type = type(exc).__name__ + error_message = clean_error_message(exc) + response_bytes = _exception_body_size(exc, status_code) + error = _exception_details(exc) or {} + else: + 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 = raw[:ERROR_MESSAGE_MAX_CHARS] if raw else error_type + provider = _str_or_none(error_info.get("llm_provider")) + if provider: + error.setdefault("llm_provider", provider) + hidden = _mapping(slo.get("hidden_params")) or {} + return replace( + base, + status_code=status_code, + provider_request_id=request_id_from_reply(reply_headers, error_message), + error_type=error_type, + error_message=error_message, + response_bytes=response_bytes, + response_headers=headers_from_response(reply_headers), + details=merge_details( + ("request", _litellm_request_details(kwargs)), + ("error", error or None), + ("litellm", _litellm_hidden_details(hidden, slo)), + ), + ) + + +def _build_litellm_logger() -> Any: + from litellm.integrations.custom_logger import CustomLogger + + class _StrixRequestLogger(CustomLogger): + """Forwards each LiteLLM attempt to the registered sinks. + + Only the async handlers are implemented: the engine calls + ``acompletion`` exclusively, and LiteLLM schedules these on the calling + task, so the agent context bound there still applies. Sync fallbacks + run in a thread pool and would lose it. + """ + + async def async_log_success_event( + self, kwargs: dict[str, Any], response_obj: Any, start_time: Any, end_time: Any + ) -> None: + _dispatch(kwargs, response_obj, start_time, end_time, outcome="success") + + async def async_log_failure_event( + self, kwargs: dict[str, Any], response_obj: Any, start_time: Any, end_time: Any + ) -> None: + _dispatch(kwargs, response_obj, start_time, end_time, outcome="error") + + return _StrixRequestLogger() + + +def _dispatch( + kwargs: Mapping[str, Any], response: object, start: object, end: object, *, outcome: Outcome +) -> None: + try: + event = event_from_litellm(kwargs, response, start, end, outcome=outcome) + except Exception: + logger.exception("could not build LLM request event from LiteLLM callback") + return + emit(event) + + +_litellm_logger: Any | None = None + + +def install() -> None: + """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 + + capture = _build_litellm_logger() + _litellm_logger = capture + # litellm types this list with a bare Callable, which strict pyright cannot resolve. + callbacks = litellm.callbacks # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] + if capture not in callbacks: + 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( + 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 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 "-", + event.model, + event.api_host or "-", + event.outcome, + event.status_code if event.status_code is not None else "-", + event.provider_request_id or "-", + 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 "-", + f"{event.cost_usd:.6f}" if event.cost_usd is not None else "-", + event.agent_id or "-", + event.retry_attempt, + f" error={event.error_type}: {event.error_message}" if event.outcome == "error" else "", + ) + + +# --------------------------------------------------------------------------- # +# Native OpenAI route # +# --------------------------------------------------------------------------- # + + +def _openai_error_fields(exc: BaseException) -> tuple[int | None, str | None]: + """(status_code, provider_request_id) from an OpenAI SDK exception.""" + status: int | None = None + request_id: str | None = None + if isinstance(exc, APIStatusError): + status = _int_or_none(exc.status_code) + request_id = _str_or_none(exc.request_id) + return status, request_id or request_id_from_reply(_openai_error_headers(exc), str(exc)) + + +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 + return { + "input_tokens": _int_or_none(usage.input_tokens), + "output_tokens": _int_or_none(usage.output_tokens), + "cached_input_tokens": _int_or_none(details.cached_tokens) if details else None, + "total_tokens": _int_or_none(usage.total_tokens), + } + + +class RequestLoggingModel(Model): + """Record one event per attempt around a Model's ``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 | 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: + return self._model_name + + async def close(self) -> None: + await self._inner.close() + + def get_retry_advice(self, request: ModelRetryAdviceRequest) -> ModelRetryAdvice | None: + return self._inner.get_retry_advice(request) + + def _event( + self, + *, + started: datetime, + started_mono: float, + streaming: bool, + 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, + ) -> 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()), + route=self._route, + provider=self._provider, + model=self._model_name, + api_host=self._host, + streaming=streaming, + outcome="success", + status_code=reply.status_code or 200, + provider_request_id=request_id_from_reply(reply.headers), + response_id=None, + error_type=None, + error_message=None, + started_at=started, + finished_at=finished, + duration_ms=duration_ms, + agent_id=context.agent_id, + agent_name=context.agent_name, + retry_attempt=context.retry_attempt, + 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 {} + response_id = _real_response_id(response) + body_source = ( + raw_response + if raw_response is not None + else (response.output if response is not None else None) + ) + return replace( + base, + response_id=response_id, + 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"), + response_bytes=json_size(body_source), + details=merge_details( + ("request", request.details), + ("response", _openai_response_details(response, raw_response)), + ), + ) + status, request_id = _openai_error_fields(exc) + error_headers = _openai_error_headers(exc) + # A reply the hook saw but the SDK turned into an exception (a + # cancel mid-stream, an unparsable 200) keeps its wire status. + status = status if status is not None else reply.status_code + return replace( + base, + outcome="error", + status_code=status, + provider_request_id=request_id or base.provider_request_id, + error_type=type(exc).__name__, + error_message=_abandonment_message(exc) or clean_error_message(exc), + response_bytes=_exception_body_size(exc, status), + response_headers=headers_from_response(error_headers) or base.response_headers, + details=merge_details( + ("request", request.details), + ("error", _exception_details(exc)), + ), + ) + + @staticmethod + def _request( + system_instructions: str | None, + input: object, # noqa: A002 + model_settings: ModelSettings, + tools: list[Tool], + *, + previous_response_id: str | None, + conversation_id: str | None, + ) -> _OpenAiRequest: + 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 + ] + body: dict[str, Any] = { + "instructions": system_instructions, + "input": input, + "tools": serialized_tools, + } + settings = _model_settings_dict(model_settings) + body.update({k: v for k, v in settings.items() if v is not None}) + if previous_response_id: + body["previous_response_id"] = previous_response_id + if conversation_id: + body["conversation_id"] = conversation_id + details: dict[str, Any] = dict(settings) + details["input_items"] = _count(input) if not isinstance(input, str) else 1 + details["tool_count"] = len(tools) + details["previous_response_id"] = previous_response_id + 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, + input: str | list[TResponseInputItem], # noqa: A002 + model_settings: ModelSettings, + tools: list[Tool], + output_schema: AgentOutputSchemaBase | None, + handoffs: list[Handoff], + tracing: ModelTracing, + *, + previous_response_id: str | None, + conversation_id: str | None, + prompt: ResponsePromptParam | None, + ) -> ModelResponse: + started, started_mono = datetime.now(UTC), time.monotonic() + + 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, + input, + model_settings, + tools, + output_schema, + handoffs, + tracing, + previous_response_id=previous_response_id, + conversation_id=conversation_id, + prompt=prompt, + ) + 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=response, + exc=None, + request=eager or request(), + reply=reply, + ) + ) + return response + + async def stream_response( + self, + system_instructions: str | None, + input: str | list[TResponseInputItem], # noqa: A002 + model_settings: ModelSettings, + tools: list[Tool], + output_schema: AgentOutputSchemaBase | None, + handoffs: list[Handoff], + tracing: ModelTracing, + *, + previous_response_id: str | None, + conversation_id: str | None, + prompt: ResponsePromptParam | None, + ) -> AsyncIterator[TResponseStreamEvent]: + started, started_mono = datetime.now(UTC), time.monotonic() + + 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, + input, + model_settings, + tools, + output_schema, + handoffs, + tracing, + previous_response_id=previous_response_id, + 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: + usage = Usage( + requests=1, + input_tokens=raw_usage.input_tokens, + output_tokens=raw_usage.output_tokens, + total_tokens=raw_usage.total_tokens, + input_tokens_details=raw_usage.input_tokens_details, + output_tokens_details=raw_usage.output_tokens_details, + ) + completed = ModelResponse(output=[], usage=usage, response_id=event.response.id) + raw_response = event.response + yield event + 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=completed, + exc=None, + request=eager or request(), + reply=reply, + 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) +class _OpenAiRequest: + """What the wrapper knows about one native request before it is sent.""" + + size: int | None + details: dict[str, Any] + + +def _model_settings_dict(model_settings: ModelSettings) -> dict[str, Any]: + try: + return dict(model_settings.to_json_dict()) + except Exception: # noqa: BLE001 - settings are telemetry here, never required + return {} + + +def _openai_response_details( + response: ModelResponse | None, raw_response: object +) -> dict[str, Any] | None: + """The Responses API object minus its output, or the SDK usage when that is all there is.""" + if isinstance(raw_response, BaseModel): + dumped = raw_response.model_dump( + exclude={"output", "instructions", "tools", "text"}, exclude_none=True + ) + if dumped.get("id") == FAKE_RESPONSES_ID: + del dumped["id"] + return dumped + if response is None: + return None + return {"usage": response.usage, "output_items": len(response.output)} + + +def _openai_finish_reason(event: ResponseCompletedEvent) -> str | None: + """``completed`` / ``incomplete:`` 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. + + LiteLLM's Anthropic mapping keeps the error body (which carries + ``request_id``) but drops the ``request-id`` header; OpenAI errors carry + only the header. Either way the id lands in the text an operator reads. + """ + text = str(exc) or type(exc).__name__ + request_id: str | None = None + if isinstance(exc, APIError): + _, request_id = _openai_error_fields(exc) + request_id = request_id or request_id_from_reply(_exception_headers(exc), text) + if request_id and request_id not in text: + text = f"{text} [provider request id: {request_id}]" + return text + + +__all__ = [ + "ERROR_MESSAGE_MAX_CHARS", + "LlmCallContext", + "LlmRequestEvent", + "LlmRequestSink", + "RequestLoggingModel", + "api_host", + "bind_call_context", + "bound_details", + "clean_error_message", + "current_call_context", + "emit", + "event_from_litellm", + "failure_text", + "headers_from_response", + "install", + "json_size", + "json_text", + "merge_details", + "register_sink", + "request_id_from_headers", + "request_id_from_reply", + "request_id_from_text", + "reset_call_context", + "set_retry_attempt", + "unregister_sink", +] diff --git a/tests/test_disable_streaming.py b/tests/test_disable_streaming.py index 00e99bcc6..9bbb13a6f 100644 --- a/tests/test_disable_streaming.py +++ b/tests/test_disable_streaming.py @@ -32,6 +32,7 @@ from openai.types.responses import ( from strix.config import codex, loader from strix.config.loader import load_settings from strix.config.models import StrixProvider, _NonStreamingModel, _TurnGuardModel +from strix.llm.request_log import RequestLoggingModel if TYPE_CHECKING: @@ -312,7 +313,8 @@ def test_get_model_keeps_streaming_by_default( model = StrixProvider().get_model("openai/gpt-4o-mini") assert isinstance(model, _TurnGuardModel) - assert model._inner is inner + assert isinstance(model._inner, RequestLoggingModel) + assert model._inner._inner is inner def test_get_model_guards_subscription_model_but_keeps_it_streaming( diff --git a/tests/test_execution_request_log.py b/tests/test_execution_request_log.py new file mode 100644 index 000000000..426c4d479 --- /dev/null +++ b/tests/test_execution_request_log.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, cast + +import httpx +import pytest +from agents import RunConfig, Runner +from openai import APIError, PermissionDeniedError + +from strix.core import execution +from strix.core.agents import AgentCoordinator +from strix.llm import request_log + + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + +def _request() -> httpx.Request: + return httpx.Request("POST", "https://api.openai.com/v1/responses") + + +def _midstream_api_error() -> APIError: + return APIError("An error occurred while processing the request.", _request(), body=None) + + +def _blocked_error() -> PermissionDeniedError: + response = httpx.Response( + 403, request=_request(), headers={"x-request-id": "req_blocked_hdr"}, text="blocked" + ) + return PermissionDeniedError("Output blocked by policy", response=response, body=None) + + +class _FakeStream: + def __init__(self, exc: BaseException | None = None) -> None: + self._exc = exc + self.run_loop_exception: BaseException | None = None + self.seen_context: request_log.LlmCallContext | None = None + + async def stream_events(self) -> AsyncIterator[Any]: + self.seen_context = request_log.current_call_context() + if self._exc is not None: + raise self._exc + items: tuple[Any, ...] = () + for item in items: + yield item + + +async def _run( + monkeypatch: pytest.MonkeyPatch, + streams: list[_FakeStream], + coordinator: AgentCoordinator | None = None, +) -> Any: + monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_BASE_DELAY_S", 0.0) + monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_MAX_DELAY_S", 0.0) + calls = {"n": 0} + + def _fake_run_streamed(*_args: Any, **_kwargs: Any) -> _FakeStream: + stream = streams[calls["n"]] + calls["n"] += 1 + return stream + + monkeypatch.setattr(Runner, "run_streamed", _fake_run_streamed) + coordinator = coordinator or AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + return await execution._run_cycle( + object(), + coordinator, + "root", + input_data="task", + run_config=cast("RunConfig", object()), + context={}, + max_turns=5, + session=None, + interactive=False, + event_sink=None, + hooks=None, + ) + + +@pytest.mark.asyncio +async def test_each_transient_replay_is_stamped_with_its_attempt_number( + monkeypatch: pytest.MonkeyPatch, +) -> None: + streams = [ + _FakeStream(exc=_midstream_api_error()), + _FakeStream(exc=_midstream_api_error()), + _FakeStream(), + ] + await _run(monkeypatch, streams) + assert [s.seen_context.retry_attempt for s in streams if s.seen_context] == [0, 1, 2] + + +@pytest.mark.asyncio +async def test_blocked_provider_failure_text_carries_request_id( + monkeypatch: pytest.MonkeyPatch, +) -> None: + coordinator = AgentCoordinator() + with pytest.raises(PermissionDeniedError): + await _run(monkeypatch, [_FakeStream(exc=_blocked_error())], coordinator) + assert coordinator.statuses["root"] == "failed" + error = coordinator.errors["root"] + assert "Output blocked by policy" in error + assert error.endswith("[provider request id: req_blocked_hdr]") + + +@pytest.mark.asyncio +async def test_run_agent_loop_binds_agent_context_and_resets( + monkeypatch: pytest.MonkeyPatch, +) -> None: + seen: dict[str, request_log.LlmCallContext] = {} + + async def _fake_loop(**_kwargs: Any) -> None: + seen["ctx"] = request_log.current_call_context() + + monkeypatch.setattr(execution, "_run_agent_loop", _fake_loop) + + class _Agent: + name = "Recon Agent" + + coordinator = AgentCoordinator() + await execution.run_agent_loop( + agent=_Agent(), + initial_input="task", + run_config=cast("RunConfig", object()), + context={}, + max_turns=1, + coordinator=coordinator, + agent_id="agent-42", + interactive=False, + ) + assert seen["ctx"].agent_id == "agent-42" + assert seen["ctx"].agent_name == "Recon Agent" + assert request_log.current_call_context().agent_id is None diff --git a/tests/test_llm_request_log.py b/tests/test_llm_request_log.py new file mode 100644 index 000000000..d82c2e889 --- /dev/null +++ b/tests/test_llm_request_log.py @@ -0,0 +1,1480 @@ +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"] diff --git a/tests/test_models.py b/tests/test_models.py index 7078ca984..061ea13fc 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -23,6 +23,7 @@ from strix.config.models import ( uses_chat_completions_tool_schema, ) from strix.config.settings import Settings +from strix.llm.request_log import RequestLoggingModel @pytest.mark.parametrize("model_name", RECOMMENDED_MODEL_NAMES) @@ -172,7 +173,7 @@ def test_routes_through_litellm_matches_the_provider( # proves the route is not LiteLLM's. assert not litellm return - while isinstance(model, _NonStreamingModel | _TurnGuardModel): + while isinstance(model, _NonStreamingModel | _TurnGuardModel | RequestLoggingModel): model = model._inner assert isinstance(model, LitellmModel) is litellm @@ -215,6 +216,6 @@ def test_api_type_overrides_the_api_base_route( monkeypatch.setenv("STRIX_API_TYPE", api_type) configure_sdk_model_defaults(Settings()) model = StrixProvider().get_model("gpt-5") - while isinstance(model, _NonStreamingModel | _TurnGuardModel): + while isinstance(model, _NonStreamingModel | _TurnGuardModel | RequestLoggingModel): model = model._inner assert isinstance(model, expected)