diff --git a/pyproject.toml b/pyproject.toml index d63d8452..ed8d2cdc 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/models.py b/strix/config/models.py index d8444ee1..f7882c33 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,25 @@ 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) + if not _routes_via_litellm(model): + model = request_log.RequestLoggingModel( + model, + model_name=model_name or llm.model or "unknown", + 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 +558,13 @@ 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) + + DEFAULT_MODEL_RETRY = ModelRetrySettings( max_retries=5, backoff=ModelRetryBackoffSettings( @@ -621,6 +641,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() diff --git a/strix/core/execution.py b/strix/core/execution.py index dfcd39fa..d5d94eb5 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 00000000..c4555cc9 --- /dev/null +++ b/strix/llm/request_log.py @@ -0,0 +1,784 @@ +"""Structured, secret-free 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 record deliberately holds no prompt, completion, header block, or +credential. Error text is redacted and truncated before it is stored. + +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 logging +import re +import time +import uuid +from collections.abc import AsyncIterator, Callable, Mapping +from contextvars import ContextVar, Token +from dataclasses import asdict, dataclass, replace +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any, Literal, cast +from urllib.parse import urlsplit + +from agents.items import ModelResponse +from agents.models.interface import Model +from agents.usage import Usage +from openai import APIError, APIStatusError +from openai.types.responses import ResponseCompletedEvent + + +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 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 + + 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)) + + +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) + + +# --------------------------------------------------------------------------- # +# Redaction # +# --------------------------------------------------------------------------- # + +# Credential shapes a provider or gateway error may echo back. Broad by design: +# a redacted-but-useless message costs nothing, a leaked key costs a rotation. +_SECRET_PATTERNS: tuple[re.Pattern[str], ...] = ( + re.compile(r"(?i)\b(?:bearer|basic)\s+[a-z0-9._~+/=-]{8,}"), + re.compile(r"\bsk-(?:ant-|proj-|or-v1-)?[A-Za-z0-9_-]{8,}"), + re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b"), + re.compile(r"\bAIza[0-9A-Za-z_-]{20,}\b"), + re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}\b"), + re.compile(r"\bxox[abprs]-[A-Za-z0-9-]{10,}\b"), + re.compile(r"\bey[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b"), + re.compile( + r"(?i)([\"']?(?:api[_-]?key|x-api-key|authorization|api[_-]?token|secret|password)" + r"[\"']?\s*[:=]\s*[\"']?)(?:(?:bearer|basic)\s+)?[^\s\"',;&]{4,}" + ), + re.compile(r"(?i)((?:api[_-]?key|token|secret|password)=)[^&\s\"']{4,}"), +) + + +def redact(text: str) -> str: + """Replace anything credential-shaped in ``text``.""" + for pattern in _SECRET_PATTERNS: + if pattern.groups: + text = pattern.sub(lambda m: f"{m.group(1)}[REDACTED]", text) + else: + text = pattern.sub("[REDACTED]", text) + return text + + +def clean_error_message(exc: BaseException) -> str: + message = str(exc).strip() or type(exc).__name__ + message = redact(message) + 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``. 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", + "cf-ray", +) +_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_-]{6,})") + + +def request_id_from_headers(headers: Mapping[str, Any] | None) -> 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 _REQUEST_ID_HEADERS: + if name in normalized: + return normalized[name] + return None + + +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 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 _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 _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. + """ + if exc is None: + return None + candidates = ( + getattr(exc, "litellm_response_headers", None), + getattr(exc, "headers", None), + getattr(getattr(exc, "response", None), "headers", None), + ) + for raw in candidates: + headers = _headers_mapping(raw) + if headers and request_id_from_headers(headers): + return headers + return 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" + host = api_host( + _str_or_none(hidden.get("api_base")) + or _str_or_none(slo.get("api_base")) + or _str_or_none(litellm_params.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()) + ) + + response_headers = _mapping(hidden.get("additional_headers")) + response_hidden = _mapping(getattr(response, "_hidden_params", None)) or {} + request_id = request_id_from_headers(response_headers) or request_id_from_headers( + _mapping(response_hidden.get("additional_headers")) + ) + + status_code: int | None = None + error_type: str | None = None + error_message: str | None = None + response_id: str | None = None + usage: dict[str, int | None] = {} + cost: float | None = None + + if outcome == "success": + status_code = 200 + response_id = _str_or_none(getattr(response, "id", None)) + usage = _litellm_usage(response) + cost = _float_or_none(kwargs.get("response_cost")) + if cost is None: + cost = _float_or_none(hidden.get("response_cost")) + else: + if exc is not None: + status_code = _int_or_none(getattr(exc, "status_code", None)) + error_type = type(exc).__name__ + error_message = clean_error_message(exc) + request_id = request_id or request_id_from_headers(_exception_headers(exc)) + if status_code is None: + code = error_info.get("error_code") + status_code = _int_or_none(code) + if status_code is None and isinstance(code, str) and code.isdigit(): + status_code = int(code) + error_type = error_type or _str_or_none(error_info.get("error_class")) or "Exception" + if error_message is None: + raw = _str_or_none(error_info.get("error_message")) or _str_or_none( + slo.get("error_str") + ) + error_message = redact(raw)[:ERROR_MESSAGE_MAX_CHARS] if raw else error_type + request_id = request_id or request_id_from_text(error_message) + + context = current_call_context() + return LlmRequestEvent( + call_id=call_id, + route="litellm", + provider=provider, + model=model, + api_host=host, + streaming=streaming, + outcome=outcome, + status_code=status_code, + provider_request_id=request_id, + response_id=response_id, + error_type=error_type, + error_message=error_message, + started_at=started, + finished_at=finished, + duration_ms=max(0, int((finished - started).total_seconds() * 1000)), + input_tokens=usage.get("input_tokens"), + output_tokens=usage.get("output_tokens"), + cached_input_tokens=usage.get("cached_input_tokens"), + total_tokens=usage.get("total_tokens"), + cost_usd=cost, + agent_id=context.agent_id, + agent_name=context.agent_name, + retry_attempt=context.retry_attempt, + ) + + +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) and the default log-line sink.""" + global _litellm_logger # noqa: PLW0603 + if _log_line_sink not in _sinks: + _sinks.insert(0, _log_line_sink) + 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 _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 " + "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.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) + if request_id is None: + request_id = request_id_from_headers(_headers_mapping(exc.response.headers)) + if request_id is None: + request_id = request_id_from_text(str(exc)) + return status, request_id + + +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): + """Wrap a Model whose calls do not pass through LiteLLM. + + The native OpenAI Responses / Chat Completions routes (``openai/…``, the + ChatGPT subscription backend) never hit ``litellm.acompletion``, so the + LiteLLM capture does not see them. This wrapper records the same event + around each ``get_response`` and ``stream_response``. + """ + + def __init__(self, inner: Model, *, model_name: str, provider: str, base_url: str | None): + 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) + + @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, + ) -> LlmRequestEvent: + finished = datetime.now(UTC) + duration_ms = max(0, int((time.monotonic() - started_mono) * 1000)) + context = current_call_context() + base = LlmRequestEvent( + call_id=str(uuid.uuid4()), + route="openai", + provider=self._provider, + model=self._model_name, + api_host=self._host, + streaming=streaming, + outcome="success", + status_code=200, + provider_request_id=None, + 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, + ) + if exc is None: + usage = _openai_usage(response) if response is not None else {} + response_id = response.response_id 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"), + ) + status, request_id = _openai_error_fields(exc) + return replace( + base, + outcome="error", + status_code=status, + provider_request_id=request_id, + error_type=type(exc).__name__, + error_message=clean_error_message(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() + 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 asyncio.CancelledError: + raise + except BaseException as exc: + emit( + self._event( + started=started, + started_mono=started_mono, + streaming=False, + response=None, + exc=exc, + ) + ) + raise + emit( + self._event( + started=started, + started_mono=started_mono, + streaming=False, + response=response, + exc=None, + ) + ) + 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() + completed: ModelResponse | None = None + 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 isinstance(event, ResponseCompletedEvent): + 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) + yield event + except asyncio.CancelledError: + raise + except BaseException as exc: + emit( + self._event( + started=started, + started_mono=started_mono, + streaming=True, + response=None, + exc=exc, + ) + ) + raise + emit( + self._event( + started=started, + started_mono=started_mono, + streaming=True, + response=completed, + exc=None, + ) + ) + + +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 = redact(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_headers(_exception_headers(exc)) + 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", + "clean_error_message", + "current_call_context", + "emit", + "event_from_litellm", + "failure_text", + "install", + "redact", + "register_sink", + "request_id_from_headers", + "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 00e99bcc..9bbb13a6 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 00000000..426c4d47 --- /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 00000000..a0201281 --- /dev/null +++ b/tests/test_llm_request_log.py @@ -0,0 +1,653 @@ +from __future__ import annotations + +import asyncio +import logging +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING, Any, ClassVar, cast + +import httpx +import litellm +import pytest +from agents.items import ModelResponse +from agents.models.interface import Model +from agents.usage import Usage +from litellm.llms.anthropic.common_utils import AnthropicError +from openai import APIStatusError, APITimeoutError, PermissionDeniedError +from openai.types.responses import Response, ResponseCompletedEvent, ResponseCreatedEvent + +from strix.llm import request_log + + +if TYPE_CHECKING: + from collections.abc import 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 + + +class _FakeUsage: + prompt_tokens = 120 + completion_tokens = 30 + total_tokens = 150 + prompt_tokens_details = None + cache_read_input_tokens = 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 + + +# --------------------------------------------------------------------------- # +# redaction # +# --------------------------------------------------------------------------- # + + +@pytest.mark.parametrize( + "secret", + [ + "sk-ant-api03-abcdefghijklmnopqrstuvwxyz0123456789", + "sk-proj-abcdefghijklmnopqrstuvwxyz", + "Bearer abcdefghijklmnopqrstuvwxyz0123456789", + "AKIAABCDEFGHIJKLMNOP", + "AIzaSyABCDEFGHIJKLMNOPQRSTUVWXYZ01234", + "ghp_abcdefghijklmnopqrstuvwxyz0123456789", + "xoxb-1234567890-abcdefghij", + "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.abcdefghijklmnopqrstuvwxyz", + ], +) +def test_redact_removes_credential_shapes(secret: str) -> None: + redacted = request_log.redact(f"upstream said: {secret} (401)") + assert secret not in redacted + assert "[REDACTED]" in redacted + assert "(401)" in redacted + + +def test_redact_key_value_forms_keep_the_key_name() -> None: + text = 'invalid "api_key": "abcd1234", authorization: Basic Zm9vOmJhcg== token=abcdef99' + redacted = request_log.redact(text) + assert "abcd1234" not in redacted + assert "Zm9vOmJhcg==" not in redacted + assert "abcdef99" not in redacted + assert '"api_key": "[REDACTED]"' in redacted + + +def test_clean_error_message_truncates_and_redacts() -> None: + exc = RuntimeError("x" * 5000 + " sk-ant-api03-abcdefghijklmnop") + message = request_log.clean_error_message(exc) + assert len(message) <= request_log.ERROR_MESSAGE_MAX_CHARS + assert "sk-ant" not in message + 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_redacted() -> None: + exc = _anthropic_error(401, "invalid x-api-key: sk-ant-api03-abcdefghijklmnopqrstu") + event = request_log.event_from_litellm( + _anthropic_kwargs(exc), None, None, None, outcome="error" + ) + assert event.error_message is not None + assert "sk-ant-api03" not in event.error_message + assert "[REDACTED]" 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 + + +# --------------------------------------------------------------------------- # +# 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_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_not_logged(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 captured == [] + + +@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" + + +@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_and_redacts() -> None: + exc = _anthropic_error( + 401, "bad key sk-ant-api03-abcdefghijklmnopqrstu", {"request-id": "req_exc_hdr"} + ) + text = request_log.failure_text(exc) + assert "sk-ant-api03" not 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" diff --git a/tests/test_models.py b/tests/test_models.py index 7078ca98..061ea13f 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)