mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 17:27:26 +02:00
Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0e28732d29 | ||
|
|
9de747d135 | ||
|
|
b313d78f60 | ||
|
|
e037d8d727 | ||
|
|
fade37025d | ||
|
|
f968f8e5a7 |
@@ -277,6 +277,51 @@ def _configure_litellm_compatibility() -> None:
|
||||
litellm.suppress_debug_info = True
|
||||
|
||||
_register_litellm_cost_callback()
|
||||
_install_openrouter_stream_cost_capture()
|
||||
|
||||
|
||||
def _install_openrouter_stream_cost_capture() -> None:
|
||||
"""Preserve OpenRouter's per-stream cost, which LiteLLM drops when streaming.
|
||||
|
||||
OpenRouter reports the real charge in ``usage.cost`` of the final stream
|
||||
chunk, but LiteLLM rebuilds streamed responses from token-only fields and
|
||||
discards it (its non-streamed path stashes the cost in hidden params; the
|
||||
streaming path does not). Every scan streams, so without this the cost is
|
||||
lost and Strix falls back to a cost-map estimate that is missing entirely
|
||||
for new models (e.g. kimi-k3), reporting $0. Subclass the OpenRouter
|
||||
streaming handler to record the cost keyed by response id so the cost
|
||||
callback can recover the exact charge for the matching rebuilt response.
|
||||
"""
|
||||
import litellm
|
||||
from litellm.llms.openrouter.chat.transformation import (
|
||||
OpenRouterChatCompletionStreamingHandler,
|
||||
OpenrouterConfig,
|
||||
)
|
||||
|
||||
from strix.report.state import streamed_openrouter_costs
|
||||
|
||||
class _StrixOpenRouterStreamingHandler(OpenRouterChatCompletionStreamingHandler):
|
||||
def chunk_parser(self, chunk: dict[str, Any]) -> Any:
|
||||
stream = super().chunk_parser(chunk)
|
||||
streamed_openrouter_costs.remember(
|
||||
chunk.get("id") or getattr(stream, "id", None), chunk.get("usage")
|
||||
)
|
||||
return stream
|
||||
|
||||
class _StrixOpenrouterConfig(OpenrouterConfig):
|
||||
def get_model_response_iterator(
|
||||
self, streaming_response: Any, sync_stream: bool, json_mode: bool | None = False
|
||||
) -> Any:
|
||||
return _StrixOpenRouterStreamingHandler(
|
||||
streaming_response=streaming_response,
|
||||
sync_stream=sync_stream,
|
||||
json_mode=json_mode,
|
||||
)
|
||||
|
||||
# LiteLLM's provider-config factory reads litellm.OpenrouterConfig at call
|
||||
# time, so overriding the attribute is enough for the subclass to take
|
||||
# effect. (type: ignore — mypy rejects reassigning a class attribute.)
|
||||
litellm.OpenrouterConfig = _StrixOpenrouterConfig # type: ignore[misc]
|
||||
|
||||
|
||||
_OPENROUTER_ATTRIBUTION_HEADERS = {
|
||||
|
||||
@@ -40,6 +40,10 @@ class LlmSettings(BaseSettings):
|
||||
default=False,
|
||||
alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
|
||||
)
|
||||
skip_tool_call_probe: bool = Field(
|
||||
default=False,
|
||||
alias="STRIX_SKIP_TOOL_CALL_PROBE",
|
||||
)
|
||||
prompt_cache: bool = Field(
|
||||
default=True,
|
||||
alias="STRIX_PROMPT_CACHE",
|
||||
|
||||
@@ -200,7 +200,9 @@ class AgentCoordinator:
|
||||
logger.info("agent.status %s=%s", agent_id, status)
|
||||
await self._maybe_snapshot()
|
||||
|
||||
async def send(self, target_agent_id: str, message: dict[str, Any]) -> bool:
|
||||
async def send(
|
||||
self, target_agent_id: str, message: dict[str, Any], *, interrupt: bool = True
|
||||
) -> bool:
|
||||
"""Deliver a user/peer message by appending it to the target SDK session."""
|
||||
if message.get("from") == "user" and self._budget_paused:
|
||||
await self.resume_from_budget_pause(exclude=target_agent_id)
|
||||
@@ -211,7 +213,7 @@ class AgentCoordinator:
|
||||
runtime = self.runtimes.setdefault(target_agent_id, AgentRuntime())
|
||||
session = runtime.session
|
||||
stream = runtime.stream
|
||||
interrupt = runtime.interrupt_on_message
|
||||
interrupt_on_message = runtime.interrupt_on_message
|
||||
if session is None:
|
||||
logger.warning(
|
||||
"agent.send dropped target=%s because its SDK session is not attached",
|
||||
@@ -230,7 +232,7 @@ class AgentCoordinator:
|
||||
async with self._lock:
|
||||
self.pending_counts[target_agent_id] = self.pending_counts.get(target_agent_id, 0) + 1
|
||||
self.runtimes.setdefault(target_agent_id, AgentRuntime()).wake.set()
|
||||
if stream is not None and interrupt:
|
||||
if stream is not None and interrupt and interrupt_on_message:
|
||||
stream.cancel(mode="immediate")
|
||||
await self._maybe_snapshot()
|
||||
return True
|
||||
|
||||
+30
-1
@@ -21,6 +21,7 @@ from openai import (
|
||||
RateLimitError,
|
||||
)
|
||||
|
||||
from strix.config import codex
|
||||
from strix.core.hooks import (
|
||||
BudgetExceededError,
|
||||
BudgetPausedError,
|
||||
@@ -88,6 +89,11 @@ async def _compact_session(
|
||||
)
|
||||
|
||||
|
||||
_GUARDRAIL_PARK_ERROR = (
|
||||
"Blocked by the model's content guardrail (flagged as a possible cybersecurity risk). "
|
||||
"Set STRIX_LLM to a model that isn't blocked and resume the scan to continue."
|
||||
)
|
||||
|
||||
_TRANSIENT_MODEL_STATUS_CODES = frozenset({408, 500, 502, 503, 504})
|
||||
_MAX_TRANSIENT_MODEL_RETRIES = 4
|
||||
_TRANSIENT_MODEL_RETRY_BASE_DELAY_S = 2.0
|
||||
@@ -304,6 +310,7 @@ async def respawn_subagents(
|
||||
if coordinator.parent_of.get(aid) is None or aid == root_id:
|
||||
continue
|
||||
md["_restored_status"] = status
|
||||
md["_restored_error"] = coordinator.errors.get(aid)
|
||||
candidates.append(
|
||||
(
|
||||
aid,
|
||||
@@ -316,7 +323,8 @@ async def respawn_subagents(
|
||||
for child_id, name, parent_id, md in candidates:
|
||||
try:
|
||||
restored_status = str(md.get("_restored_status") or "running")
|
||||
start_parked = interactive and restored_status != "running"
|
||||
recoverable_park = restored_status == "waiting" and bool(md.get("_restored_error"))
|
||||
start_parked = interactive and restored_status != "running" and not recoverable_park
|
||||
|
||||
if start_parked:
|
||||
logger.warning(
|
||||
@@ -572,6 +580,10 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
|
||||
if session is not None:
|
||||
input_data = []
|
||||
continue
|
||||
if codex.is_content_guardrail_error(exc):
|
||||
return await _handle_content_guardrail(
|
||||
coordinator, agent_id, exc, interactive=interactive
|
||||
)
|
||||
if not interactive:
|
||||
raise
|
||||
if isinstance(exc, MaxTurnsExceeded):
|
||||
@@ -589,6 +601,22 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
|
||||
return stream
|
||||
|
||||
|
||||
async def _handle_content_guardrail(
|
||||
coordinator: AgentCoordinator,
|
||||
agent_id: str,
|
||||
exc: BaseException,
|
||||
*,
|
||||
interactive: bool,
|
||||
) -> RunResultBase | None:
|
||||
logger.warning("agent %s blocked by the model's content guardrail: %s", agent_id, exc)
|
||||
if interactive:
|
||||
await coordinator.set_status(agent_id, "waiting", error=_GUARDRAIL_PARK_ERROR)
|
||||
return None
|
||||
await coordinator.set_status(agent_id, "failed", error=_GUARDRAIL_PARK_ERROR)
|
||||
await _notify_parent_on_terminal(coordinator, agent_id, "failed")
|
||||
return None
|
||||
|
||||
|
||||
async def _settle_run_result(
|
||||
coordinator: AgentCoordinator,
|
||||
agent_id: str,
|
||||
@@ -685,6 +713,7 @@ async def _notify_parent_on_terminal(
|
||||
"priority": "high",
|
||||
"content": template.format(name=name, agent_id=agent_id),
|
||||
},
|
||||
interrupt=False,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -376,6 +376,12 @@ async def run_strix_scan(
|
||||
|
||||
async with coordinator._lock:
|
||||
root_status = coordinator.statuses.get(root_id)
|
||||
root_error = coordinator.errors.get(root_id)
|
||||
|
||||
root_recoverable_park = root_status == "waiting" and bool(root_error)
|
||||
root_start_parked = bool(
|
||||
interactive and is_resume and root_status != "running" and not root_recoverable_park
|
||||
)
|
||||
|
||||
result = await run_agent_loop(
|
||||
agent=root_agent,
|
||||
@@ -387,7 +393,7 @@ async def run_strix_scan(
|
||||
agent_id=root_id,
|
||||
interactive=interactive,
|
||||
session=root_session,
|
||||
start_parked=bool(interactive and is_resume and root_status != "running"),
|
||||
start_parked=root_start_parked,
|
||||
event_sink=event_sink,
|
||||
hooks=hooks,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,190 @@
|
||||
"""Preflight probe: verify the model emits structured tool calls when streamed.
|
||||
|
||||
Strix is entirely tool-driven and runs every agent turn as a streamed request.
|
||||
Some OpenAI-compatible endpoints return a valid ``tool_calls`` response when a
|
||||
completion is requested non-streamed, but under streaming they emit the tool
|
||||
call as plain assistant text (or drop it entirely) and close the stream. The
|
||||
Agents SDK then sees a normal final message, the scan makes no progress, and it
|
||||
either stalls waiting for input or burns turns on empty output.
|
||||
|
||||
There is no safe client-side way to execute a tool call the endpoint never
|
||||
streamed, so we detect the missing capability up front — using the same
|
||||
streaming path the scan uses — and fail loudly with actionable guidance.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from agents import ModelSettings, ModelTracing
|
||||
from agents.tool import FunctionTool
|
||||
from openai.types.responses import ResponseFunctionToolCall
|
||||
|
||||
from strix.config.models import StrixProvider
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from strix.config.settings import Settings
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_PROBE_RETRIES = 2
|
||||
|
||||
_GUIDANCE = (
|
||||
"The configured LLM endpoint did not return a structured tool call when "
|
||||
"streamed.\n\n"
|
||||
"Strix drives every action through native `tool_calls`, and it streams every "
|
||||
"turn. Some OpenAI-compatible servers return tool calls correctly for a "
|
||||
"non-streamed request but, when streamed, emit the tool call as plain text "
|
||||
"(or omit it) and end the response — so Strix can never act.\n\n"
|
||||
"Fixes:\n"
|
||||
" - llama.cpp / llama-server: start with `--jinja` so the chat template "
|
||||
"produces streamed `tool_calls` deltas.\n"
|
||||
" - Ollama: use a model whose template wires tool calling, and disable "
|
||||
"'thinking' if the template can't stream tools alongside it.\n"
|
||||
" - vLLM: set a matching `--tool-call-parser` (and `--enable-auto-tool-choice`) "
|
||||
"for the served model.\n"
|
||||
" - Other gateways: confirm streamed tool calling works for this model "
|
||||
"(a non-streamed test is not enough).\n\n"
|
||||
"If you know the endpoint streams tool calls correctly, set "
|
||||
"STRIX_SKIP_TOOL_CALL_PROBE=1 to skip this check."
|
||||
)
|
||||
|
||||
_TOOL_CONFIG_ERROR_MARKERS = (
|
||||
"jinja",
|
||||
"tool call parser",
|
||||
"tool-call-parser",
|
||||
"tool_choice",
|
||||
"does not support tools",
|
||||
"tools param",
|
||||
"tool use is not supported",
|
||||
)
|
||||
|
||||
|
||||
class ToolCallingUnsupportedError(RuntimeError):
|
||||
"""The endpoint cannot return structured tool calls over a streamed request."""
|
||||
|
||||
|
||||
async def _noop_invoke(_ctx: Any, _args: str) -> str:
|
||||
return "ok"
|
||||
|
||||
|
||||
_PROBE_TOOL = FunctionTool(
|
||||
name="strix_ready_check",
|
||||
description="Report readiness. Call this to acknowledge you can use tools.",
|
||||
params_json_schema={
|
||||
"type": "object",
|
||||
"additionalProperties": False,
|
||||
"properties": {"status": {"type": "string"}},
|
||||
"required": ["status"],
|
||||
},
|
||||
on_invoke_tool=_noop_invoke,
|
||||
strict_json_schema=True,
|
||||
)
|
||||
|
||||
_PROBE_SYSTEM = (
|
||||
"You are a setup probe. You can only respond by calling the provided tool. "
|
||||
"Do not produce any other output."
|
||||
)
|
||||
_PROBE_INPUT = 'Call the `strix_ready_check` tool now with {"status": "ok"}.'
|
||||
|
||||
|
||||
def requires_tool_call_probe(model_name: str, settings: Settings) -> bool:
|
||||
"""Only self-hosted / OpenAI-compatible routes, where the streaming leak happens."""
|
||||
return model_name.startswith("ollama/") or bool(settings.llm.api_base)
|
||||
|
||||
|
||||
def _is_tool_config_error(exc: Exception) -> bool:
|
||||
text = str(exc).lower()
|
||||
return any(marker in text for marker in _TOOL_CONFIG_ERROR_MARKERS)
|
||||
|
||||
|
||||
async def _stream_saw_tool_call(
|
||||
model_name: str, model_settings: ModelSettings, *, timeout: float | None
|
||||
) -> bool:
|
||||
model = StrixProvider().get_model(model_name)
|
||||
|
||||
async def _run() -> bool:
|
||||
stream = model.stream_response(
|
||||
system_instructions=_PROBE_SYSTEM,
|
||||
input=_PROBE_INPUT,
|
||||
model_settings=model_settings,
|
||||
tools=[_PROBE_TOOL],
|
||||
output_schema=None,
|
||||
handoffs=[],
|
||||
tracing=ModelTracing.DISABLED,
|
||||
previous_response_id=None,
|
||||
conversation_id=None,
|
||||
prompt=None,
|
||||
)
|
||||
# Drain the whole stream rather than returning early: the SDK wraps it in
|
||||
# a span that must be closed in the context it was opened in.
|
||||
saw_tool_call = False
|
||||
async for event in stream:
|
||||
item = getattr(event, "item", None)
|
||||
if isinstance(item, ResponseFunctionToolCall):
|
||||
saw_tool_call = True
|
||||
response = getattr(event, "response", None)
|
||||
if response is not None and any(
|
||||
isinstance(out, ResponseFunctionToolCall)
|
||||
for out in getattr(response, "output", []) or []
|
||||
):
|
||||
saw_tool_call = True
|
||||
return saw_tool_call
|
||||
|
||||
if timeout is not None:
|
||||
return await asyncio.wait_for(_run(), timeout=timeout)
|
||||
return await _run()
|
||||
|
||||
|
||||
async def probe_tool_calling(
|
||||
model_name: str,
|
||||
settings: Settings,
|
||||
*,
|
||||
request_timeout: float | None = None,
|
||||
) -> None:
|
||||
"""Fail fast if a streamed request to ``model_name`` yields no structured tool call.
|
||||
|
||||
No-op for hosted providers and when ``STRIX_SKIP_TOOL_CALL_PROBE`` is set.
|
||||
"""
|
||||
if settings.llm.skip_tool_call_probe or not requires_tool_call_probe(model_name, settings):
|
||||
return
|
||||
|
||||
model_settings = ModelSettings(
|
||||
parallel_tool_calls=False,
|
||||
include_usage=True,
|
||||
)
|
||||
|
||||
last_exc: Exception | None = None
|
||||
for attempt in range(_PROBE_RETRIES + 1):
|
||||
try:
|
||||
saw_tool_call = await _stream_saw_tool_call(
|
||||
model_name, model_settings, timeout=request_timeout
|
||||
)
|
||||
except ToolCallingUnsupportedError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
if _is_tool_config_error(exc):
|
||||
logger.debug("Tool-call probe hit a tool-config error", exc_info=True)
|
||||
raise ToolCallingUnsupportedError(_GUIDANCE) from exc
|
||||
last_exc = exc
|
||||
logger.debug(
|
||||
"Tool-call probe attempt %d/%d failed transiently",
|
||||
attempt + 1,
|
||||
_PROBE_RETRIES + 1,
|
||||
exc_info=True,
|
||||
)
|
||||
continue
|
||||
|
||||
if saw_tool_call:
|
||||
logger.info("Tool-call probe passed for model %s", model_name)
|
||||
return
|
||||
raise ToolCallingUnsupportedError(_GUIDANCE)
|
||||
|
||||
# All attempts raised transient errors; surface the last one unchanged so the
|
||||
# caller's existing connection-error handling reports it.
|
||||
if last_exc is not None:
|
||||
raise last_exc
|
||||
+23
-1
@@ -33,6 +33,7 @@ from strix.config.models import (
|
||||
)
|
||||
from strix.core.inputs import DEFAULT_MAX_TURNS
|
||||
from strix.core.paths import run_dir_for, runtime_state_dir
|
||||
from strix.core.warmup import ToolCallingUnsupportedError, probe_tool_calling
|
||||
from strix.interface.cli import run_cli
|
||||
from strix.interface.tui import run_tui
|
||||
from strix.interface.update_check import (
|
||||
@@ -422,6 +423,27 @@ async def warm_up_llm(show_model_warning: bool = True) -> None:
|
||||
)
|
||||
logger.info("LLM warm-up succeeded for dedupe model %s", dedupe_model)
|
||||
|
||||
raw_model = (llm.model or "").strip()
|
||||
await probe_tool_calling(raw_model, settings, request_timeout=llm.timeout)
|
||||
|
||||
except ToolCallingUnsupportedError as e:
|
||||
logger.debug("Tool-call probe failed", exc_info=True)
|
||||
error_text = Text()
|
||||
error_text.append("TOOL CALLING NOT SUPPORTED", style="bold red")
|
||||
error_text.append("\n\n", style="white")
|
||||
error_text.append(str(e), style="white")
|
||||
console.print("\n")
|
||||
console.print(
|
||||
Panel(
|
||||
error_text,
|
||||
title="[bold white]STRIX",
|
||||
title_align="left",
|
||||
border_style="red",
|
||||
padding=(1, 2),
|
||||
),
|
||||
)
|
||||
console.print()
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
logger.debug("LLM warm-up failed", exc_info=True)
|
||||
error_text = Text()
|
||||
@@ -884,7 +906,7 @@ def display_completion_message(args: argparse.Namespace, results_path: Path) ->
|
||||
view_text = Text()
|
||||
view_text.append("\n")
|
||||
view_text.append("View", style="dim")
|
||||
view_text.append(" ")
|
||||
view_text.append(" ")
|
||||
view_text.append(f"strix view {args.run_name}", style="#22c55e")
|
||||
panel_parts.extend(["\n", view_text])
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@ class TuiLiveView:
|
||||
self.events: list[dict[str, Any]] = []
|
||||
self._next_event_id = 1
|
||||
self._open_assistant_event_by_agent: dict[str, dict[str, Any]] = {}
|
||||
self._tool_event_by_call_id: dict[str, dict[str, Any]] = {}
|
||||
self._tool_event_by_agent_and_call_id: dict[tuple[str, str], dict[str, Any]] = {}
|
||||
|
||||
def hydrate_from_run_dir(self, run_dir: Path) -> None:
|
||||
state_dir = runtime_state_dir(run_dir)
|
||||
@@ -223,7 +223,8 @@ class TuiLiveView:
|
||||
timestamp: str | None = None,
|
||||
) -> None:
|
||||
call_id = call["call_id"]
|
||||
existing = self._tool_event_by_call_id.get(call_id)
|
||||
event_key = (agent_id, call_id)
|
||||
existing = self._tool_event_by_agent_and_call_id.get(event_key)
|
||||
tool_data = {
|
||||
"tool_name": call["tool_name"],
|
||||
"args": call["args"],
|
||||
@@ -233,7 +234,7 @@ class TuiLiveView:
|
||||
}
|
||||
if existing is None:
|
||||
event = self._append_event(agent_id, "tool", tool_data, timestamp=timestamp)
|
||||
self._tool_event_by_call_id[call_id] = event
|
||||
self._tool_event_by_agent_and_call_id[event_key] = event
|
||||
else:
|
||||
existing["data"].update(tool_data)
|
||||
self._bump_event(existing, timestamp=timestamp)
|
||||
@@ -249,7 +250,8 @@ class TuiLiveView:
|
||||
timestamp: str | None = None,
|
||||
) -> None:
|
||||
call_id = output["call_id"]
|
||||
event = self._tool_event_by_call_id.get(call_id)
|
||||
event_key = (agent_id, call_id)
|
||||
event = self._tool_event_by_agent_and_call_id.get(event_key)
|
||||
if event is None:
|
||||
event = self._append_event(
|
||||
agent_id,
|
||||
@@ -263,7 +265,7 @@ class TuiLiveView:
|
||||
},
|
||||
timestamp=timestamp,
|
||||
)
|
||||
self._tool_event_by_call_id[call_id] = event
|
||||
self._tool_event_by_agent_and_call_id[event_key] = event
|
||||
|
||||
result = _parse_json_value(output["output"])
|
||||
event["data"]["result"] = result
|
||||
|
||||
@@ -107,8 +107,11 @@ def resolve_run_dir(base_dir: Path, run_param: str | None, default_run_dir: Path
|
||||
return candidate
|
||||
|
||||
|
||||
# Name of the cookie carrying the per-process session capability.
|
||||
SESSION_COOKIE = "strix_viewer_session"
|
||||
# Prefix of the cookie carrying the per-process session capability. The bound
|
||||
# port is appended (``strix_viewer_session_<port>``) because browsers scope
|
||||
# cookies by host only, never by port: concurrent viewers on 127.0.0.1 would
|
||||
# otherwise share one cookie slot and clobber each other's session.
|
||||
SESSION_COOKIE_PREFIX = "strix_viewer_session"
|
||||
|
||||
|
||||
class _ViewerState:
|
||||
@@ -135,6 +138,9 @@ class _ViewerState:
|
||||
# enough to steer a live scan, trigger a report, or browse history --
|
||||
# the token is never handed to a caller who merely reaches ``/``.
|
||||
self.session_token = secrets.token_urlsafe(32)
|
||||
# Finalized in ``serve()`` once the port is known (the server binds
|
||||
# after this state is constructed); see SESSION_COOKIE_PREFIX.
|
||||
self.cookie_name = SESSION_COOKIE_PREFIX
|
||||
|
||||
|
||||
def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
|
||||
@@ -476,7 +482,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
|
||||
the browser this process handed the page to can pass. A direct
|
||||
caller on an exposed port has no cookie and is rejected.
|
||||
"""
|
||||
supplied = self._cookies().get(SESSION_COOKIE, "")
|
||||
supplied = self._cookies().get(state.cookie_name, "")
|
||||
return bool(supplied) and secrets.compare_digest(supplied, state.session_token)
|
||||
|
||||
def _token_presented(self, query: dict[str, list[str]]) -> bool:
|
||||
@@ -512,7 +518,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
|
||||
# SameSite=Strict (never sent from a cross-site context).
|
||||
self.send_header(
|
||||
"Set-Cookie",
|
||||
f"{SESSION_COOKIE}={state.session_token}; Path=/; HttpOnly; SameSite=Strict",
|
||||
f"{state.cookie_name}={state.session_token}; Path=/; HttpOnly; SameSite=Strict",
|
||||
)
|
||||
self.end_headers()
|
||||
self.wfile.write(content)
|
||||
@@ -586,6 +592,7 @@ def serve(
|
||||
|
||||
httpd.daemon_threads = True
|
||||
bound_port = int(httpd.server_address[1])
|
||||
state.cookie_name = f"{SESSION_COOKIE_PREFIX}_{bound_port}"
|
||||
url = f"http://{host}:{bound_port}"
|
||||
|
||||
thread = threading.Thread(target=httpd.serve_forever, name="strix-viewer", daemon=True)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import json
|
||||
import logging
|
||||
import subprocess
|
||||
import threading
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime
|
||||
from importlib.metadata import PackageNotFoundError, version
|
||||
@@ -95,6 +96,8 @@ def get_global_report_state() -> Optional["ReportState"]:
|
||||
def set_global_report_state(report_state: "ReportState") -> None:
|
||||
global _global_report_state # noqa: PLW0603
|
||||
_global_report_state = report_state
|
||||
# New run: drop any streamed-cost entries a prior run left unconsumed.
|
||||
streamed_openrouter_costs.clear()
|
||||
|
||||
|
||||
class ReportState:
|
||||
@@ -507,6 +510,72 @@ class ReportState:
|
||||
self._sync_llm_usage_record()
|
||||
|
||||
|
||||
def openrouter_stream_cost(usage: Any) -> float | None:
|
||||
"""Total OpenRouter-reported cost from a raw stream ``usage`` block, or None.
|
||||
|
||||
Non-BYOK responses bill everything to ``usage.cost``. BYOK responses put the
|
||||
OpenRouter fee in ``usage.cost`` (often 0) and the provider charge in
|
||||
``usage.cost_details.upstream_inference_cost``, so BYOK totals sum the two.
|
||||
"""
|
||||
if not isinstance(usage, dict):
|
||||
return None
|
||||
total = 0.0
|
||||
cost = usage.get("cost")
|
||||
if isinstance(cost, int | float) and cost > 0:
|
||||
total += float(cost)
|
||||
if bool(usage.get("is_byok")):
|
||||
details = usage.get("cost_details")
|
||||
upstream = details.get("upstream_inference_cost") if isinstance(details, dict) else None
|
||||
if isinstance(upstream, int | float) and upstream > 0:
|
||||
total += float(upstream)
|
||||
return total if total > 0 else None
|
||||
|
||||
|
||||
def _response_id(completion_response: Any) -> str | None:
|
||||
response_id = getattr(completion_response, "id", None)
|
||||
if response_id is None and isinstance(completion_response, dict):
|
||||
response_id = cast("dict[str, Any]", completion_response).get("id")
|
||||
return response_id if isinstance(response_id, str) and response_id else None
|
||||
|
||||
|
||||
class StreamedOpenRouterCosts:
|
||||
"""Correlates OpenRouter's per-stream cost from the parser to the cost callback.
|
||||
|
||||
LiteLLM rebuilds streamed responses from token-only chunks and drops the
|
||||
``usage.cost`` OpenRouter reports in its final stream chunk (its non-streamed
|
||||
path preserves it; streaming snapshots hidden params at stream start). Every
|
||||
scan streams, so the OpenRouter streaming handler (see strix.config.models)
|
||||
records the cost here keyed by response id, and the callback takes it back out
|
||||
for the matching rebuilt response. Entries are removed on read; ``clear()``
|
||||
runs per scan so nothing accumulates across runs.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._costs: dict[str, float] = {}
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def remember(self, response_id: Any, usage: Any) -> None:
|
||||
cost = openrouter_stream_cost(usage)
|
||||
if cost is None or not (isinstance(response_id, str) and response_id):
|
||||
return
|
||||
with self._lock:
|
||||
self._costs[response_id] = cost
|
||||
|
||||
def take(self, completion_response: Any) -> float | None:
|
||||
response_id = _response_id(completion_response)
|
||||
if response_id is None:
|
||||
return None
|
||||
with self._lock:
|
||||
return self._costs.pop(response_id, None)
|
||||
|
||||
def clear(self) -> None:
|
||||
with self._lock:
|
||||
self._costs.clear()
|
||||
|
||||
|
||||
streamed_openrouter_costs = StreamedOpenRouterCosts()
|
||||
|
||||
|
||||
def litellm_cost_callback(
|
||||
kwargs: Any,
|
||||
completion_response: Any,
|
||||
@@ -541,6 +610,11 @@ def litellm_cost_callback(
|
||||
if cost is None:
|
||||
cost = _usage_reported_cost(completion_response)
|
||||
|
||||
# Recover the exact OpenRouter cost the streaming handler stashed for this
|
||||
# response — LiteLLM drops it from streamed usage, so nothing above sees it.
|
||||
if cost is None:
|
||||
cost = streamed_openrouter_costs.take(completion_response)
|
||||
|
||||
if cost is None:
|
||||
cost = _estimate_response_cost(kwargs, completion_response)
|
||||
|
||||
|
||||
+105
-2
@@ -7,9 +7,25 @@ from unittest.mock import MagicMock, patch
|
||||
|
||||
import litellm
|
||||
import pytest
|
||||
from litellm.types.utils import LlmProviders
|
||||
from litellm.utils import ProviderConfigManager
|
||||
|
||||
from strix.config.models import _configure_litellm_compatibility
|
||||
from strix.report.state import litellm_cost_callback
|
||||
from strix.config.models import (
|
||||
_configure_litellm_compatibility,
|
||||
_install_openrouter_stream_cost_capture,
|
||||
)
|
||||
from strix.report.state import (
|
||||
ReportState,
|
||||
litellm_cost_callback,
|
||||
openrouter_stream_cost,
|
||||
set_global_report_state,
|
||||
streamed_openrouter_costs,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_streamed_costs() -> None:
|
||||
streamed_openrouter_costs.clear()
|
||||
|
||||
|
||||
def test_streaming_logging_stays_enabled_for_cost_callback() -> None:
|
||||
@@ -151,3 +167,90 @@ def test_cost_callback_records_nothing_when_no_cost_available() -> None:
|
||||
litellm_cost_callback({"response_cost": None, "model": "x/y"}, response)
|
||||
|
||||
report_state.record_observed_llm_cost.assert_not_called()
|
||||
|
||||
|
||||
def test_openrouter_stream_cost_extracts_plain_and_byok_totals() -> None:
|
||||
assert openrouter_stream_cost({"cost": 0.003168}) == pytest.approx(0.003168)
|
||||
assert openrouter_stream_cost(
|
||||
{"cost": 0.01, "is_byok": True, "cost_details": {"upstream_inference_cost": 0.2}}
|
||||
) == pytest.approx(0.21)
|
||||
# Upstream cost is only added for BYOK responses.
|
||||
assert openrouter_stream_cost(
|
||||
{"cost": 0.05, "is_byok": False, "cost_details": {"upstream_inference_cost": 0.04}}
|
||||
) == pytest.approx(0.05)
|
||||
assert openrouter_stream_cost({"prompt_tokens": 10}) is None
|
||||
assert openrouter_stream_cost(None) is None
|
||||
|
||||
|
||||
def test_cost_callback_recovers_streamed_openrouter_cost_by_response_id() -> None:
|
||||
report_state = MagicMock()
|
||||
streamed_openrouter_costs.remember("gen-abc", {"cost": 0.42})
|
||||
# LiteLLM strips cost from the rebuilt streamed usage; only the id survives.
|
||||
response = SimpleNamespace(id="gen-abc", usage=SimpleNamespace(cost=None), _hidden_params={})
|
||||
|
||||
with (
|
||||
patch("strix.report.state.get_global_report_state", return_value=report_state),
|
||||
patch("litellm.completion_cost", side_effect=ValueError("unknown model")),
|
||||
):
|
||||
litellm_cost_callback({"response_cost": None, "model": "moonshotai/kimi-k3"}, response)
|
||||
|
||||
report_state.record_observed_llm_cost.assert_called_once_with(0.42)
|
||||
# The entry is consumed so a later response cannot double-count it.
|
||||
assert streamed_openrouter_costs.take(response) is None
|
||||
|
||||
|
||||
def test_streamed_openrouter_cost_prefers_provider_report_over_estimate() -> None:
|
||||
report_state = MagicMock()
|
||||
streamed_openrouter_costs.remember("gen-xyz", {"cost": 0.9})
|
||||
response = SimpleNamespace(
|
||||
id="gen-xyz",
|
||||
usage=SimpleNamespace(prompt_tokens=10, completion_tokens=5, total_tokens=15),
|
||||
_hidden_params={},
|
||||
)
|
||||
|
||||
with (
|
||||
patch("strix.report.state.get_global_report_state", return_value=report_state),
|
||||
patch("litellm.completion_cost", return_value=0.1) as estimate,
|
||||
):
|
||||
litellm_cost_callback({"response_cost": None, "model": "moonshotai/kimi-k3"}, response)
|
||||
|
||||
report_state.record_observed_llm_cost.assert_called_once_with(0.9)
|
||||
estimate.assert_not_called()
|
||||
|
||||
|
||||
def test_streamed_openrouter_costs_ignores_entries_without_cost() -> None:
|
||||
streamed_openrouter_costs.remember("gen-none", {"prompt_tokens": 10})
|
||||
streamed_openrouter_costs.remember("", {"cost": 0.5})
|
||||
assert streamed_openrouter_costs.take(SimpleNamespace(id="gen-none")) is None
|
||||
|
||||
|
||||
def test_streamed_openrouter_costs_cleared_on_new_run() -> None:
|
||||
streamed_openrouter_costs.remember("gen-stale", {"cost": 0.7})
|
||||
set_global_report_state(ReportState.__new__(ReportState))
|
||||
assert streamed_openrouter_costs.take(SimpleNamespace(id="gen-stale")) is None
|
||||
|
||||
|
||||
def test_openrouter_stream_handler_records_cost() -> None:
|
||||
_install_openrouter_stream_cost_capture()
|
||||
# Resolve the config the way LiteLLM does in production so we prove the
|
||||
# override is actually reachable through provider resolution, not just as a
|
||||
# directly-constructed class.
|
||||
config = ProviderConfigManager.get_provider_chat_config(
|
||||
model="moonshotai/kimi-k3", provider=LlmProviders.OPENROUTER
|
||||
)
|
||||
assert config is not None
|
||||
assert type(config).__name__ == "_StrixOpenrouterConfig"
|
||||
handler = config.get_model_response_iterator(streaming_response=iter([]), sync_stream=True)
|
||||
|
||||
chunk = {
|
||||
"id": "gen-stream",
|
||||
"created": 1,
|
||||
"model": "moonshotai/kimi-k3",
|
||||
"choices": [{"index": 0, "delta": {"content": None}}],
|
||||
"usage": {"prompt_tokens": 89, "completion_tokens": 138, "cost": 0.0035055},
|
||||
}
|
||||
handler.chunk_parser(chunk)
|
||||
|
||||
assert streamed_openrouter_costs.take(SimpleNamespace(id="gen-stream")) == pytest.approx(
|
||||
0.0035055
|
||||
)
|
||||
|
||||
+112
-1
@@ -6,13 +6,21 @@ import asyncio
|
||||
import contextlib
|
||||
import json
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from agents.memory import SQLiteSession
|
||||
from agents.tool_context import ToolContext
|
||||
|
||||
from strix.config import codex
|
||||
from strix.core import execution
|
||||
from strix.core.agents import AgentCoordinator
|
||||
from strix.core.execution import _notify_parent_on_terminal, _notify_root_on_budget_reserve
|
||||
from strix.core.execution import (
|
||||
_handle_content_guardrail,
|
||||
_notify_parent_on_terminal,
|
||||
_notify_root_on_budget_reserve,
|
||||
respawn_subagents,
|
||||
)
|
||||
from strix.tools.finish.tool import finish_scan
|
||||
|
||||
|
||||
@@ -465,3 +473,106 @@ async def test_notify_parent_on_terminal_ignores_non_terminal_status(tmp_path: A
|
||||
|
||||
assert coordinator.pending_counts.get("root", 0) == 0
|
||||
session.close()
|
||||
|
||||
|
||||
class _RecordingStream:
|
||||
def __init__(self) -> None:
|
||||
self.cancelled = False
|
||||
self.cancel_mode: str | None = None
|
||||
|
||||
def cancel(self, mode: str = "immediate") -> None:
|
||||
self.cancelled = True
|
||||
self.cancel_mode = mode
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_terminal_notice_does_not_cancel_parent_stream(tmp_path: Any) -> None:
|
||||
coordinator = AgentCoordinator()
|
||||
await coordinator.register("root", "strix", parent_id=None)
|
||||
await coordinator.register("child", "recon", parent_id="root")
|
||||
session = SQLiteSession("root", tmp_path / "agents.db")
|
||||
stream = _RecordingStream()
|
||||
await coordinator.attach_runtime("root", session=session, interrupt_on_message=True)
|
||||
await coordinator.attach_stream("root", stream)
|
||||
|
||||
await _notify_parent_on_terminal(coordinator, "child", "crashed")
|
||||
|
||||
assert stream.cancelled is False
|
||||
assert coordinator.pending_counts.get("root", 0) > 0
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_guardrail_interactive_parks_agent_wakeable(tmp_path: Any) -> None:
|
||||
coordinator = AgentCoordinator()
|
||||
await coordinator.register("root", "strix", parent_id=None)
|
||||
await coordinator.register("child", "recon", parent_id="root")
|
||||
exc = codex.CodexContentGuardrailError("chatgpt/gpt-5.6-sol")
|
||||
|
||||
result = await _handle_content_guardrail(coordinator, "child", exc, interactive=True)
|
||||
|
||||
assert result is None
|
||||
assert coordinator.statuses["child"] == "waiting"
|
||||
assert "STRIX_LLM" in coordinator.errors["child"]
|
||||
|
||||
waiter = asyncio.create_task(coordinator.wait_for_message("child"))
|
||||
await asyncio.sleep(0)
|
||||
assert not waiter.done()
|
||||
session = SQLiteSession("child", tmp_path / "agents.db")
|
||||
await coordinator.attach_runtime("child", session=session)
|
||||
await coordinator.send("child", {"from": "user", "content": "switched model, resume"})
|
||||
await asyncio.wait_for(waiter, timeout=1.0)
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_guardrail_noninteractive_fails_only_blocked_agent(tmp_path: Any) -> None:
|
||||
coordinator = AgentCoordinator()
|
||||
await coordinator.register("root", "strix", parent_id=None)
|
||||
await coordinator.register("child", "recon", parent_id="root")
|
||||
session = SQLiteSession("root", tmp_path / "agents.db")
|
||||
await coordinator.attach_runtime("root", session=session)
|
||||
exc = codex.CodexContentGuardrailError("chatgpt/gpt-5.6-sol")
|
||||
|
||||
result = await _handle_content_guardrail(coordinator, "child", exc, interactive=False)
|
||||
|
||||
assert result is None
|
||||
assert coordinator.statuses["child"] == "failed"
|
||||
assert "STRIX_LLM" in coordinator.errors["child"]
|
||||
assert coordinator.statuses["root"] == "running"
|
||||
assert coordinator.pending_counts.get("root", 0) > 0
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resume_revives_guardrail_parked_child_but_not_plain_waiting(
|
||||
tmp_path: Any, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
coordinator = AgentCoordinator()
|
||||
await coordinator.register("root", "strix", parent_id=None)
|
||||
await coordinator.register("blocked", "recon", parent_id="root")
|
||||
await coordinator.register("peer_waiter", "recon", parent_id="root")
|
||||
await coordinator.set_status("blocked", "waiting", error="STRIX_LLM guardrail")
|
||||
await coordinator.set_status("peer_waiter", "waiting")
|
||||
|
||||
parked: dict[str, bool] = {}
|
||||
|
||||
async def _fake_start_child_runner(**kwargs: Any) -> None:
|
||||
parked[kwargs["child_id"]] = bool(kwargs["start_parked"])
|
||||
|
||||
monkeypatch.setattr(execution, "_start_child_runner", _fake_start_child_runner)
|
||||
|
||||
await respawn_subagents(
|
||||
coordinator=coordinator,
|
||||
factory=lambda **_kwargs: object(),
|
||||
agents_db_path=tmp_path / "agents.db",
|
||||
sessions_to_close=[],
|
||||
run_config=MagicMock(),
|
||||
max_turns=10,
|
||||
interactive=True,
|
||||
parent_ctx={"agent_id": "root", "parent_id": None},
|
||||
root_id="root",
|
||||
)
|
||||
|
||||
assert parked["blocked"] is False
|
||||
assert parked["peer_waiter"] is True
|
||||
|
||||
+125
-2
@@ -4,9 +4,11 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from typing import TYPE_CHECKING
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from strix.core.paths import latest_run_dir, runs_base_dir
|
||||
from strix.interface.viewer.server import serve
|
||||
@@ -86,6 +88,75 @@ def test_build_run_state_from_agents_json(tmp_path: Path) -> None:
|
||||
assert state["events"] == []
|
||||
|
||||
|
||||
def test_build_run_state_keeps_same_call_id_separate_per_agent(tmp_path: Path) -> None:
|
||||
run_dir = _make_run(tmp_path, "tools", status="completed", end_time=None)
|
||||
agents_db = run_dir / ".state" / "agents.db"
|
||||
rows = [
|
||||
(
|
||||
"root",
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "exec_command_0",
|
||||
"name": "exec_command",
|
||||
"arguments": json.dumps({"cmd": "echo root"}),
|
||||
},
|
||||
),
|
||||
(
|
||||
"root",
|
||||
{
|
||||
"type": "function_call_output",
|
||||
"call_id": "exec_command_0",
|
||||
"output": json.dumps({"success": True, "output": "root"}),
|
||||
},
|
||||
),
|
||||
(
|
||||
"child",
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "exec_command_0",
|
||||
"name": "exec_command",
|
||||
"arguments": json.dumps({"cmd": "echo child"}),
|
||||
},
|
||||
),
|
||||
(
|
||||
"child",
|
||||
{
|
||||
"type": "function_call_output",
|
||||
"call_id": "exec_command_0",
|
||||
"output": json.dumps({"success": True, "output": "child"}),
|
||||
},
|
||||
),
|
||||
]
|
||||
with sqlite3.connect(agents_db) as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
create table agent_messages (
|
||||
id integer primary key,
|
||||
session_id text not null,
|
||||
message_data text not null,
|
||||
created_at text not null
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.executemany(
|
||||
"""
|
||||
insert into agent_messages (session_id, message_data, created_at)
|
||||
values (?, ?, '2026-01-01T00:00:00+00:00')
|
||||
""",
|
||||
[(agent_id, json.dumps(message)) for agent_id, message in rows],
|
||||
)
|
||||
|
||||
state = build_run_state(run_dir)
|
||||
tools = [event for event in state["events"] if event["type"] == "tool"]
|
||||
|
||||
assert len(tools) == 2
|
||||
by_agent = {event["agent_id"]: event for event in tools}
|
||||
assert by_agent["root"]["data"]["args"] == {"cmd": "echo root"}
|
||||
assert by_agent["root"]["data"]["result"]["output"] == "root"
|
||||
assert by_agent["child"]["data"]["args"] == {"cmd": "echo child"}
|
||||
assert by_agent["child"]["data"]["result"]["output"] == "child"
|
||||
|
||||
|
||||
def _get(url: str, *, cookie: str | None = None) -> tuple[int, str, bytes]:
|
||||
headers = {"Cookie": cookie} if cookie else {}
|
||||
req = urllib.request.Request(url, headers=headers) # noqa: S310 - localhost test server
|
||||
@@ -271,6 +342,11 @@ def _session_cookie(url: str, token: str) -> str:
|
||||
return raw.split(";", 1)[0]
|
||||
|
||||
|
||||
def _cookie_name(url: str) -> str:
|
||||
"""The per-server session cookie name, derived from the bound port."""
|
||||
return f"strix_viewer_session_{urlsplit(url).port}"
|
||||
|
||||
|
||||
def _get_status(url: str, *, cookie: str | None = None) -> int:
|
||||
headers = {"Cookie": cookie} if cookie else {}
|
||||
req = urllib.request.Request(url, headers=headers) # noqa: S310 - localhost test server
|
||||
@@ -311,7 +387,7 @@ def test_capability_issued_only_for_tokened_bootstrap(
|
||||
# Only the correct bootstrap token mints the session cookie.
|
||||
with urllib.request.urlopen(f"{url}/?token={token}") as resp: # noqa: S310 # nosec B310
|
||||
cookie = str(resp.headers.get("Set-Cookie", ""))
|
||||
assert "strix_viewer_session=" in cookie
|
||||
assert f"{_cookie_name(url)}=" in cookie
|
||||
assert "HttpOnly" in cookie and "SameSite=Strict" in cookie
|
||||
|
||||
# Static assets never carry it.
|
||||
@@ -344,7 +420,7 @@ def test_unauthorized_client_cannot_acquire_capability(
|
||||
url,
|
||||
"/api/agents/steer",
|
||||
{"agent_id": "root", "message": "pwn"},
|
||||
cookie="strix_viewer_session=",
|
||||
cookie=f"{_cookie_name(url)}=",
|
||||
)
|
||||
assert status == 403
|
||||
assert delivered == []
|
||||
@@ -541,6 +617,53 @@ def test_runs_list_requires_session_and_verification(
|
||||
httpd.server_close()
|
||||
|
||||
|
||||
def test_concurrent_servers_use_distinct_cookies(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Cookies are host-scoped, not port-scoped: two viewers on 127.0.0.1 must
|
||||
not share a cookie slot, and one server's cookie must not pass the other's
|
||||
session gate."""
|
||||
run_a = _make_run(tmp_path / "a", "run-a", status="running", end_time=None)
|
||||
run_b = _make_run(tmp_path / "b", "run-b", status="running", end_time=None)
|
||||
_bundle(tmp_path, monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
"strix.interface.viewer.auth.read_auth", lambda: {"email": "a@b.com", "token": "t"}
|
||||
)
|
||||
monkeypatch.setattr("strix.interface.viewer.auth.is_verified", lambda: True)
|
||||
|
||||
httpd_a, url_a, token_a = serve(run_a, open_browser=False)
|
||||
httpd_b, url_b, token_b = serve(run_b, open_browser=False)
|
||||
try:
|
||||
cookie_a = _session_cookie(url_a, token_a)
|
||||
cookie_b = _session_cookie(url_b, token_b)
|
||||
|
||||
# The two servers mint differently named cookies, so a browser stores both.
|
||||
assert cookie_a.split("=", 1)[0] == _cookie_name(url_a)
|
||||
assert cookie_b.split("=", 1)[0] == _cookie_name(url_b)
|
||||
assert cookie_a.split("=", 1)[0] != cookie_b.split("=", 1)[0]
|
||||
|
||||
def _status(url: str, cookie: str) -> dict[str, object]:
|
||||
_, _, body = _get(f"{url}/api/auth/status", cookie=cookie)
|
||||
return dict(json.loads(body))
|
||||
|
||||
# Each server honors its own cookie...
|
||||
assert _status(url_a, cookie_a)["verified"] is True
|
||||
assert _status(url_b, cookie_b)["verified"] is True
|
||||
# ...but treats the other server's cookie as session-less.
|
||||
assert _status(url_a, cookie_b)["verified"] is False
|
||||
assert _status(url_b, cookie_a)["verified"] is False
|
||||
# Even both cookies together (what a real browser would send) only
|
||||
# match the token minted by the receiving server.
|
||||
both = f"{cookie_a}; {cookie_b}"
|
||||
assert _status(url_a, both)["verified"] is True
|
||||
assert _status(url_b, both)["verified"] is True
|
||||
finally:
|
||||
httpd_a.shutdown()
|
||||
httpd_a.server_close()
|
||||
httpd_b.shutdown()
|
||||
httpd_b.server_close()
|
||||
|
||||
|
||||
def test_server_rejects_path_traversal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
run_dir = _make_run(tmp_path, "guard", status="completed", end_time="2026-01-01T00:00:00Z")
|
||||
secret = tmp_path / "secret.txt"
|
||||
|
||||
@@ -0,0 +1,158 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import types
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import pytest
|
||||
from openai.types.responses import ResponseFunctionToolCall
|
||||
|
||||
from strix.core import warmup
|
||||
from strix.core.warmup import (
|
||||
ToolCallingUnsupportedError,
|
||||
probe_tool_calling,
|
||||
requires_tool_call_probe,
|
||||
)
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
|
||||
def _settings(*, api_base: str | None = None, skip: bool = False) -> Any:
|
||||
return types.SimpleNamespace(
|
||||
llm=types.SimpleNamespace(api_base=api_base, skip_tool_call_probe=skip),
|
||||
)
|
||||
|
||||
|
||||
def _tool_call_event() -> Any:
|
||||
return types.SimpleNamespace(
|
||||
item=ResponseFunctionToolCall(
|
||||
arguments='{"status": "ok"}',
|
||||
call_id="call_1",
|
||||
name="strix_ready_check",
|
||||
type="function_call",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _text_event() -> Any:
|
||||
# A completed response whose output is a plain message, no tool call.
|
||||
return types.SimpleNamespace(
|
||||
item=None,
|
||||
response=types.SimpleNamespace(output=[types.SimpleNamespace(type="message")]),
|
||||
)
|
||||
|
||||
|
||||
class _FakeModel:
|
||||
def __init__(self, events: list[Any] | None = None, raises: Exception | None = None) -> None:
|
||||
self._events = events or []
|
||||
self._raises = raises
|
||||
|
||||
def stream_response(self, **_kwargs: Any) -> AsyncIterator[Any]:
|
||||
events = self._events
|
||||
raises = self._raises
|
||||
|
||||
async def _gen() -> AsyncIterator[Any]:
|
||||
if raises is not None:
|
||||
raise raises
|
||||
for event in events:
|
||||
yield event
|
||||
|
||||
return _gen()
|
||||
|
||||
|
||||
def _patch_model(monkeypatch: pytest.MonkeyPatch, model: _FakeModel) -> None:
|
||||
monkeypatch.setattr(
|
||||
warmup, "StrixProvider", lambda: types.SimpleNamespace(get_model=lambda _m: model)
|
||||
)
|
||||
|
||||
|
||||
def test_requires_probe_only_for_custom_endpoints_and_ollama() -> None:
|
||||
assert requires_tool_call_probe("openai/glm-5.2", _settings(api_base="http://x")) is True
|
||||
assert requires_tool_call_probe("ollama/llama3", _settings()) is True
|
||||
assert requires_tool_call_probe("openai/gpt-4o", _settings()) is False
|
||||
assert requires_tool_call_probe("anthropic/claude", _settings()) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_probe_skipped_for_hosted_provider(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
# Would raise if it tried to stream; gating must short-circuit first.
|
||||
_patch_model(monkeypatch, _FakeModel(raises=RuntimeError("should not be called")))
|
||||
await probe_tool_calling("openai/gpt-4o", _settings())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_probe_skipped_when_setting_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_patch_model(monkeypatch, _FakeModel(raises=RuntimeError("should not be called")))
|
||||
await probe_tool_calling("openai/glm-5.2", _settings(api_base="http://x", skip=True))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_probe_passes_on_streamed_tool_call(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_patch_model(monkeypatch, _FakeModel(events=[_tool_call_event()]))
|
||||
await probe_tool_calling("openai/glm-5.2", _settings(api_base="http://x"))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_probe_passes_when_tool_call_only_in_final_response(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
completed = types.SimpleNamespace(
|
||||
item=None,
|
||||
response=types.SimpleNamespace(
|
||||
output=[
|
||||
ResponseFunctionToolCall(
|
||||
arguments="{}",
|
||||
call_id="c",
|
||||
name="strix_ready_check",
|
||||
type="function_call",
|
||||
)
|
||||
]
|
||||
),
|
||||
)
|
||||
_patch_model(monkeypatch, _FakeModel(events=[completed]))
|
||||
await probe_tool_calling("openai/glm-5.2", _settings(api_base="http://x"))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_probe_aborts_when_only_text_streamed(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_patch_model(monkeypatch, _FakeModel(events=[_text_event()]))
|
||||
with pytest.raises(ToolCallingUnsupportedError):
|
||||
await probe_tool_calling("openai/glm-5.2", _settings(api_base="http://x"))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_probe_aborts_on_tool_config_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_patch_model(monkeypatch, _FakeModel(raises=RuntimeError("tools param requires --jinja flag")))
|
||||
with pytest.raises(ToolCallingUnsupportedError):
|
||||
await probe_tool_calling("ollama/llama3", _settings(api_base="http://x"))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_probe_retries_transient_then_passes(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
calls = {"n": 0}
|
||||
good = _FakeModel(events=[_tool_call_event()])
|
||||
|
||||
class _Flaky:
|
||||
def stream_response(self, **kwargs: Any) -> AsyncIterator[Any]:
|
||||
calls["n"] += 1
|
||||
if calls["n"] == 1:
|
||||
|
||||
async def _boom() -> AsyncIterator[Any]:
|
||||
for _ in range(0): # make this a generator without an unreachable yield
|
||||
yield None
|
||||
raise ConnectionError("transient")
|
||||
|
||||
return _boom()
|
||||
return good.stream_response(**kwargs)
|
||||
|
||||
_patch_model(monkeypatch, _Flaky()) # type: ignore[arg-type]
|
||||
await probe_tool_calling("openai/glm-5.2", _settings(api_base="http://x"))
|
||||
assert calls["n"] == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_probe_surfaces_persistent_transient_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_patch_model(monkeypatch, _FakeModel(raises=ConnectionError("down")))
|
||||
with pytest.raises(ConnectionError):
|
||||
await probe_tool_calling("openai/glm-5.2", _settings(api_base="http://x"))
|
||||
Reference in New Issue
Block a user