mirror of
https://github.com/usestrix/strix.git
synced 2026-10-13 03:57:59 +00:00
feat(llm): structured per-attempt provider request log with provider request ids
Every model-provider attempt (LiteLLM callback + native OpenAI/Codex wrapper) emits an LlmRequestEvent: provider, model, host, outcome, HTTP status, provider request id (response header first, Anthropic error body fallback), usage/cost, duration, agent + retry attempt. Prompts, completions, headers and credentials are never recorded; error text is redacted and truncated. Agent failure reasons now go through failure_text() so a blocked Anthropic call's request id lands in the operator-visible failure text.
This commit is contained in:
Ahmed Allam
parent
56e9ae982c
commit
432edf0df6
8 changed files with 1651 additions and 10 deletions
|
|
@ -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"]
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
784
strix/llm/request_log.py
Normal file
784
strix/llm/request_log.py
Normal file
|
|
@ -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",
|
||||
]
|
||||
|
|
@ -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(
|
||||
|
|
|
|||
134
tests/test_execution_request_log.py
Normal file
134
tests/test_execution_request_log.py
Normal file
|
|
@ -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
|
||||
653
tests/test_llm_request_log.py
Normal file
653
tests/test_llm_request_log.py
Normal file
|
|
@ -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"
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue