Compare commits

..
Author SHA1 Message Date
Ahmed Allam 460ceab075 feat(llm): Claude prompt caching (system + tools + rolling tail), Bedrock-safe
Enables Anthropic/Bedrock prompt caching for Claude routes via LiteLLM
cache_control_injection_points: system prompt + latest message everywhere,
plus tool_config on Bedrock Converse only (the sole route whose transform
consumes it; elsewhere it leaks as an unknown top-level field that native
Anthropic 400-rejects). Unmapped Bedrock Claude models run uncached instead
of crashing. Adds STRIX_PROMPT_CACHE opt-out (default on).
2026-07-26 23:29:41 +00:00
Devin AI fb6606279f Merge branch 'pr-772' into devin/1785108025-claude-prompt-caching 2026-07-26 23:20:26 +00:00
Sean TurnerandClaude Opus 4.8 efb698a4d0 feat(llm): also cache the append-only conversation tail
The two prefix breakpoints (system + tool_config) only cache the FIXED
prefix. A Strix scan's transcript is append-only, so the growing
conversation body is re-sent at full input price every turn and
cache-read decays as the transcript grows — a denominator effect, not
the prefix missing.

Add a third rolling breakpoint at index:-1 (the last message). Because
prior turns are immutable, this re-caches the whole prefix-so-far each
turn and hits on the next; LiteLLM resolves the negative index against
the live message list. Measured on a 29-turn Bedrock scan the fixed
prefix stayed pinned at ~56k tokens while per-turn input grew to ~256k
and cache-read fell 90% -> 22%; the tail point lifts modelled cache-read
to ~96% and cuts full-price input ~16x. Degrades gracefully on older
LiteLLM (unrecognised location simply not injected).

Adds an end-to-end test driving LiteLLM 1.90.1's _apply_message_injections
to confirm the breakpoint tracks the tail across a growing transcript.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-17 13:33:30 +01:00
seanturner83andClaude Opus 4.8 cc2b3351b8 fix(llm): don't inject prompt-cache marker for unmapped Bedrock Claude
The cache breakpoints are gated on _is_claude_model (name contains
"claude"), but LiteLLM's AnthropicCacheControlHook only *consumes*
cache_control_injection_points for models it recognises as cache-capable
via its statically bundled model map. On a Bedrock route whose model
isn't in that map, the marker passes straight through and Bedrock's
Converse API rejects it outright:

  ValidationException: cache_control_injection_points: Extra inputs are
  not permitted

— which fails the whole scan at the first LLM call. This bites any
Bedrock Claude model LiteLLM hasn't mapped yet (a just-released model),
and is made worse when LiteLLM can't refresh its remote model map (e.g.
behind a TLS-intercepting corporate proxy) and falls back to a stale
local copy. Observed live on bedrock/global.anthropic.claude-sonnet-5.

Fix: withhold the marker only for a Bedrock route LiteLLM can't confirm
supports prompt caching. Scope is deliberately narrow — Anthropic-native,
Vertex, and OpenRouter Claude tolerate/ignore the marker (or LiteLLM maps
them under keys we don't resolve), so gating those on confirmed support
would DISABLE caching for capable models — the opposite of this PR's
intent. Only Bedrock hard-rejects, so only Bedrock is guarded.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-16 14:22:25 +01:00
Sean Turner af7769507a review: merge cache extra_args explicitly + note graceful degradation
Address Greptile feedback on #772:
- Build extra_args as {**existing, **cache} at the call site rather than
  leaning on ModelSettings.resolve()'s dict-merge — makes preservation of
  unrelated LiteLLM options obvious to a reader (resolve() does merge, but
  it's non-obvious). No behaviour change: make_model_settings builds from
  scratch so the base extra_args is None today.
- Document that an unrecognised injection-point location degrades gracefully
  (not injected, no error) on older LiteLLM pins; tool_config is honoured by
  the Bedrock Converse transform on versions that support it (litellm 1.90.1
  verified).
2026-07-15 13:07:19 +01:00
Sean Turner d93a99a355 feat(llm): enable Bedrock/Anthropic prompt caching for Claude models
A Strix scan is a long multi-turn agentic loop that re-sends a large, STABLE
prefix every turn — the system prompt plus the tool schemas — while only the
conversation tail changes. Without a caching breakpoint that whole prefix is
re-tokenised and billed at full input rate on every turn; on Bedrock Claude
it's the single biggest lever on scan cost. Measured on a real scan: cache-read
went 0% -> 57% once these injection points are set (roughly halving input cost,
and the ratio climbs on longer scans where the stable prefix dominates more
turns).

LiteLLM already implements this end to end: when `cache_control_injection_points`
is present in the call kwargs its `AnthropicCacheControlHook` fires and emits the
provider-appropriate breakpoint (Anthropic `cache_control`; Bedrock Converse
`cachePoint`), honouring Anthropic's 4-breakpoint cap. `LitellmModel` forwards
`ModelSettings.extra_args` straight into `litellm.acompletion()`, so passing the
points there is all that's needed. We mark the two big stable segments (system
prompt + tool_config = 2 of 4 breakpoints, headroom left).

Deliberately kept at the LiteLLM-config layer rather than a general ModelSettings
caching flag — that's the direction the Agents SDK maintainer prescribed when
declining a native `cache_system_prompt` field
(openai/openai-agents-python#3008 / #3009): caching is a LiteLLM/provider
behaviour, and a ModelSettings flag would let strict OpenAI-compatible paths emit
non-standard cache_control parts. Gating on Claude keeps it a strict no-op for
every other provider (no injection points -> the hook never fires); only
Claude-family routes (Anthropic native, Bedrock, Vertex, OpenRouter -> Claude)
honour the marker.

Tests: parametrised, non-vacuous — Claude routes (bedrock/native/openrouter) get
the two injection points; non-Claude (gpt-5/gemini/o3) get extra_args=None.
2026-07-15 12:54:20 +01:00
37 changed files with 296 additions and 2769 deletions
-16
View File
@@ -21,8 +21,6 @@ jobs:
target: macos-x86_64
- os: ubuntu-22.04
target: linux-x86_64
- os: ubuntu-22.04-arm
target: linux-arm64
- os: windows-latest
target: windows-x86_64
@@ -45,20 +43,6 @@ jobs:
uv sync --frozen
uv run pyinstaller strix.spec --noconfirm
if [[ "${{ runner.os }}" == "Windows" ]]; then
dist/strix.exe --version
else
dist/strix --version
fi
if [[ "${{ matrix.target }}" == "linux-arm64" ]]; then
file dist/strix
file dist/strix | grep -q "ARM aarch64" || {
echo "::error::linux-arm64 artifact is not an ARM aarch64 binary"
exit 1
}
fi
VERSION=$(grep '^version' pyproject.toml | head -1 | sed 's/.*"\(.*\)"/\1/')
mkdir -p dist/release
+3 -36
View File
@@ -61,28 +61,11 @@ strix (--target <target> | --target-list <path> | --mount <path>) [options]
Path to a custom config file (JSON) to use instead of `~/.strix/cli-config.json`.
</ParamField>
<ParamField path="--max-budget" type="number">
<ParamField path="--max-budget-usd" type="number">
Maximum LLM spend in USD for the whole scan, counted cumulatively across the
root agent and every child agent. The budget is checked after each model
response.
In non-interactive mode (`-n`), once the running cost reaches the threshold,
the scan stops cleanly with a `stopped` status (not a failure) and the sandbox
is torn down. Sub-agents are stopped early, at 90% of the budget, reserving
the final slice for the root agent to wind down and produce the final report.
In interactive mode, reaching the budget pauses the scan instead of ending
it: every agent parks, and sending any message resumes the scan with the cap
extended by the original budget amount. There is no sub-agent reserve in
interactive mode.
As the budget is approached, graduated wrap-up warnings are surfaced to
**every** agent so they can finish their work and call their lifecycle tool
before the hard stop. The bands sit just below each role's own stop point: the
root is warned at **70%, 85% and 95%** (it stops at 100%), while sub-agents are
warned at **75%, 80% and 85%** (they stop at the 90% reserve). In interactive
mode every agent uses the **70%, 85% and 95%** bands. Percentages shown in the
warnings are the real cumulative spend against the full budget.
response; once the running cost reaches the threshold, the scan stops cleanly
with a `stopped` status (not a failure) and the sandbox is torn down.
Must be greater than `0`. Omit the flag for no limit.
@@ -101,19 +84,6 @@ strix (--target <target> | --target-list <path> | --mount <path>) [options]
counts.
</ParamField>
<ParamField path="--max-turns" type="integer" default="500">
Maximum number of turns (one model response plus its tool round) allotted to
**each** agent, applied per run. When an agent reaches this limit it is
force-stopped.
As the limit is approached, graduated wrap-up warnings (at 70%, 85% and 95%)
are injected into that agent's next model turn so it can prioritise its
remaining work and call its lifecycle tool (`finish_scan` for the root agent,
`agent_finish` for sub-agents) before the hard stop.
Must be greater than `0`.
</ParamField>
## Examples
```bash
@@ -129,9 +99,6 @@ strix --target api.example.com --instruction "Focus on IDOR and auth bypass"
# CI/CD mode
strix -n --target ./ --scan-mode quick
# Cap cost and per-agent turns
strix --target https://example.com --max-budget 25 --max-turns 300
# Force diff-scope against a specific base ref
strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "strix-agent"
version = "1.4.1"
version = "1.3.1"
description = "Open-source AI Hackers for your apps"
readme = "README.md"
license = "Apache-2.0"
+1 -1
View File
@@ -41,7 +41,7 @@ fi
combo="$os-$arch"
case "$combo" in
linux-x86_64|linux-arm64|macos-x86_64|macos-arm64|windows-x86_64)
linux-x86_64|macos-x86_64|macos-arm64|windows-x86_64)
;;
*)
echo -e "${RED}Unsupported OS/Arch: $os/$arch${NC}"
+20 -13
View File
@@ -18,12 +18,12 @@ import logging
import secrets
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import TYPE_CHECKING, Any
import requests
if TYPE_CHECKING:
from collections.abc import Iterator
@@ -221,19 +221,26 @@ def _first(query: dict[str, list[str]], key: str) -> str | None:
def _post_form(payload: dict[str, str]) -> dict[str, Any]:
body = urllib.parse.urlencode(payload).encode("ascii")
request = urllib.request.Request( # noqa: S310 - fixed https OAuth endpoint
TOKEN_URL,
data=body,
headers={
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json",
},
method="POST",
)
try:
response = requests.post(
TOKEN_URL,
data=payload,
headers={"Accept": "application/json"},
timeout=_TOKEN_TIMEOUT,
)
except requests.RequestException as exc:
with urllib.request.urlopen( # noqa: S310 # nosec B310 - fixed https endpoint
request, timeout=_TOKEN_TIMEOUT
) as response:
data = json.loads(response.read() or b"{}")
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", "replace")[:300]
raise CodexAuthError("token_http_error", f"HTTP {exc.code}: {detail}") from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
raise CodexAuthError("unavailable", str(exc)) from exc
if response.status_code >= 400:
detail = response.text[:300]
raise CodexAuthError("token_http_error", f"HTTP {response.status_code}: {detail}")
data = json.loads(response.content or b"{}")
if not isinstance(data, dict):
raise CodexAuthError("bad_response", "token endpoint returned non-object")
return data
+26 -49
View File
@@ -277,51 +277,6 @@ 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 = {
@@ -481,13 +436,24 @@ def is_claude_model(model_name: str) -> bool:
def is_bedrock_route(model_name: str) -> bool:
"""Whether ``model_name`` resolves to an AWS Bedrock route.
Matches the ``bedrock/...`` LiteLLM route prefix and bare Bedrock model ids
(``[region.]anthropic.claude-...``).
"""
name = (model_name or "").strip().lower()
return name.startswith("bedrock/") or "anthropic." in name
def _prompt_cache_name_candidates(model_name: str) -> list[str]:
# LiteLLM's model map keys the same model under several names; strip the
# route prefix, then leading dotted segments (region, provider).
"""Candidate LiteLLM model-map keys for ``model_name``, most→least specific.
LiteLLM keys the same model under several names (``bedrock/global.anthropic.
claude-opus-4-1``, ``anthropic.claude-opus-4-1``, ``claude-opus-4-1``) and not
every provider/region-prefixed variant is present for every model. Strip the
LiteLLM route prefix, then leading dotted segments (region, then provider) so
a prefixed name still resolves to a bare key.
"""
name = (model_name or "").strip().lower()
for prefix in ("litellm/", "bedrock/"):
if name.startswith(prefix):
@@ -502,13 +468,24 @@ def _prompt_cache_name_candidates(model_name: str) -> list[str]:
def bedrock_route_supports_prompt_caching(model_name: str) -> bool:
# Bedrock rejects the cache marker for models LiteLLM's map doesn't
# recognise as cache-capable, so callers withhold it unless confirmed here.
"""Whether LiteLLM can confirm this Bedrock model supports prompt caching.
Bedrock's Converse API rejects unknown request fields outright
(``ValidationException: cache_control_injection_points: Extra inputs are
not permitted``), and LiteLLM only consumes the cache marker for models its
(statically bundled) model map recognises as cache-capable. For a Bedrock
model missing from that map — a just-released model, or any model when the
remote model-map refresh fails and a stale local copy is used — the marker
would pass straight through and fail every call, so callers must withhold
it unless support is confirmed here.
"""
import litellm
checker = getattr(getattr(litellm, "utils", None), "supports_prompt_caching", None)
for cand in _prompt_cache_name_candidates(model_name):
if checker is not None:
# supports_prompt_caching raises for models missing from the map;
# keep checking the remaining name candidates.
with contextlib.suppress(Exception):
if checker(cand):
return True
+2 -5
View File
@@ -36,11 +36,8 @@ class LlmSettings(BaseSettings):
),
)
reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
# None = auto: force required tool choice on OpenAI-compatible custom
# endpoints (where reasoning models otherwise burn tokens thinking without
# ever calling a tool), off elsewhere. True/False overrides explicitly.
force_required_tool_choice: bool | None = Field(
default=None,
force_required_tool_choice: bool = Field(
default=False,
alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
)
prompt_cache: bool = Field(
+5 -86
View File
@@ -14,15 +14,13 @@ from strix.core.sessions import session_write_lock
if TYPE_CHECKING:
from collections.abc import Callable
from agents.items import TResponseInputItem
from agents.memory import Session
logger = logging.getLogger(__name__)
Status = Literal["running", "waiting", "completed", "stopped", "crashed", "failed", "budget_paused"]
Status = Literal["running", "waiting", "completed", "stopped", "crashed", "failed"]
@dataclass(slots=True)
@@ -49,9 +47,6 @@ class AgentCoordinator:
self._snapshot_path: Path | None = None
self.is_shutting_down = False
self._budget_stopped = False
self._reserve_stopped = False
self._budget_paused = False
self._extend_budget: Callable[[], None] | None = None
def set_snapshot_path(self, path: Path) -> None:
self._snapshot_path = path
@@ -70,71 +65,6 @@ class AgentCoordinator:
for runtime in self.runtimes.values():
runtime.wake.set()
@property
def reserve_stopped(self) -> bool:
return self._reserve_stopped
@property
def budget_paused(self) -> bool:
return self._budget_paused
def set_budget_extender(self, extend: Callable[[], None]) -> None:
self._extend_budget = extend
async def pause_for_budget(self, agent_id: str) -> None:
async with self._lock:
self._budget_paused = True
await self.set_status(agent_id, "budget_paused")
async def resume_from_budget_pause(self, *, exclude: str | None = None) -> None:
async with self._lock:
if not self._budget_paused:
return
self._budget_paused = False
paused = [aid for aid, status in self.statuses.items() if status == "budget_paused"]
if self._extend_budget is not None:
self._extend_budget()
for aid in paused:
await self.set_status(aid, "waiting")
if aid != exclude:
await self.send(
aid,
{
"from": "system",
"type": "budget_extended",
"content": (
"[Budget] The user extended the scan budget \u2014 continue your "
"current task."
),
},
)
async def reset_budget_stops(
self,
*,
budget_stopped: bool,
reserve_stopped: bool,
budget_paused: bool = False,
) -> None:
async with self._lock:
self._budget_stopped = budget_stopped
self._reserve_stopped = reserve_stopped
if not budget_paused:
self._budget_paused = False
for aid, status in self.statuses.items():
if status == "budget_paused":
self.statuses[aid] = "waiting"
await self._maybe_snapshot()
async def claim_reserve_notification(self) -> str | None:
async with self._lock:
if self._reserve_stopped:
return None
self._reserve_stopped = True
for runtime in self.runtimes.values():
runtime.wake.set()
return next((aid for aid, parent in self.parent_of.items() if parent is None), None)
async def register(
self,
agent_id: str,
@@ -200,12 +130,8 @@ 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], *, interrupt: bool = True
) -> bool:
async def send(self, target_agent_id: str, message: dict[str, Any]) -> 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)
async with self._lock:
if target_agent_id not in self.statuses:
logger.debug("agent.send dropped unknown target=%s", target_agent_id)
@@ -213,7 +139,7 @@ class AgentCoordinator:
runtime = self.runtimes.setdefault(target_agent_id, AgentRuntime())
session = runtime.session
stream = runtime.stream
interrupt_on_message = runtime.interrupt_on_message
interrupt = runtime.interrupt_on_message
if session is None:
logger.warning(
"agent.send dropped target=%s because its SDK session is not attached",
@@ -232,7 +158,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 and interrupt_on_message:
if stream is not None and interrupt:
stream.cancel(mode="immediate")
await self._maybe_snapshot()
return True
@@ -240,8 +166,7 @@ class AgentCoordinator:
async def wait_for_message(self, agent_id: str) -> None:
while True:
async with self._lock:
reserve_exit = self._reserve_stopped and self.parent_of.get(agent_id) is not None
if self._budget_stopped or reserve_exit or self.pending_counts.get(agent_id, 0) > 0:
if self._budget_stopped or self.pending_counts.get(agent_id, 0) > 0:
return
wake = self.runtimes.setdefault(agent_id, AgentRuntime()).wake
wake.clear()
@@ -375,9 +300,6 @@ class AgentCoordinator:
"metadata": {aid: dict(md) for aid, md in self.metadata.items()},
"pending_counts": dict(self.pending_counts),
"errors": dict(self.errors),
"budget_stopped": self._budget_stopped,
"reserve_stopped": self._reserve_stopped,
"budget_paused": self._budget_paused,
}
async def restore(self, snap: dict[str, Any]) -> None:
@@ -388,9 +310,6 @@ class AgentCoordinator:
self.metadata = {aid: dict(md) for aid, md in snap.get("metadata", {}).items()}
self.pending_counts = dict(snap.get("pending_counts", {}))
self.errors = dict(snap.get("errors", {}))
self._budget_stopped = bool(snap.get("budget_stopped", False))
self._reserve_stopped = bool(snap.get("reserve_stopped", False))
self._budget_paused = bool(snap.get("budget_paused", False))
for aid in self.statuses:
self.runtimes.setdefault(aid, AgentRuntime())
+41 -194
View File
@@ -13,20 +13,9 @@ from agents import RunConfig, Runner
from agents.exceptions import AgentsException, MaxTurnsExceeded, UserError
from agents.sandbox.errors import ExecTransportError
from docker import errors as docker_errors # type: ignore[import-untyped, unused-ignore]
from openai import (
APIConnectionError,
APIError,
APIStatusError,
APITimeoutError,
RateLimitError,
)
from openai import APIError
from strix.config import codex
from strix.core.hooks import (
BudgetExceededError,
BudgetPausedError,
SubagentBudgetReservedError,
)
from strix.core.hooks import BudgetExceededError
from strix.core.inputs import child_initial_input
from strix.core.sessions import (
enforce_image_budget,
@@ -89,39 +78,6 @@ 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
_TRANSIENT_MODEL_RETRY_MAX_DELAY_S = 30.0
def _model_error_status_code(exc: BaseException) -> int | None:
code = getattr(exc, "status_code", None)
return code if isinstance(code, int) else None
def _is_transient_model_error(exc: BaseException) -> bool:
if isinstance(exc, RateLimitError):
return False
if isinstance(exc, APITimeoutError | APIConnectionError):
return True
if isinstance(exc, APIStatusError):
return exc.status_code in _TRANSIENT_MODEL_STATUS_CODES
if isinstance(exc, APIError):
return _model_error_status_code(exc) is None
return False
def _transient_model_retry_delay(attempt: int) -> float:
delay = _TRANSIENT_MODEL_RETRY_BASE_DELAY_S * float(2 ** (attempt - 1))
return min(delay, _TRANSIENT_MODEL_RETRY_MAX_DELAY_S)
async def run_agent_loop(
*,
agent: Any,
@@ -144,34 +100,21 @@ async def run_agent_loop(
)
result: RunResultBase | None = None
budget_stopped = coordinator.budget_stopped
reserve_stopped = coordinator.reserve_stopped
if budget_stopped:
await coordinator.set_status(agent_id, "stopped")
raise BudgetExceededError("scan budget reached")
if reserve_stopped and context.get("parent_id") is not None:
await coordinator.set_status(agent_id, "stopped")
raise SubagentBudgetReservedError("scan reached the sub-agent budget reserve")
if reserve_stopped and start_parked and interactive and context.get("parent_id") is None:
await coordinator.send(agent_id, _reserve_notice())
if not (start_parked and interactive):
if interactive:
with contextlib.suppress(BudgetPausedError):
result = await _run_cycle(
agent,
coordinator,
agent_id,
input_data=initial_input,
run_config=run_config,
context=context,
max_turns=max_turns,
session=session,
interactive=interactive,
event_sink=event_sink,
hooks=hooks,
)
result = await _run_cycle(
agent,
coordinator,
agent_id,
input_data=initial_input,
run_config=run_config,
context=context,
max_turns=max_turns,
session=session,
interactive=interactive,
event_sink=event_sink,
hooks=hooks,
)
else:
result = await _run_noninteractive_until_lifecycle(
agent,
@@ -199,25 +142,20 @@ async def run_agent_loop(
await coordinator.set_status(agent_id, "stopped")
raise BudgetExceededError("scan budget reached")
if coordinator.reserve_stopped and context.get("parent_id") is not None:
await coordinator.set_status(agent_id, "stopped")
raise SubagentBudgetReservedError("scan reached the sub-agent budget reserve")
await coordinator.consume_pending(agent_id)
with contextlib.suppress(BudgetPausedError):
result = await _run_cycle(
agent,
coordinator,
agent_id,
input_data=[],
run_config=run_config,
context=context,
max_turns=max_turns,
session=session,
interactive=interactive,
event_sink=event_sink,
hooks=hooks,
)
result = await _run_cycle(
agent,
coordinator,
agent_id,
input_data=[],
run_config=run_config,
context=context,
max_turns=max_turns,
session=session,
interactive=interactive,
event_sink=event_sink,
hooks=hooks,
)
async def spawn_child_agent(
@@ -310,7 +248,6 @@ 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,
@@ -323,8 +260,7 @@ async def respawn_subagents(
for child_id, name, parent_id, md in candidates:
try:
restored_status = str(md.get("_restored_status") or "running")
recoverable_park = restored_status == "waiting" and bool(md.get("_restored_error"))
start_parked = interactive and restored_status != "running" and not recoverable_park
start_parked = interactive and restored_status != "running"
if start_parked:
logger.warning(
@@ -391,10 +327,6 @@ async def _run_noninteractive_until_lifecycle(
await coordinator.set_status(agent_id, "stopped")
raise BudgetExceededError("scan budget reached")
if coordinator.reserve_stopped and context.get("parent_id") is not None:
await coordinator.set_status(agent_id, "stopped")
raise SubagentBudgetReservedError("scan reached the sub-agent budget reserve")
result = await _run_cycle(
agent,
coordinator,
@@ -425,7 +357,7 @@ async def _run_noninteractive_until_lifecycle(
if invalid_final_outputs >= invalid_final_output_limit:
await coordinator.set_status(agent_id, "crashed")
await _notify_parent_on_terminal(coordinator, agent_id, "crashed")
await _notify_parent_on_crash(coordinator, agent_id, "crashed")
raise MaxTurnsExceeded(
"Agent exhausted non-interactive recovery attempts without calling "
"finish_scan or agent_finish."
@@ -455,7 +387,6 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
) -> RunResultBase | None:
image_strips = 0
compactions = 0
model_retries = 0
while True:
try:
await coordinator.mark_running(agent_id)
@@ -490,7 +421,9 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
logger.exception("stream event sink failed for %s", agent_id)
if stream.run_loop_exception is not None:
raise stream.run_loop_exception
except (BudgetExceededError, BudgetPausedError, SubagentBudgetReservedError):
except BudgetExceededError:
# A RuntimeError subclass: re-raise explicitly so it is never
# mistaken for the LiteLLM "after shutdown" race below.
raise
except RuntimeError as stream_exc:
if "after shutdown" not in str(stream_exc):
@@ -509,15 +442,6 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
)
finally:
await coordinator.detach_stream(agent_id, stream)
except BudgetPausedError as exc:
logger.info("agent %s paused at the scan budget limit: %s", agent_id, exc)
await coordinator.pause_for_budget(agent_id)
raise
except SubagentBudgetReservedError as exc:
logger.info("sub-agent %s stopped at the budget reserve: %s", agent_id, exc)
await coordinator.set_status(agent_id, "stopped")
await _notify_root_on_budget_reserve(coordinator)
raise
except BudgetExceededError as exc:
logger.info(
"agent %s reached the scan budget limit; stopping the scan: %s", agent_id, exc
@@ -564,26 +488,6 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
)
input_data = []
continue
if model_retries < _MAX_TRANSIENT_MODEL_RETRIES and _is_transient_model_error(exc):
model_retries += 1
delay = _transient_model_retry_delay(model_retries)
logger.warning(
"transient model/provider error for %s; replaying turn "
"(attempt %d/%d, backoff %.1fs): %r",
agent_id,
model_retries,
_MAX_TRANSIENT_MODEL_RETRIES,
delay,
exc,
)
await asyncio.sleep(delay)
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):
@@ -594,29 +498,13 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
status = "crashed"
logger.exception("agent run failed for %s; parking as %s", agent_id, status)
await coordinator.set_status(agent_id, status, error=str(exc) or type(exc).__name__)
await _notify_parent_on_terminal(coordinator, agent_id, status)
await _notify_parent_on_crash(coordinator, agent_id, status)
return None
else:
await _settle_run_result(coordinator, agent_id, interactive)
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,
@@ -674,31 +562,12 @@ async def _append_noninteractive_tool_required_message(
return []
_TERMINAL_NOTICE = {
"crashed": (
"[Agent crash] {name} ({agent_id}) terminated unexpectedly. "
"Stop waiting on this child unless you want to message it again."
),
"failed": (
"[Agent failed] {name} ({agent_id}) stopped with an error and will not "
"send a completion report. Stop waiting on this child unless you want to "
"message it again."
),
"stopped": (
"[Agent capped] {name} ({agent_id}) hit its turn limit and was stopped "
"before finishing. It will not send a completion report, so stop waiting "
"on this child; account for its capped subtask and continue."
),
}
async def _notify_parent_on_terminal(
async def _notify_parent_on_crash(
coordinator: AgentCoordinator,
agent_id: str,
status: str,
) -> None:
template = _TERMINAL_NOTICE.get(status)
if template is None:
if status != "crashed":
return
async with coordinator._lock:
parent = coordinator.parent_of.get(agent_id)
@@ -709,36 +578,16 @@ async def _notify_parent_on_terminal(
parent,
{
"from": agent_id,
"type": status,
"type": "crash",
"priority": "high",
"content": template.format(name=name, agent_id=agent_id),
"content": (
f"[Agent crash] {name} ({agent_id}) terminated unexpectedly. "
"Stop waiting on this child unless you want to message it again."
),
},
interrupt=False,
)
def _reserve_notice() -> dict[str, Any]:
return {
"from": "system",
"type": "budget_reserve_stop",
"priority": "high",
"content": (
"[Budget reserve] The scan has reached the sub-agent budget reserve: every "
"sub-agent is being force-stopped as soon as its in-flight turn completes, and "
"none will send a completion report. Their confirmed vulnerabilities are "
"already filed as they were found. Do not wait on any sub-agents and do not "
"spawn new ones — wrap up now and call finish_scan."
),
}
async def _notify_root_on_budget_reserve(coordinator: AgentCoordinator) -> None:
root = await coordinator.claim_reserve_notification()
if root is None:
return
await coordinator.send(root, _reserve_notice())
async def _start_child_runner(
*,
parent_ctx: dict[str, Any],
@@ -790,8 +639,6 @@ async def _start_child_runner(
)
except BudgetExceededError:
logger.info("child %s stopped after reaching the scan budget limit", child_id)
except SubagentBudgetReservedError:
logger.info("child %s stopped at the sub-agent budget reserve", child_id)
task_handle = asyncio.create_task(_child_loop(), name=f"agent-{name}-{child_id}")
await coordinator.attach_runtime(child_id, task=task_handle)
+3 -202
View File
@@ -14,210 +14,26 @@ from strix.report.state import get_global_report_state
if TYPE_CHECKING:
from agents import RunContextWrapper
from agents.agent import Agent
from agents.items import ModelResponse, TResponseInputItem
from agents.items import ModelResponse
logger = logging.getLogger(__name__)
_STAGE_LABELS: tuple[str, ...] = ("NOTICE", "URGENT", "CRITICAL")
_TURN_WARN_BANDS: tuple[float, ...] = (0.70, 0.85, 0.95)
_ROOT_BUDGET_WARN_BANDS: tuple[float, ...] = (0.70, 0.85, 0.95)
_SUBAGENT_BUDGET_WARN_BANDS: tuple[float, ...] = (0.75, 0.80, 0.85)
_SUBAGENT_BUDGET_RESERVE = 0.90
class BudgetExceededError(RuntimeError):
"""Raised when the accumulated LLM cost reaches the configured budget."""
class SubagentBudgetReservedError(RuntimeError):
"""Raised to stop a single sub-agent once the reserve threshold is crossed."""
class BudgetPausedError(RuntimeError):
"""Raised to park one agent when an interactive scan reaches its budget."""
def recomputed_budget_flags(
cost: float,
max_budget_usd: float | None,
*,
interactive: bool,
) -> tuple[bool, bool]:
"""Return the (budget_stopped, reserve_stopped) flags a resumed scan should carry."""
if max_budget_usd is None:
return False, False
if interactive:
return False, False
budget_stopped = cost >= max_budget_usd
reserve_stopped = cost >= max_budget_usd * _SUBAGENT_BUDGET_RESERVE
return budget_stopped, reserve_stopped
def _crossed_stage(fraction: float, bands: tuple[float, ...]) -> int | None:
crossed: int | None = None
for index, band in enumerate(bands):
if fraction >= band:
crossed = index
return crossed
_ROOT_DIRECTIVES: tuple[str, ...] = (
(
"As the root agent, begin planning your wind-down of the whole scan: avoid "
"starting large new lines of investigation, and keep your required objectives on "
"track so you can call finish_scan comfortably before the limit."
),
(
"As the root agent, prioritize wrapping up the whole scan now: stop opening new "
"lines of investigation, close out only what is essential, and move toward calling "
"finish_scan to compile and deliver the final report."
),
(
"As the root agent, STOP all other work on the whole scan and finish immediately: "
"secure your findings and call finish_scan now — anything left unfinished when the "
"limit is hit is discarded."
),
)
_SUBAGENT_DIRECTIVES: tuple[str, ...] = (
(
"As a sub-agent, begin planning your wind-down: avoid starting large new subtasks, "
"and if you are close to a confirmed, validated vulnerability, drive it to a result "
"you can report."
),
(
"As a sub-agent, prioritize wrapping up your task now: report any confirmed, "
"validated vulnerability, finish work that is nearly done rather than starting "
"anything new, and prepare to call agent_finish."
),
(
"As a sub-agent, STOP all other work and finish immediately: report any confirmed "
"vulnerability right now and call agent_finish to hand your results back to your "
"parent before you are cut off."
),
)
def _wrapup_directive(context: RunContextWrapper[dict[str, Any]], stage: int) -> str:
is_root = context.context.get("parent_id") is None
directives = _ROOT_DIRECTIVES if is_root else _SUBAGENT_DIRECTIVES
return directives[stage]
def _urgency(stage: int) -> str:
return _STAGE_LABELS[stage]
class ReportUsageHooks(RunHooks[dict[str, Any]]):
"""Persist SDK-native usage and warn/stop as turn and cost budgets are consumed."""
"""Persist SDK-native usage after every model response."""
def __init__(
self,
*,
model: str,
max_budget_usd: float | None = None,
max_turns: int | None = None,
interactive: bool = False,
) -> None:
def __init__(self, *, model: str, max_budget_usd: float | None = None) -> None:
if max_budget_usd is not None and (
not math.isfinite(max_budget_usd) or max_budget_usd <= 0
):
raise ValueError("max_budget_usd must be a finite number greater than 0")
if max_turns is not None and max_turns <= 0:
raise ValueError("max_turns must be a positive integer")
self._model = model
self._max_budget_usd = max_budget_usd
self._budget_increment = max_budget_usd
self._max_turns = max_turns
self._interactive = interactive
def extend_budget(self) -> None:
if self._max_budget_usd is None or self._budget_increment is None:
return
self._max_budget_usd += self._budget_increment
async def on_llm_start(
self,
context: RunContextWrapper[dict[str, Any]],
agent: Agent[dict[str, Any]], # noqa: ARG002
system_prompt: str | None, # noqa: ARG002
input_items: list[TResponseInputItem],
) -> None:
try:
self._maybe_warn_turns(context, input_items)
self._maybe_warn_budget(context, input_items)
except Exception:
logger.exception("budget/turn warning injection failed")
def _maybe_warn_turns(
self,
context: RunContextWrapper[dict[str, Any]],
input_items: list[TResponseInputItem],
) -> None:
if not self._max_turns:
return
usage = getattr(context, "usage", None)
requests = getattr(usage, "requests", None)
if not isinstance(requests, int):
return
turns_used = requests + 1
stage = _crossed_stage(turns_used / self._max_turns, _TURN_WARN_BANDS)
if stage is None:
return
remaining = max(self._max_turns - turns_used, 0)
pct = round(100 * turns_used / self._max_turns)
content = (
f"[{_urgency(stage)}] Turn budget: {turns_used}/{self._max_turns} used ({pct}%). "
f"About {remaining} turn(s) remain before this agent is force-stopped and any "
f"in-progress work is discarded. {_wrapup_directive(context, stage)}"
)
input_items.append({"role": "user", "content": content})
def _maybe_warn_budget(
self,
context: RunContextWrapper[dict[str, Any]],
input_items: list[TResponseInputItem],
) -> None:
if self._max_budget_usd is None:
return
report_state = get_global_report_state()
if report_state is None:
return
cost = report_state.get_total_llm_cost()
is_root = context.context.get("parent_id") is None
if self._interactive:
bands = _ROOT_BUDGET_WARN_BANDS
else:
bands = _ROOT_BUDGET_WARN_BANDS if is_root else _SUBAGENT_BUDGET_WARN_BANDS
stage = _crossed_stage(cost / self._max_budget_usd, bands)
if stage is None:
return
pct = round(100 * cost / self._max_budget_usd)
reserve_pct = round(_SUBAGENT_BUDGET_RESERVE * 100)
if self._interactive:
content = (
f"[{_urgency(stage)}] Scan cost budget: ${cost:.2f}/${self._max_budget_usd:.2f} "
f"spent ({pct}%). This budget is shared across every agent in the scan; when it "
"is reached all agents are paused until the user chooses to continue. "
f"{_wrapup_directive(context, stage)}"
)
elif is_root:
content = (
f"[{_urgency(stage)}] Scan cost budget: ${cost:.2f}/${self._max_budget_usd:.2f} "
f"spent ({pct}%). This budget is shared across every agent in the scan; when it "
"is reached the whole scan is stopped immediately, and sub-agents are stopped at "
f"{reserve_pct}% to reserve the remainder for your final report. "
f"{_wrapup_directive(context, stage)}"
)
else:
content = (
f"[{_urgency(stage)}] Scan cost budget: ${cost:.2f}/${self._max_budget_usd:.2f} "
f"spent ({pct}%). This budget is shared across every agent in the scan; "
f"sub-agents are stopped at {reserve_pct}% to leave the remainder for the root "
f"agent's final report. {_wrapup_directive(context, stage)}"
)
input_items.append({"role": "user", "content": content})
async def on_llm_end(
self,
@@ -250,21 +66,6 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
if self._max_budget_usd is not None:
cost = report_state.get_total_llm_cost()
if cost >= self._max_budget_usd:
if self._interactive:
raise BudgetPausedError(
f"Scan budget of ${self._max_budget_usd:.2f} reached "
f"(spent ${cost:.4f}); pausing until the user continues"
)
raise BudgetExceededError(
f"Token budget of ${self._max_budget_usd:.2f} exceeded (spent ${cost:.4f})"
)
is_root = ctx.get("parent_id") is None
if not self._interactive and not is_root:
reserve_limit = self._max_budget_usd * _SUBAGENT_BUDGET_RESERVE
if cost >= reserve_limit:
raise SubagentBudgetReservedError(
f"Sub-agent budget reserve reached: spent ${cost:.4f} of "
f"${self._max_budget_usd:.2f} "
f"(>= {round(_SUBAGENT_BUDGET_RESERVE * 100)}% reserve); stopping this "
"sub-agent so the root agent can finish the scan."
)
+66 -17
View File
@@ -129,8 +129,7 @@ def make_model_settings(
reasoning_effort: ReasoningEffort | None,
*,
model_name: str,
force_required_tool_choice: bool | None = None,
custom_api_base: bool = False,
force_required_tool_choice: bool = False,
request_timeout: float | None = None,
prompt_cache: bool = True,
) -> ModelSettings:
@@ -148,18 +147,14 @@ def make_model_settings(
model_settings = model_settings.resolve(
ModelSettings(reasoning=Reasoning(effort=reasoning_effort)),
)
# Strix is fully tool-driven, so a turn that returns prose instead of a tool
# call stalls the scan. Reasoning models on OpenAI-compatible custom
# endpoints (e.g. GLM / Kimi via cortecs) do exactly that under the default
# ``auto`` tool choice, so default to ``required`` there; ``None`` means auto.
use_required = (
custom_api_base if force_required_tool_choice is None else force_required_tool_choice
)
if use_required and _accepts_required_tool_choice(model_name):
if force_required_tool_choice and _accepts_required_tool_choice(model_name):
model_settings = model_settings.resolve(ModelSettings(tool_choice="required"))
cache_extra_args = _prompt_cache_extra_args(model_name) if prompt_cache else None
if cache_extra_args:
# Merge into any existing extra_args (e.g. the request timeout) rather
# than relying on resolve()'s dict-merge semantics, so it is obvious at
# the call site that unrelated LiteLLM options are preserved.
model_settings = model_settings.resolve(
ModelSettings(
extra_args={**(model_settings.extra_args or {}), **cache_extra_args},
@@ -169,24 +164,78 @@ def make_model_settings(
def _prompt_cache_extra_args(model_name: str) -> dict[str, Any] | None:
"""LiteLLM ``cache_control_injection_points`` for Claude prompt caching.
"""LiteLLM ``extra_args`` that enable Anthropic/Bedrock prompt caching.
System prompt + rolling last-message breakpoint everywhere; ``tool_config``
only on Bedrock Converse (the only route whose LiteLLM transform consumes
it — elsewhere it leaks onto the wire and native Anthropic 400s). Unmapped
Bedrock models get no points at all: Bedrock rejects the passed-through
field outright.
A Strix scan is a long, multi-turn agentic loop that re-sends a large,
STABLE prefix every turn — the system prompt plus the tool schemas — AND an
append-only conversation transcript that only grows. Without caching
breakpoints the whole request is re-tokenised and billed at the full input
rate on every turn; on Claude that is the single biggest lever on scan cost
(measured: ``cache-read 0% -> ~66%`` on a real scan once these points are set).
This mirrors the caching policy of production agent harnesses (e.g.
anomalyco/opencode's ``cache-policy``): cache the tool schemas, the system
prompt, and the latest conversation message, capped at Anthropic's 4
breakpoints. We express it through LiteLLM's ``cache_control_injection_points``
— the ``AnthropicCacheControlHook`` fires on that kwarg and emits the
provider-appropriate breakpoint (Anthropic ``cache_control``; Bedrock
Converse ``cachePoint``). ``LitellmModel`` forwards ``ModelSettings.extra_args``
straight into ``litellm.acompletion()``, so passing the points there is all
that is required; this is the LiteLLM-config-layer approach the Agents SDK
maintainer prescribed over a native ``ModelSettings`` caching flag
(openai/openai-agents-python#3008 / #3009).
Returns ``None`` (a strict no-op — the hook never fires) for every route
that would not benefit or could break:
- non-Claude models, and
- Bedrock Claude routes LiteLLM can't confirm as cache-capable. Bedrock's
Converse API rejects unknown request fields outright
(``ValidationException: cache_control_injection_points: Extra inputs are
not permitted``) and LiteLLM only consumes the marker for models its
model map recognises; an unmapped Bedrock model would pass the marker
straight through and crash the first call. Only Bedrock hard-rejects, so
only Bedrock is guarded — gating Anthropic-native/Vertex/OpenRouter on
confirmed support would needlessly disable caching for capable models.
"""
if not is_claude_model(model_name):
return None
if is_bedrock_route(model_name) and not bedrock_route_supports_prompt_caching(model_name):
return None
points = _prompt_cache_injection_points(model_name)
return {"cache_control_injection_points": points}
def _prompt_cache_injection_points(model_name: str) -> list[dict[str, Any]]:
"""Cache breakpoints for a Claude route (system + tools + latest message).
At most 3 of Anthropic's 4 allowed breakpoints, leaving headroom:
- system prompt (``role: system``) — the largest repeated span.
- tool schemas (``tool_config``) — Bedrock Converse ONLY. LiteLLM's
``tool_config`` location is implemented solely by the Bedrock Converse
transform (which appends a ``cachePoint`` to the tool list); on any other
route it is not consumed and would leak onto the wire as an unknown
top-level ``cache_control_injection_points`` field. It is also redundant
elsewhere: Anthropic orders tools BEFORE the system prompt, so the system
breakpoint already caches the tool schemas in the shared prefix.
- latest message (``index: -1``) — a ROLLING breakpoint on the last
message. A scan transcript is append-only (prior turns are immutable, each
turn just appends new assistant/tool messages), so without it the growing
body is re-sent at full input price every turn and cache-read decays as a
denominator effect even though the prefix keeps hitting. Re-caching the
whole prefix-so-far each turn keeps cache-read high on long scans. (This
is the Strix analogue of opencode's ``latest-user-message``; ``index: -1``
tracks the true tail because Strix appends tool-role, not user-role,
messages each turn.)
Unrecognised locations degrade gracefully on older LiteLLM — they are simply
not injected (no error).
"""
points: list[dict[str, Any]] = [{"location": "message", "role": "system"}]
if is_bedrock_route(model_name):
points.append({"location": "tool_config"})
points.append({"location": "message", "index": -1})
return {"cache_control_injection_points": points}
return points
def child_initial_input(
+3 -30
View File
@@ -31,7 +31,7 @@ from strix.core.execution import (
from strix.core.execution import (
spawn_child_agent as start_child_agent,
)
from strix.core.hooks import BudgetExceededError, ReportUsageHooks, recomputed_budget_flags
from strix.core.hooks import BudgetExceededError, ReportUsageHooks
from strix.core.inputs import (
DEFAULT_MAX_TURNS,
build_root_task,
@@ -40,7 +40,6 @@ from strix.core.inputs import (
)
from strix.core.paths import run_dir_for, runtime_state_dir
from strix.core.sessions import open_agent_session
from strix.report.state import get_global_report_state
from strix.runtime import session_manager
from strix.telemetry.logging import set_scan_id, setup_scan_logging
from strix.tools.output_store import (
@@ -186,18 +185,6 @@ async def run_strix_scan(
f"Cannot resume scan {scan_id}: missing SDK session database at {agents_db}",
)
await coordinator.restore(snap)
report_state = get_global_report_state()
if report_state is not None:
budget_stopped, reserve_stopped = recomputed_budget_flags(
report_state.get_total_llm_cost(),
max_budget_usd,
interactive=interactive,
)
await coordinator.reset_budget_stops(
budget_stopped=budget_stopped,
reserve_stopped=reserve_stopped,
budget_paused=interactive and coordinator.budget_paused,
)
for aid, parent in coordinator.parent_of.items():
if parent is None:
root_id = aid
@@ -248,7 +235,6 @@ async def run_strix_scan(
settings.llm.reasoning_effort,
model_name=resolved_model,
force_required_tool_choice=settings.llm.force_required_tool_choice,
custom_api_base=bool(settings.llm.api_base),
request_timeout=settings.llm.timeout,
prompt_cache=settings.llm.prompt_cache,
)
@@ -259,14 +245,7 @@ async def run_strix_scan(
sandbox=SandboxRunConfig(client=bundle["client"], session=bundle["session"]),
trace_include_sensitive_data=False,
)
hooks = ReportUsageHooks(
model=resolved_model,
max_budget_usd=max_budget_usd,
max_turns=max_turns,
interactive=interactive,
)
if interactive:
coordinator.set_budget_extender(hooks.extend_budget)
hooks = ReportUsageHooks(model=resolved_model, max_budget_usd=max_budget_usd)
scope_context = build_scope_context(scan_config)
root_context = _merge_root_prompt_context(scope_context, extra_system_prompt_context)
@@ -377,12 +356,6 @@ 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,
@@ -394,7 +367,7 @@ async def run_strix_scan(
agent_id=root_id,
interactive=interactive,
session=root_session,
start_parked=root_start_parked,
start_parked=bool(interactive and is_resume and root_status != "running"),
event_sink=event_sink,
hooks=hooks,
)
-2
View File
@@ -13,7 +13,6 @@ from rich.panel import Panel
from rich.text import Text
from strix.config import load_settings
from strix.core.inputs import DEFAULT_MAX_TURNS
from strix.core.runner import run_strix_scan
from strix.report.state import ReportState, set_global_report_state
from strix.runtime import session_manager
@@ -185,7 +184,6 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
local_sources=getattr(args, "local_sources", None) or [],
interactive=bool(getattr(args, "interactive", False)),
max_budget_usd=getattr(args, "max_budget_usd", None),
max_turns=getattr(args, "max_turns", DEFAULT_MAX_TURNS),
)
finally:
stop_updates.set()
+7 -35
View File
@@ -31,7 +31,6 @@ from strix.config.models import (
is_known_openai_bare_model,
is_recommended_or_frontier_model,
)
from strix.core.inputs import DEFAULT_MAX_TURNS
from strix.core.paths import run_dir_for, runtime_state_dir
from strix.interface.cli import run_cli
from strix.interface.tui import run_tui
@@ -211,7 +210,7 @@ def validate_environment() -> None:
padding=(1, 2),
)
logger.debug("Missing required env vars: %s", missing_required_vars)
logger.error("Missing required env vars: %s", missing_required_vars)
console.print("\n")
console.print(panel)
console.print()
@@ -224,7 +223,7 @@ def validate_environment() -> None:
def check_docker_installed() -> None:
if shutil.which("docker") is None:
logger.debug("Docker CLI not found in PATH")
logger.error("Docker CLI not found in PATH")
console = Console()
error_text = Text()
error_text.append("DOCKER NOT INSTALLED", style="bold red")
@@ -423,7 +422,7 @@ async def warm_up_llm(show_model_warning: bool = True) -> None:
logger.info("LLM warm-up succeeded for dedupe model %s", dedupe_model)
except Exception as e:
logger.debug("LLM warm-up failed", exc_info=True)
logger.exception("LLM warm-up failed")
error_text = Text()
sub_hint = _subscription_error_hint(e)
if sub_hint is not None:
@@ -482,16 +481,6 @@ def _positive_budget(value: str) -> float:
return budget
def _positive_int(value: str) -> int:
try:
parsed = int(value)
except ValueError as exc:
raise argparse.ArgumentTypeError(f"invalid int value: {value!r}") from exc
if parsed <= 0:
raise argparse.ArgumentTypeError("must be an integer greater than 0")
return parsed
def parse_arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Strix Multi-Agent Cybersecurity Penetration Testing Tool",
@@ -647,27 +636,10 @@ Examples:
)
parser.add_argument(
"--max-budget",
dest="max_budget_usd",
metavar="USD",
"--max-budget-usd",
type=_positive_budget,
default=None,
help=(
"Maximum LLM cost in USD (> 0). The scan stops cleanly when this limit is reached. "
"Graduated wrap-up warnings are sent to all agents as it is approached."
),
)
parser.add_argument(
"--max-turns",
dest="max_turns",
metavar="N",
type=_positive_int,
default=DEFAULT_MAX_TURNS,
help=(
"Maximum turns per agent (> 0, default %(default)s). Each agent is force-stopped "
"when it reaches this limit, with graduated wrap-up warnings as it is approached."
),
help="Maximum LLM cost in USD (> 0). The scan stops cleanly when this limit is reached.",
)
parser.add_argument(
@@ -884,7 +856,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])
@@ -946,7 +918,7 @@ def pull_docker_image() -> None:
last_update = process_pull_line(line, layers_info, status, last_update)
except DockerException as e:
logger.debug("Failed to pull docker image %s", image, exc_info=True)
logger.exception("Failed to pull docker image %s", image)
console.print()
error_text = Text()
error_text.append("FAILED TO PULL IMAGE", style="bold red")
+8 -35
View File
@@ -34,7 +34,6 @@ from textual.widgets.tree import TreeNode
from strix.config import load_settings
from strix.config.models import is_recommended_or_frontier_model
from strix.core.hooks import BudgetExceededError
from strix.core.inputs import DEFAULT_MAX_TURNS
from strix.core.runner import run_strix_scan
from strix.interface.tui.live_view import TuiLiveView
from strix.interface.tui.messages import send_user_message_to_agent
@@ -815,7 +814,6 @@ class StrixTUIApp(App): # type: ignore[misc]
self._scan_completed = threading.Event()
self._scan_error: BaseException | None = None
self._error_noted_agents: set[str] = set()
self._budget_pause_notified = False
self._spinner_frame_index: int = 0
self._sweep_num_squares: int = 6
@@ -1048,7 +1046,6 @@ class StrixTUIApp(App): # type: ignore[misc]
self.live_view.record_agent_error(agent_id, error)
else:
self._error_noted_agents.discard(agent_id)
self._notify_budget_pause(statuses)
if self._scan_loop is None or self._scan_loop.is_closed():
return
@@ -1060,19 +1057,6 @@ class StrixTUIApp(App): # type: ignore[misc]
self._agent_graph_sync_future = asyncio.run_coroutine_threadsafe(collect(), self._scan_loop)
def _notify_budget_pause(self, statuses: dict[str, Any]) -> None:
paused = any(status == "budget_paused" for status in statuses.values())
if paused and not self._budget_pause_notified:
self._budget_pause_notified = True
self.notify(
"Budget limit reached \u2014 agents paused. Send a message to continue "
"(this extends the budget), or ctrl-q to quit.",
severity="warning",
timeout=15,
)
elif not paused:
self._budget_pause_notified = False
def _update_agent_node(self, agent_id: str, agent_data: dict[str, Any]) -> bool:
if agent_id not in self.agent_nodes:
return False
@@ -1085,7 +1069,6 @@ class StrixTUIApp(App): # type: ignore[misc]
status_indicators = {
"running": "",
"waiting": "",
"budget_paused": "",
"completed": "🟢",
"failed": "🔴",
"crashed": "🔴",
@@ -1283,17 +1266,10 @@ class StrixTUIApp(App): # type: ignore[misc]
self._stop_dot_animation()
return (text, Text(), False)
if status in {"waiting", "budget_paused"}:
if status == "waiting":
text = Text()
keymap = Text()
if status == "budget_paused":
text.append("Budget limit reached", style="yellow")
text.append(" \u00b7 ", style="dim")
text.append("Send a message to continue", style="dim")
keymap = keymap_styled([("ctrl-q", "quit")])
else:
text.append("Send message to resume", style="dim")
return (text, keymap, False)
text.append("Send message to resume", style="dim")
return (text, Text(), False)
if status == "running":
if self._agent_has_real_activity(agent_id):
@@ -1518,7 +1494,6 @@ class StrixTUIApp(App): # type: ignore[misc]
coordinator=self.coordinator,
interactive=True,
max_budget_usd=getattr(self.args, "max_budget_usd", None),
max_turns=getattr(self.args, "max_turns", DEFAULT_MAX_TURNS),
event_sink=self._capture_sdk_event,
),
)
@@ -1526,7 +1501,10 @@ class StrixTUIApp(App): # type: ignore[misc]
except (KeyboardInterrupt, asyncio.CancelledError):
logger.info("Scan interrupted by user")
except BudgetExceededError:
logger.info("Scan stopped: --max-budget limit reached")
# Defensive: the runner stops the scan cleanly on budget and
# returns, so this normally never propagates. Treat it as a
# graceful stop, not a scan error, if it ever does.
logger.info("Scan stopped: --max-budget-usd limit reached")
except (ConnectionError, TimeoutError) as e:
logging.exception("Network error during scan")
self._scan_error = e
@@ -1581,7 +1559,6 @@ class StrixTUIApp(App): # type: ignore[misc]
status_indicators = {
"running": "",
"waiting": "",
"budget_paused": "",
"completed": "🟢",
"failed": "🔴",
"crashed": "🔴",
@@ -1628,7 +1605,6 @@ class StrixTUIApp(App): # type: ignore[misc]
status_indicators = {
"running": "",
"waiting": "",
"budget_paused": "",
"completed": "🟢",
"failed": "🔴",
"crashed": "🔴",
@@ -1753,10 +1729,7 @@ class StrixTUIApp(App): # type: ignore[misc]
message=message,
)
if not submitted:
if self._scan_completed.is_set():
self.notify("The scan has ended; message was not sent", severity="warning")
else:
self.notify("Scan loop is not ready; message was not sent", severity="warning")
self.notify("Scan loop is not ready; message was not sent", severity="warning")
return
self._displayed_events.clear()
+5 -7
View File
@@ -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_agent_and_call_id: dict[tuple[str, str], dict[str, Any]] = {}
self._tool_event_by_call_id: dict[str, dict[str, Any]] = {}
def hydrate_from_run_dir(self, run_dir: Path) -> None:
state_dir = runtime_state_dir(run_dir)
@@ -223,8 +223,7 @@ class TuiLiveView:
timestamp: str | None = None,
) -> None:
call_id = call["call_id"]
event_key = (agent_id, call_id)
existing = self._tool_event_by_agent_and_call_id.get(event_key)
existing = self._tool_event_by_call_id.get(call_id)
tool_data = {
"tool_name": call["tool_name"],
"args": call["args"],
@@ -234,7 +233,7 @@ class TuiLiveView:
}
if existing is None:
event = self._append_event(agent_id, "tool", tool_data, timestamp=timestamp)
self._tool_event_by_agent_and_call_id[event_key] = event
self._tool_event_by_call_id[call_id] = event
else:
existing["data"].update(tool_data)
self._bump_event(existing, timestamp=timestamp)
@@ -250,8 +249,7 @@ class TuiLiveView:
timestamp: str | None = None,
) -> None:
call_id = output["call_id"]
event_key = (agent_id, call_id)
event = self._tool_event_by_agent_and_call_id.get(event_key)
event = self._tool_event_by_call_id.get(call_id)
if event is None:
event = self._append_event(
agent_id,
@@ -265,7 +263,7 @@ class TuiLiveView:
},
timestamp=timestamp,
)
self._tool_event_by_agent_and_call_id[event_key] = event
self._tool_event_by_call_id[call_id] = event
result = _parse_json_value(output["output"])
event["data"]["result"] = result
+1 -7
View File
@@ -271,13 +271,7 @@ def _release_target() -> str | None:
if os_name is None:
return None
target = f"{os_name}-{arch}"
supported = {
"linux-x86_64",
"linux-arm64",
"macos-x86_64",
"macos-arm64",
"windows-x86_64",
}
supported = {"linux-x86_64", "macos-x86_64", "macos-arm64", "windows-x86_64"}
return target if target in supported else None
+8 -6
View File
@@ -11,10 +11,11 @@ import tempfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import Request, urlopen
import docker
import requests
from docker.errors import DockerException, ImageNotFound
from rich.console import Console
from rich.panel import Panel
@@ -1087,12 +1088,13 @@ def resolve_diff_scope_context(
def _is_http_git_repo(url: str) -> bool:
check_url = f"{url.rstrip('/')}/info/refs?service=git-upload-pack"
try:
resp = requests.get(check_url, headers={"User-Agent": "git/strix"}, timeout=10)
except (requests.RequestException, ValueError):
req = Request(check_url, headers={"User-Agent": "git/strix"}) # noqa: S310
with urlopen(req, timeout=10) as resp: # noqa: S310 # nosec B310
return "x-git-upload-pack-advertisement" in resp.headers.get("Content-Type", "")
except HTTPError as e:
return e.code == 401
except (URLError, OSError, ValueError):
return False
if resp.status_code >= 400:
return resp.status_code == 401
return "x-git-upload-pack-advertisement" in resp.headers.get("Content-Type", "")
def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR0911
+14 -10
View File
@@ -15,12 +15,12 @@ import base64
import contextlib
import json
import logging
import urllib.error
import urllib.request
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
import requests
from strix.config.loader import load_settings
@@ -147,17 +147,21 @@ def _post_json(path: str, payload: dict[str, Any], *, timeout: int) -> tuple[int
map, not raised.
"""
url = f"{_app_url()}{path}"
body = json.dumps(payload).encode("utf-8")
request = urllib.request.Request( # noqa: S310 - fixed https relay URL
url,
data=body,
headers={"Content-Type": "application/json", "Accept": "application/json"},
method="POST",
)
try:
response = requests.post(
url,
json=payload,
headers={"Accept": "application/json"},
timeout=timeout,
)
except requests.RequestException as exc:
with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 # nosec B310
return response.status, _parse_body(response.read())
except urllib.error.HTTPError as exc:
return exc.code, _parse_body(exc.read())
except (urllib.error.URLError, TimeoutError, OSError) as exc:
logger.warning("relay request to %s failed: %s", path, exc)
raise RelayError("unavailable") from exc
return response.status_code, _parse_body(response.content)
def _parse_body(raw: bytes) -> dict[str, Any]:
+4 -11
View File
@@ -107,11 +107,8 @@ def resolve_run_dir(base_dir: Path, run_param: str | None, default_run_dir: Path
return candidate
# 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"
# Name of the cookie carrying the per-process session capability.
SESSION_COOKIE = "strix_viewer_session"
class _ViewerState:
@@ -138,9 +135,6 @@ 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]:
@@ -482,7 +476,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(state.cookie_name, "")
supplied = self._cookies().get(SESSION_COOKIE, "")
return bool(supplied) and secrets.compare_digest(supplied, state.session_token)
def _token_presented(self, query: dict[str, list[str]]) -> bool:
@@ -518,7 +512,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
# SameSite=Strict (never sent from a cross-site context).
self.send_header(
"Set-Cookie",
f"{state.cookie_name}={state.session_token}; Path=/; HttpOnly; SameSite=Strict",
f"{SESSION_COOKIE}={state.session_token}; Path=/; HttpOnly; SameSite=Strict",
)
self.end_headers()
self.wfile.write(content)
@@ -592,7 +586,6 @@ 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)
-74
View File
@@ -1,7 +1,6 @@
import json
import logging
import subprocess
import threading
from collections.abc import Callable
from datetime import UTC, datetime
from importlib.metadata import PackageNotFoundError, version
@@ -96,8 +95,6 @@ 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:
@@ -510,72 +507,6 @@ 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,
@@ -610,11 +541,6 @@ 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)
+9 -3
View File
@@ -1,9 +1,9 @@
import json
import logging
import urllib.request
from datetime import datetime
from typing import TYPE_CHECKING, Any
import requests
from strix.config import load_settings
from strix.telemetry._common import (
SESSION_ID,
@@ -37,7 +37,13 @@ def _send(event: str, properties: dict[str, Any]) -> bool:
"distinct_id": SESSION_ID,
"properties": properties,
}
requests.post(f"{_POSTHOG_HOST}/capture/", json=payload, timeout=10)
req = urllib.request.Request( # noqa: S310
f"{_POSTHOG_HOST}/capture/",
data=json.dumps(payload).encode(),
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=10): # noqa: S310 # nosec B310
pass
except Exception: # noqa: BLE001
logger.debug("posthog send failed for event %s", event, exc_info=True)
return False
+4 -3
View File
@@ -2,11 +2,10 @@ from __future__ import annotations
import logging
import urllib.parse
import urllib.request
from datetime import datetime
from typing import TYPE_CHECKING, Any
import requests
from strix.config import load_settings
from strix.telemetry._common import (
SESSION_ID,
@@ -43,7 +42,9 @@ def _send(event: str, properties: dict[str, Any]) -> bool:
url = f"{_SCARF_ENDPOINT}{path}"
if query:
url = f"{url}?{query}"
requests.post(url, timeout=10)
req = urllib.request.Request(url, method="POST") # noqa: S310
with urllib.request.urlopen(req, timeout=10): # noqa: S310 # nosec B310
pass
except Exception: # noqa: BLE001
logger.debug("scarf send failed for event %s", event, exc_info=True)
return False
-2
View File
@@ -253,8 +253,6 @@ async def finish_scan(
parent_id = inner.get("parent_id")
if coordinator is not None and parent_id is None and me is not None:
active_agents = await coordinator.active_agents_except(me)
if active_agents and coordinator.reserve_stopped:
active_agents = []
else:
active_agents = []
-14
View File
@@ -7,10 +7,8 @@ import hashlib
import json
import time
from typing import TYPE_CHECKING, Any
from unittest import mock
import pytest
import requests
from strix.config import codex
@@ -54,18 +52,6 @@ def test_authorize_url_carries_pkce_and_client() -> None:
assert "state=st8" in url
def test_post_form_returns_parsed_body() -> None:
resp = mock.MagicMock()
resp.status_code = 200
resp.content = b'{"access_token": "tok"}'
with mock.patch.object(requests, "post", return_value=resp) as post:
data = codex._post_form({"grant_type": "refresh_token"})
assert data == {"access_token": "tok"}
assert post.call_args.kwargs["timeout"] == codex._TOKEN_TIMEOUT
@pytest.mark.parametrize(
("value", "expected"),
[
+2 -105
View File
@@ -7,25 +7,9 @@ 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,
_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()
from strix.config.models import _configure_litellm_compatibility
from strix.report.state import litellm_cost_callback
def test_streaming_logging_stays_enabled_for_cost_callback() -> None:
@@ -167,90 +151,3 @@ 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
)
-357
View File
@@ -1,357 +0,0 @@
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock, patch
import pytest
from strix.core import execution
from strix.core.agents import AgentCoordinator
from strix.core.execution import _start_child_runner, run_agent_loop
from strix.core.hooks import BudgetExceededError, ReportUsageHooks
from strix.core.sessions import open_agent_session
if TYPE_CHECKING:
from collections.abc import AsyncIterator, Callable
from pathlib import Path
MAX_BUDGET = 10.0
COST_PER_CALL = 1.0
class _FakeLedger:
def __init__(self) -> None:
self.cost = 0.0
self.calls: list[str] = []
def record_sdk_usage(self, **_kwargs: Any) -> None:
return
def get_total_llm_cost(self) -> float:
return self.cost
class _FakeStream:
def __init__(
self,
*,
ledger: _FakeLedger,
hooks: ReportUsageHooks,
context: dict[str, Any],
agent: Any,
) -> None:
self._ledger = ledger
self._hooks = hooks
self._context = context
self._agent = agent
self.run_loop_exception: BaseException | None = None
self.final_output = None
async def stream_events(self) -> AsyncIterator[Any]:
self._ledger.cost += COST_PER_CALL
self._ledger.calls.append(str(self._context.get("agent_id")))
ctx_wrapper = MagicMock()
ctx_wrapper.context = self._context
try:
await self._hooks.on_llm_end(ctx_wrapper, self._agent, MagicMock())
except Exception as exc: # noqa: BLE001
self.run_loop_exception = exc
items: tuple[Any, ...] = ()
for item in items:
yield item
def cancel(self, mode: str = "immediate") -> None: # noqa: ARG002
return
def _fake_runner(ledger: _FakeLedger) -> Any:
class _FakeRunner:
@staticmethod
def run_streamed(
agent: Any,
input: Any, # noqa: A002, ARG004
*,
run_config: Any, # noqa: ARG004
context: dict[str, Any],
max_turns: int, # noqa: ARG004
session: Any, # noqa: ARG004
hooks: ReportUsageHooks,
) -> _FakeStream:
return _FakeStream(ledger=ledger, hooks=hooks, context=context, agent=agent)
return _FakeRunner
async def _noop_compact(*_args: Any, **_kwargs: Any) -> bool:
return False
async def _wait_until(predicate: Callable[[], bool], *, timeout: float = 5.0) -> None:
async def _poll() -> None:
while not predicate():
await asyncio.sleep(0.01)
await asyncio.wait_for(_poll(), timeout=timeout)
@pytest.mark.asyncio
async def test_full_budget_lifecycle_reserve_then_cap( # noqa: PLR0915
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
ledger = _FakeLedger()
hooks = ReportUsageHooks(model="test-model", max_budget_usd=MAX_BUDGET)
monkeypatch.setattr(execution, "Runner", _fake_runner(ledger))
monkeypatch.setattr(execution, "_compact_session", _noop_compact)
coordinator = AgentCoordinator()
db_path = tmp_path / "agents.sqlite"
sessions: list[Any] = []
run_config = MagicMock()
await coordinator.register("root", "strix", parent_id=None)
root_session = open_agent_session("root", db_path)
sessions.append(root_session)
root_exc: list[BaseException] = []
async def _root_loop() -> None:
try:
await run_agent_loop(
agent=MagicMock(),
initial_input=[],
run_config=run_config,
context={"agent_id": "root", "parent_id": None},
max_turns=500,
coordinator=coordinator,
agent_id="root",
interactive=True,
session=root_session,
start_parked=True,
hooks=hooks,
)
except BaseException as exc:
root_exc.append(exc)
raise
with patch("strix.core.hooks.get_global_report_state", return_value=ledger):
root_task = asyncio.create_task(_root_loop())
await asyncio.sleep(0.05)
for child_id in ("child-a", "child-b"):
await coordinator.register(child_id, "recon", parent_id="root")
await _start_child_runner(
parent_ctx={"agent_id": "root", "parent_id": None},
coordinator=coordinator,
agents_db_path=db_path,
sessions_to_close=sessions,
run_config=run_config,
max_turns=500,
interactive=True,
child_agent=MagicMock(),
child_id=child_id,
name=f"recon-{child_id}",
parent_id="root",
task="probe things",
initial_input=[],
hooks=hooks,
)
await _wait_until(lambda: ledger.cost >= 2.0)
reserve_before = coordinator.reserve_stopped
assert reserve_before is False
async def _wait_spend_above(amount: float) -> None:
await _wait_until(lambda: ledger.cost > amount)
turn = 0
while ledger.cost < MAX_BUDGET * 0.90 - 1e-9:
target = ("child-a", "child-b")[turn % 2]
spent_before = ledger.cost
assert await coordinator.send(target, {"from": "user", "content": "keep going"})
await _wait_spend_above(spent_before)
turn += 1
await _wait_until(lambda: coordinator.reserve_stopped)
await _wait_until(
lambda: (
coordinator.statuses["child-a"] == "stopped"
and coordinator.statuses["child-b"] == "stopped"
)
)
assert coordinator.reserve_stopped is True
await _wait_until(lambda: coordinator.budget_stopped)
assert ledger.cost == pytest.approx(MAX_BUDGET)
assert len(ledger.calls) == 10
assert set(ledger.calls[:9]) == {"child-a", "child-b"}
assert ledger.calls[9] == "root"
root_items = await root_session.get_items()
notices = [item for item in root_items if "Budget reserve" in str(item)]
assert len(notices) == 1
with pytest.raises(BudgetExceededError):
await root_task
assert root_exc and isinstance(root_exc[0], BudgetExceededError)
assert {aid: str(status) for aid, status in coordinator.statuses.items()} == {
"root": "stopped",
"child-a": "stopped",
"child-b": "stopped",
}
assert coordinator.budget_stopped is True
assert coordinator.reserve_stopped is True
for session in sessions:
session.close()
@pytest.mark.asyncio
async def test_respawned_children_after_reserve_never_spend(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
ledger = _FakeLedger()
ledger.cost = 9.5
hooks = ReportUsageHooks(model="test-model", max_budget_usd=MAX_BUDGET)
monkeypatch.setattr(execution, "Runner", _fake_runner(ledger))
monkeypatch.setattr(execution, "_compact_session", _noop_compact)
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child-a", "recon", parent_id="root")
snap = await coordinator.snapshot()
snap["reserve_stopped"] = True
restored = AgentCoordinator()
await restored.restore(snap)
assert restored.reserve_stopped is True
sessions: list[Any] = []
with patch("strix.core.hooks.get_global_report_state", return_value=ledger):
await _start_child_runner(
parent_ctx={"agent_id": "root", "parent_id": None},
coordinator=restored,
agents_db_path=tmp_path / "agents.sqlite",
sessions_to_close=sessions,
run_config=MagicMock(),
max_turns=500,
interactive=True,
child_agent=MagicMock(),
child_id="child-a",
name="recon-child-a",
parent_id="root",
task="probe things",
initial_input=[],
hooks=hooks,
)
await _wait_until(lambda: restored.statuses["child-a"] == "stopped")
assert ledger.cost == pytest.approx(9.5)
assert ledger.calls == []
for session in sessions:
session.close()
@pytest.mark.asyncio
async def test_resumed_parked_root_after_reserve_is_renotified_and_finalizes(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
ledger = _FakeLedger()
ledger.cost = 9.0
hooks = ReportUsageHooks(model="test-model", max_budget_usd=MAX_BUDGET)
monkeypatch.setattr(execution, "Runner", _fake_runner(ledger))
monkeypatch.setattr(execution, "_compact_session", _noop_compact)
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.set_status("root", "waiting")
snap = await coordinator.snapshot()
snap["reserve_stopped"] = True
restored = AgentCoordinator()
await restored.restore(snap)
assert restored.reserve_stopped is True
root_session = open_agent_session("root", tmp_path / "agents.sqlite")
with patch("strix.core.hooks.get_global_report_state", return_value=ledger):
root_task = asyncio.create_task(
run_agent_loop(
agent=MagicMock(),
initial_input=[],
run_config=MagicMock(),
context={"agent_id": "root", "parent_id": None},
max_turns=500,
coordinator=restored,
agent_id="root",
interactive=True,
session=root_session,
start_parked=True,
hooks=hooks,
)
)
with pytest.raises(BudgetExceededError):
await asyncio.wait_for(root_task, timeout=5.0)
assert ledger.calls == ["root"]
assert ledger.cost == pytest.approx(MAX_BUDGET)
root_items = await root_session.get_items()
notices = [item for item in root_items if "Budget reserve" in str(item)]
assert len(notices) == 1
root_session.close()
@pytest.mark.asyncio
async def test_interactive_budget_pause_then_user_message_extends_and_resumes(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
ledger = _FakeLedger()
ledger.cost = 9.0
hooks = ReportUsageHooks(model="test-model", max_budget_usd=MAX_BUDGET, interactive=True)
monkeypatch.setattr(execution, "Runner", _fake_runner(ledger))
monkeypatch.setattr(execution, "_compact_session", _noop_compact)
coordinator = AgentCoordinator()
coordinator.set_budget_extender(hooks.extend_budget)
await coordinator.register("root", "strix", parent_id=None)
root_session = open_agent_session("root", tmp_path / "agents.sqlite")
with patch("strix.core.hooks.get_global_report_state", return_value=ledger):
root_task = asyncio.create_task(
run_agent_loop(
agent=MagicMock(),
initial_input=[],
run_config=MagicMock(),
context={"agent_id": "root", "parent_id": None},
max_turns=500,
coordinator=coordinator,
agent_id="root",
interactive=True,
session=root_session,
start_parked=True,
hooks=hooks,
)
)
await asyncio.sleep(0.05)
assert await coordinator.send("root", {"from": "user", "content": "go"})
await _wait_until(lambda: coordinator.budget_paused)
assert coordinator.statuses["root"] == "budget_paused"
assert ledger.cost == pytest.approx(MAX_BUDGET)
assert not root_task.done()
assert coordinator.budget_stopped is False
assert await coordinator.send("root", {"from": "user", "content": "keep going"})
await _wait_until(lambda: not coordinator.budget_paused)
await _wait_until(lambda: ledger.cost > MAX_BUDGET)
await _wait_until(lambda: coordinator.statuses["root"] == "waiting")
assert not root_task.done()
root_task.cancel()
await root_task
root_session.close()
-534
View File
@@ -3,273 +3,10 @@
from __future__ import annotations
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 (
_handle_content_guardrail,
_notify_parent_on_terminal,
_notify_root_on_budget_reserve,
respawn_subagents,
)
from strix.tools.finish.tool import finish_scan
async def _call_finish_scan(
coordinator: AgentCoordinator, agent_id: str, parent_id: str | None
) -> dict[str, Any]:
ctx = ToolContext(
context={"coordinator": coordinator, "agent_id": agent_id, "parent_id": parent_id},
tool_name="finish_scan",
tool_call_id="call-1",
tool_arguments="{}",
)
fields = ("executive_summary", "methodology", "technical_analysis", "recommendations")
result: str = await finish_scan.on_invoke_tool(ctx, json.dumps(dict.fromkeys(fields, "x")))
parsed: dict[str, Any] = json.loads(result)
return parsed
@pytest.mark.asyncio
async def test_reserve_stop_notifies_root_once(monkeypatch: pytest.MonkeyPatch) -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child-a", "recon", parent_id="root")
await coordinator.register("child-b", "recon", parent_id="root")
sent: list[tuple[str, dict[str, Any]]] = []
async def _record(target_agent_id: str, message: dict[str, Any]) -> bool:
sent.append((target_agent_id, message))
return True
monkeypatch.setattr(coordinator, "send", _record)
await _notify_root_on_budget_reserve(coordinator)
await _notify_root_on_budget_reserve(coordinator)
assert len(sent) == 1
target, message = sent[0]
assert target == "root"
assert message["type"] == "budget_reserve_stop"
assert "finish_scan" in str(message["content"])
@pytest.mark.asyncio
async def test_concurrent_reserve_claims_yield_single_root() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
for i in range(12):
await coordinator.register(f"child-{i}", "recon", parent_id="root")
results = await asyncio.gather(*(coordinator.claim_reserve_notification() for _ in range(12)))
assert results.count("root") == 1
assert all(r is None for r in results if r != "root")
@pytest.mark.asyncio
async def test_claim_reserve_sets_flag_and_wakes_parked_agents() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "recon", parent_id="root")
flag_before = coordinator.reserve_stopped
assert flag_before is False
waiter = asyncio.create_task(coordinator.wait_for_message("child"))
await asyncio.sleep(0)
assert not waiter.done()
await coordinator.claim_reserve_notification()
flag_after = coordinator.reserve_stopped
assert flag_after is True
await asyncio.wait_for(waiter, timeout=1.0)
@pytest.mark.asyncio
async def test_finish_scan_bypasses_active_agent_guard_after_reserve() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "recon", parent_id="root")
await coordinator.set_status("child", "running")
blocked = await _call_finish_scan(coordinator, "root", None)
assert blocked["scan_completed"] is False
assert blocked["active_agents"]
await coordinator.claim_reserve_notification()
finished = await _call_finish_scan(coordinator, "root", None)
assert finished["scan_completed"] is True
assert coordinator.statuses["root"] == "completed"
@pytest.mark.asyncio
async def test_finish_scan_gate_ignores_sub_agent_caller() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "recon", parent_id="root")
await coordinator.set_status("child", "running")
result = await _call_finish_scan(coordinator, "child", "root")
assert "active_agents" not in result
assert result["success"] is False
assert "root" in result["error"]
@pytest.mark.asyncio
async def test_reserve_stop_notify_noop_without_root(monkeypatch: pytest.MonkeyPatch) -> None:
coordinator = AgentCoordinator()
await coordinator.register("child", "recon", parent_id="missing")
sent: list[tuple[str, dict[str, Any]]] = []
async def _record(target_agent_id: str, message: dict[str, Any]) -> bool:
sent.append((target_agent_id, message))
return True
monkeypatch.setattr(coordinator, "send", _record)
await _notify_root_on_budget_reserve(coordinator)
assert sent == []
@pytest.mark.asyncio
async def test_snapshot_round_trip_preserves_stop_flags() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.trigger_budget_stop()
await coordinator.claim_reserve_notification()
snap = await coordinator.snapshot()
assert snap["budget_stopped"] is True
assert snap["reserve_stopped"] is True
restored = AgentCoordinator()
await restored.restore(snap)
assert restored.budget_stopped is True
assert restored.reserve_stopped is True
@pytest.mark.asyncio
async def test_legacy_snapshot_without_stop_flags_defaults_to_false() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
snap = await coordinator.snapshot()
del snap["budget_stopped"]
del snap["reserve_stopped"]
restored = AgentCoordinator()
await restored.restore(snap)
assert restored.budget_stopped is False
assert restored.reserve_stopped is False
@pytest.mark.asyncio
async def test_randomized_reserve_claim_race_many_interleavings() -> None:
for seed in range(25):
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
child_ids = [f"child-{i}" for i in range(8)]
for child_id in child_ids:
await coordinator.register(child_id, "recon", parent_id="root")
waiters = [asyncio.create_task(coordinator.wait_for_message(cid)) for cid in child_ids]
await asyncio.sleep(0)
async def _claim(delay: float, coord: AgentCoordinator = coordinator) -> str | None:
await asyncio.sleep(delay)
return await coord.claim_reserve_notification()
delays = [((seed * 31 + i * 17) % 50) / 10_000 for i in range(len(child_ids))]
results = await asyncio.gather(*(_claim(delay) for delay in delays))
assert results.count("root") == 1, f"seed {seed}: expected exactly one winner"
await asyncio.wait_for(asyncio.gather(*waiters), timeout=1.0)
assert coordinator.reserve_stopped is True
@pytest.mark.asyncio
async def test_reserve_claim_never_loses_root_wake() -> None:
for _ in range(10):
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "recon", parent_id="root")
root_waiter = asyncio.create_task(coordinator.wait_for_message("root"))
await asyncio.sleep(0)
assert not root_waiter.done()
await coordinator.claim_reserve_notification()
await asyncio.sleep(0)
async with coordinator._lock:
coordinator.pending_counts["root"] = 1
coordinator.runtimes["root"].wake.set()
await asyncio.wait_for(root_waiter, timeout=1.0)
@pytest.mark.asyncio
async def test_budget_stop_takes_precedence_over_reserve_for_all_roles() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "recon", parent_id="root")
await coordinator.claim_reserve_notification()
await coordinator.trigger_budget_stop()
await asyncio.wait_for(coordinator.wait_for_message("root"), timeout=1.0)
await asyncio.wait_for(coordinator.wait_for_message("child"), timeout=1.0)
assert coordinator.budget_stopped is True
assert coordinator.reserve_stopped is True
@pytest.mark.asyncio
async def test_root_not_released_by_reserve_alone() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "recon", parent_id="root")
await coordinator.claim_reserve_notification()
root_waiter = asyncio.create_task(coordinator.wait_for_message("root"))
await asyncio.sleep(0.02)
assert not root_waiter.done()
root_waiter.cancel()
with contextlib.suppress(asyncio.CancelledError):
await root_waiter
@pytest.mark.asyncio
async def test_snapshot_during_concurrent_claims_is_consistent() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
for i in range(6):
await coordinator.register(f"child-{i}", "recon", parent_id="root")
claims = [asyncio.create_task(coordinator.claim_reserve_notification()) for _ in range(6)]
snap = await coordinator.snapshot()
await asyncio.gather(*claims)
assert isinstance(snap["reserve_stopped"], bool)
final_snap = await coordinator.snapshot()
assert final_snap["reserve_stopped"] is True
restored = AgentCoordinator()
await restored.restore(final_snap)
assert restored.reserve_stopped is True
assert await restored.claim_reserve_notification() is None
@pytest.mark.asyncio
@@ -305,274 +42,3 @@ async def test_wait_for_message_returns_immediately_after_budget_stop() -> None:
# No pending messages, but the stop flag short-circuits the wait.
await asyncio.wait_for(coordinator.wait_for_message("agent"), timeout=1.0)
@pytest.mark.asyncio
async def test_pause_for_budget_sets_flag_and_status() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.pause_for_budget("root")
assert coordinator.budget_paused is True
assert coordinator.statuses["root"] == "budget_paused"
@pytest.mark.asyncio
async def test_resume_from_budget_pause_extends_and_nudges(
monkeypatch: pytest.MonkeyPatch,
) -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child-a", "recon", parent_id="root")
await coordinator.register("child-b", "recon", parent_id="root")
await coordinator.pause_for_budget("root")
await coordinator.pause_for_budget("child-a")
await coordinator.pause_for_budget("child-b")
extensions: list[int] = []
coordinator.set_budget_extender(lambda: extensions.append(1))
sent: list[tuple[str, dict[str, Any]]] = []
async def _record(target_agent_id: str, message: dict[str, Any]) -> bool:
sent.append((target_agent_id, message))
return True
monkeypatch.setattr(coordinator, "send", _record)
await coordinator.resume_from_budget_pause(exclude="root")
assert coordinator.budget_paused is False
assert len(extensions) == 1
assert all(coordinator.statuses[aid] == "waiting" for aid in ("root", "child-a", "child-b"))
assert sorted(target for target, _ in sent) == ["child-a", "child-b"]
assert all(message["type"] == "budget_extended" for _, message in sent)
await coordinator.resume_from_budget_pause(exclude="root")
assert len(extensions) == 1
@pytest.mark.asyncio
async def test_user_send_resumes_budget_pause(tmp_path: Any) -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
session = SQLiteSession("root", tmp_path / "agents.db")
await coordinator.attach_runtime("root", session=session)
await coordinator.pause_for_budget("root")
extensions: list[int] = []
coordinator.set_budget_extender(lambda: extensions.append(1))
delivered = await coordinator.send("root", {"from": "user", "content": "keep going"})
assert delivered is True
assert coordinator.budget_paused is False
assert len(extensions) == 1
assert coordinator.statuses["root"] == "waiting"
assert coordinator.pending_counts["root"] == 1
session.close()
@pytest.mark.asyncio
async def test_non_user_send_does_not_resume_budget_pause(tmp_path: Any) -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
session = SQLiteSession("root", tmp_path / "agents.db")
await coordinator.attach_runtime("root", session=session)
await coordinator.pause_for_budget("root")
extensions: list[int] = []
coordinator.set_budget_extender(lambda: extensions.append(1))
await coordinator.send("root", {"from": "system", "content": "status"})
assert coordinator.budget_paused is True
assert extensions == []
assert coordinator.statuses["root"] == "budget_paused"
session.close()
@pytest.mark.asyncio
async def test_reset_budget_stops_clears_pause_and_normalizes_statuses() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.trigger_budget_stop()
await coordinator.claim_reserve_notification()
await coordinator.pause_for_budget("root")
await coordinator.reset_budget_stops(budget_stopped=False, reserve_stopped=False)
assert coordinator.budget_stopped is False
assert coordinator.reserve_stopped is False
assert coordinator.budget_paused is False
assert coordinator.statuses["root"] == "waiting"
@pytest.mark.asyncio
async def test_reset_budget_stops_can_preserve_pause() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.pause_for_budget("root")
await coordinator.reset_budget_stops(
budget_stopped=False, reserve_stopped=False, budget_paused=True
)
assert coordinator.budget_paused is True
assert coordinator.statuses["root"] == "budget_paused"
@pytest.mark.asyncio
async def test_snapshot_round_trip_preserves_budget_pause() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.pause_for_budget("root")
snap = await coordinator.snapshot()
assert snap["budget_paused"] is True
restored = AgentCoordinator()
await restored.restore(snap)
assert restored.budget_paused is True
assert restored.statuses["root"] == "budget_paused"
@pytest.mark.asyncio
@pytest.mark.parametrize("status", ["stopped", "failed", "crashed"])
async def test_terminal_child_wakes_parked_parent(tmp_path: Any, status: str) -> None:
# Regression for #870: a child reaching a terminal state (e.g. MaxTurnsExceeded
# -> "stopped") must wake the parent parked in wait_for_message, so the root can
# finalize the scan instead of hanging for a completion report that never arrives.
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child", "SQL Injection", parent_id="root")
session = SQLiteSession("root", tmp_path / "agents.db")
await coordinator.attach_runtime("root", session=session)
root_waiter = asyncio.create_task(coordinator.wait_for_message("root"))
await asyncio.sleep(0)
assert not root_waiter.done()
await coordinator.set_status("child", status, error="Max turns (500) exceeded")
await _notify_parent_on_terminal(coordinator, "child", status)
await asyncio.wait_for(root_waiter, timeout=1.0)
assert coordinator.pending_counts.get("root", 0) > 0
session.close()
@pytest.mark.asyncio
async def test_notify_parent_on_terminal_ignores_non_terminal_status(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)
await _notify_parent_on_terminal(coordinator, "child", "waiting")
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
-157
View File
@@ -1,157 +0,0 @@
from __future__ import annotations
from typing import Any, cast
import httpx
import pytest
from agents import RunConfig, Runner
from openai import (
APIConnectionError,
APIError,
APIStatusError,
APITimeoutError,
BadRequestError,
InternalServerError,
RateLimitError,
)
from strix.core import execution
from strix.core.agents import AgentCoordinator
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 _status_error(status: int) -> APIStatusError:
return APIStatusError(
f"status {status}",
response=httpx.Response(status_code=status, request=_request()),
body=None,
)
def test_midstream_api_error_is_transient() -> None:
assert execution._is_transient_model_error(_midstream_api_error()) is True
def test_network_errors_are_transient() -> None:
assert execution._is_transient_model_error(APITimeoutError(_request())) is True
assert execution._is_transient_model_error(APIConnectionError(request=_request())) is True
def test_server_errors_are_transient() -> None:
assert (
execution._is_transient_model_error(
InternalServerError("boom", response=httpx.Response(500, request=_request()), body=None)
)
is True
)
for status in (502, 503, 504, 408):
assert execution._is_transient_model_error(_status_error(status)) is True
def test_rate_limit_is_not_retried_here() -> None:
rate_limited = RateLimitError(
"slow down", response=httpx.Response(429, request=_request()), body=None
)
assert execution._is_transient_model_error(rate_limited) is False
def test_client_errors_are_not_transient() -> None:
bad_request = BadRequestError(
"bad", response=httpx.Response(400, request=_request()), body=None
)
assert execution._is_transient_model_error(bad_request) is False
assert execution._is_transient_model_error(_status_error(404)) is False
assert execution._is_transient_model_error(ValueError("nope")) is False
class _FakeStream:
def __init__(self, exc: BaseException | None = None) -> None:
self._exc = exc
self._events: list[Any] = []
self.run_loop_exception: BaseException | None = None
async def stream_events(self) -> Any:
if self._exc is not None:
raise self._exc
for event in self._events:
yield event
def _patch_fast_backoff(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_BASE_DELAY_S", 0.0)
monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_MAX_DELAY_S", 0.0)
async def _run_once(
monkeypatch: pytest.MonkeyPatch,
streams: list[_FakeStream],
) -> Any:
_patch_fast_backoff(monkeypatch)
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 = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
result = 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,
)
return result, calls["n"], coordinator
@pytest.mark.asyncio
async def test_run_cycle_retries_transient_midstream_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
streams = [_FakeStream(exc=_midstream_api_error()), _FakeStream()]
result, attempts, _coordinator = await _run_once(monkeypatch, streams)
assert result is streams[1]
assert attempts == 2
@pytest.mark.asyncio
async def test_run_cycle_gives_up_after_max_retries(
monkeypatch: pytest.MonkeyPatch,
) -> None:
streams = [
_FakeStream(exc=_midstream_api_error())
for _ in range(execution._MAX_TRANSIENT_MODEL_RETRIES + 1)
]
with pytest.raises(APIError):
await _run_once(monkeypatch, streams)
@pytest.mark.asyncio
async def test_run_cycle_does_not_retry_permanent_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
bad_request = BadRequestError(
"bad", response=httpx.Response(400, request=_request()), body=None
)
streams = [_FakeStream(exc=bad_request), _FakeStream()]
with pytest.raises(BadRequestError):
await _run_once(monkeypatch, streams)
+3 -388
View File
@@ -2,18 +2,11 @@
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from strix.core.hooks import (
BudgetExceededError,
BudgetPausedError,
ReportUsageHooks,
SubagentBudgetReservedError,
recomputed_budget_flags,
)
from strix.core.hooks import BudgetExceededError, ReportUsageHooks
def _make_hooks(max_budget: float | None) -> ReportUsageHooks:
@@ -27,22 +20,9 @@ def _make_report_state(cost: float) -> MagicMock:
return state
def _make_context(agent_id: str = "test-agent", parent_id: str | None = None) -> MagicMock:
def _make_context(agent_id: str = "test-agent") -> MagicMock:
ctx: MagicMock = MagicMock()
ctx.context = {"agent_id": agent_id, "parent_id": parent_id}
return ctx
def _make_warn_context(
*,
requests: int,
parent_id: str | None = None,
agent_id: str = "test-agent",
) -> MagicMock:
ctx: MagicMock = MagicMock()
ctx.context = {"agent_id": agent_id, "parent_id": parent_id}
ctx.usage = MagicMock()
ctx.usage.requests = requests
ctx.context = {"agent_id": agent_id}
return ctx
@@ -109,127 +89,6 @@ async def test_error_message_includes_amounts() -> None:
assert "7.1234" in str(exc_info.value)
@pytest.mark.asyncio
async def test_subagent_stops_at_reserve() -> None:
hooks = _make_hooks(10.0)
state = _make_report_state(9.0)
with (
patch("strix.core.hooks.get_global_report_state", return_value=state),
pytest.raises(SubagentBudgetReservedError),
):
await hooks.on_llm_end(_make_context(parent_id="root-1"), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_subagent_below_reserve_does_not_raise() -> None:
hooks = _make_hooks(10.0)
state = _make_report_state(8.99)
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_end(_make_context(parent_id="root-1"), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_subagent_overshoot_to_full_budget_triggers_scan_wide_stop() -> None:
hooks = _make_hooks(10.0)
state = _make_report_state(10.5)
with (
patch("strix.core.hooks.get_global_report_state", return_value=state),
pytest.raises(BudgetExceededError),
):
await hooks.on_llm_end(_make_context(parent_id="root-1"), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_root_keeps_running_inside_reserve() -> None:
hooks = _make_hooks(10.0)
state = _make_report_state(9.5)
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_end(_make_context(parent_id=None), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_root_hard_stop_stays_at_full_budget() -> None:
hooks = _make_hooks(10.0)
state = _make_report_state(10.0)
with (
patch("strix.core.hooks.get_global_report_state", return_value=state),
pytest.raises(BudgetExceededError),
):
await hooks.on_llm_end(_make_context(parent_id=None), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_budget_warning_mentions_reserve() -> None:
hooks = ReportUsageHooks(model="test-model", max_budget_usd=10.0)
state = _make_report_state(7.5)
root_items: list[Any] = []
sub_items: list[Any] = []
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_start(
_make_warn_context(requests=0, parent_id=None), MagicMock(), None, root_items
)
await hooks.on_llm_start(
_make_warn_context(requests=0, parent_id="root-1"), MagicMock(), None, sub_items
)
assert "stopped at 90%" in root_items[0]["content"]
assert "stopped at 90%" in sub_items[0]["content"]
assert "root agent's final report" in sub_items[0]["content"]
@pytest.mark.asyncio
async def test_subagent_critical_budget_warning_reachable_before_reserve() -> None:
hooks = ReportUsageHooks(model="test-model", max_budget_usd=10.0)
state = _make_report_state(8.6)
sub_items: list[Any] = []
root_items: list[Any] = []
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_start(
_make_warn_context(requests=0, parent_id="root-1"), MagicMock(), None, sub_items
)
await hooks.on_llm_start(
_make_warn_context(requests=0, parent_id=None), MagicMock(), None, root_items
)
assert "[CRITICAL]" in sub_items[0]["content"]
assert "[URGENT]" in root_items[0]["content"]
@pytest.mark.parametrize(
("parent_id", "cost", "expected"),
[
("root-1", 0.0, None),
("root-1", 8.9999, None),
("root-1", 9.0, SubagentBudgetReservedError),
("root-1", 9.0001, SubagentBudgetReservedError),
("root-1", 9.5, SubagentBudgetReservedError),
("root-1", 9.9999, SubagentBudgetReservedError),
("root-1", 10.0, BudgetExceededError),
("root-1", 10.0001, BudgetExceededError),
("root-1", 25.0, BudgetExceededError),
(None, 0.0, None),
(None, 8.9999, None),
(None, 9.0, None),
(None, 9.5, None),
(None, 9.9999, None),
(None, 10.0, BudgetExceededError),
(None, 10.0001, BudgetExceededError),
(None, 25.0, BudgetExceededError),
],
)
@pytest.mark.asyncio
async def test_budget_enforcement_decision_table(
parent_id: str | None, cost: float, expected: type[Exception] | None
) -> None:
hooks = _make_hooks(10.0)
state = _make_report_state(cost)
with patch("strix.core.hooks.get_global_report_state", return_value=state):
if expected is None:
await hooks.on_llm_end(_make_context(parent_id=parent_id), MagicMock(), MagicMock())
else:
with pytest.raises(expected):
await hooks.on_llm_end(_make_context(parent_id=parent_id), MagicMock(), MagicMock())
state.record_sdk_usage.assert_called_once()
@pytest.mark.asyncio
async def test_no_raise_when_report_state_none() -> None:
hooks = _make_hooks(1.0)
@@ -247,247 +106,3 @@ def test_non_positive_budget_rejected(bad_budget: float) -> None:
def test_budget_exceeded_error_is_runtime_error() -> None:
err = BudgetExceededError("test")
assert isinstance(err, RuntimeError)
def test_non_positive_max_turns_rejected() -> None:
with pytest.raises(ValueError, match="positive integer"):
ReportUsageHooks(model="test-model", max_turns=0)
@pytest.mark.asyncio
async def test_no_turn_warning_below_first_band() -> None:
hooks = ReportUsageHooks(model="test-model", max_turns=100)
items: list[Any] = []
await hooks.on_llm_start(_make_warn_context(requests=68), MagicMock(), None, items)
assert items == []
@pytest.mark.asyncio
async def test_turn_warning_notice_band() -> None:
hooks = ReportUsageHooks(model="test-model", max_turns=100)
items: list[Any] = []
await hooks.on_llm_start(_make_warn_context(requests=69), MagicMock(), None, items)
assert len(items) == 1
content = items[0]["content"]
assert "[NOTICE]" in content
assert "finish_scan" in content
@pytest.mark.asyncio
async def test_turn_warning_escalates_and_names_subagent_tool() -> None:
hooks = ReportUsageHooks(model="test-model", max_turns=100)
items: list[Any] = []
await hooks.on_llm_start(
_make_warn_context(requests=95, parent_id="root-1"), MagicMock(), None, items
)
assert len(items) == 1
content = items[0]["content"]
assert "[CRITICAL]" in content
assert "agent_finish" in content
@pytest.mark.asyncio
async def test_turn_warning_root_directive_distinct_from_subagent() -> None:
hooks = ReportUsageHooks(model="test-model", max_turns=100)
root_items: list[Any] = []
await hooks.on_llm_start(
_make_warn_context(requests=85, parent_id=None), MagicMock(), None, root_items
)
root = root_items[0]["content"]
sub_items: list[Any] = []
await hooks.on_llm_start(
_make_warn_context(requests=85, parent_id="root-1"), MagicMock(), None, sub_items
)
sub = sub_items[0]["content"]
assert root != sub
assert "root agent" in root
assert "finish_scan" in root
assert "agent_finish" not in root
assert "whole scan" in root
assert "sub-agent" in sub
assert "agent_finish" in sub
assert "finish_scan" not in sub
assert "confirmed" in sub
@pytest.mark.asyncio
async def test_budget_warning_root_directive_distinct_from_subagent() -> None:
hooks = ReportUsageHooks(model="test-model", max_budget_usd=10.0)
state = _make_report_state(8.6)
root_items: list[Any] = []
sub_items: list[Any] = []
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_start(
_make_warn_context(requests=0, parent_id=None), MagicMock(), None, root_items
)
await hooks.on_llm_start(
_make_warn_context(requests=0, parent_id="root-1"), MagicMock(), None, sub_items
)
root = root_items[0]["content"]
sub = sub_items[0]["content"]
assert "finish_scan" in root and "agent_finish" not in root
assert "agent_finish" in sub and "finish_scan" not in sub
assert "confirmed" in sub
@pytest.mark.parametrize("parent_id", [None, "root-1"])
@pytest.mark.asyncio
async def test_turn_warning_directive_escalates_per_stage(parent_id: str | None) -> None:
hooks = ReportUsageHooks(model="test-model", max_turns=100)
contents: dict[str, str] = {}
for label, requests in (("notice", 69), ("urgent", 85), ("critical", 95)):
items: list[Any] = []
await hooks.on_llm_start(
_make_warn_context(requests=requests, parent_id=parent_id), MagicMock(), None, items
)
contents[label] = items[0]["content"]
assert len({contents["notice"], contents["urgent"], contents["critical"]}) == 3
assert "[NOTICE]" in contents["notice"] and "begin planning" in contents["notice"]
assert "[URGENT]" in contents["urgent"] and "prioritize" in contents["urgent"]
assert "[CRITICAL]" in contents["critical"] and "STOP" in contents["critical"]
@pytest.mark.asyncio
async def test_no_turn_warning_when_max_turns_unset() -> None:
hooks = ReportUsageHooks(model="test-model")
items: list[Any] = []
await hooks.on_llm_start(_make_warn_context(requests=999), MagicMock(), None, items)
assert items == []
@pytest.mark.asyncio
async def test_no_budget_warning_below_first_band() -> None:
hooks = ReportUsageHooks(model="test-model", max_budget_usd=10.0)
state = _make_report_state(6.9)
items: list[Any] = []
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_start(_make_warn_context(requests=0), MagicMock(), None, items)
assert items == []
@pytest.mark.asyncio
async def test_budget_warning_broadcast_content() -> None:
hooks = ReportUsageHooks(model="test-model", max_budget_usd=10.0)
state = _make_report_state(9.6)
items: list[Any] = []
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_start(_make_warn_context(requests=0), MagicMock(), None, items)
assert len(items) == 1
content = items[0]["content"]
assert "[CRITICAL]" in content
assert "shared across every agent" in content
@pytest.mark.asyncio
async def test_turn_and_budget_warnings_stack() -> None:
hooks = ReportUsageHooks(model="test-model", max_budget_usd=10.0, max_turns=100)
state = _make_report_state(8.6)
items: list[Any] = []
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_start(_make_warn_context(requests=89), MagicMock(), None, items)
assert len(items) == 2
joined = " ".join(i["content"] for i in items)
assert "Turn budget" in joined
assert "cost budget" in joined
def _make_interactive_hooks(max_budget: float | None) -> ReportUsageHooks:
return ReportUsageHooks(model="test-model", max_budget_usd=max_budget, interactive=True)
@pytest.mark.asyncio
async def test_interactive_at_budget_pauses_instead_of_stopping() -> None:
hooks = _make_interactive_hooks(10.0)
state = _make_report_state(10.0)
with (
patch("strix.core.hooks.get_global_report_state", return_value=state),
pytest.raises(BudgetPausedError),
):
await hooks.on_llm_end(_make_context(parent_id=None), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_interactive_subagent_has_no_reserve() -> None:
hooks = _make_interactive_hooks(10.0)
state = _make_report_state(9.5)
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_end(_make_context(parent_id="root-1"), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_interactive_subagent_pauses_at_full_budget() -> None:
hooks = _make_interactive_hooks(10.0)
state = _make_report_state(10.5)
with (
patch("strix.core.hooks.get_global_report_state", return_value=state),
pytest.raises(BudgetPausedError),
):
await hooks.on_llm_end(_make_context(parent_id="root-1"), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_extend_budget_lifts_the_pause() -> None:
hooks = _make_interactive_hooks(10.0)
state = _make_report_state(10.5)
hooks.extend_budget()
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_end(_make_context(parent_id=None), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_extend_budget_adds_original_amount_each_time() -> None:
hooks = _make_interactive_hooks(10.0)
hooks.extend_budget()
hooks.extend_budget()
state = _make_report_state(29.9)
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_end(_make_context(parent_id=None), MagicMock(), MagicMock())
state = _make_report_state(30.0)
with (
patch("strix.core.hooks.get_global_report_state", return_value=state),
pytest.raises(BudgetPausedError),
):
await hooks.on_llm_end(_make_context(parent_id=None), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_interactive_subagent_uses_root_warning_bands() -> None:
hooks = _make_interactive_hooks(10.0)
state = _make_report_state(7.4)
items: list[Any] = []
with patch("strix.core.hooks.get_global_report_state", return_value=state):
await hooks.on_llm_start(
_make_warn_context(requests=0, parent_id="root-1"), MagicMock(), None, items
)
assert len(items) == 1
content = items[0]["content"]
assert "[NOTICE]" in content
assert "paused until the user chooses to continue" in content
assert "reserve" not in content.lower()
@pytest.mark.parametrize(
("cost", "max_budget", "interactive", "expected"),
[
(0.0, None, False, (False, False)),
(100.0, None, False, (False, False)),
(5.0, 10.0, False, (False, False)),
(9.0, 10.0, False, (False, True)),
(10.0, 10.0, False, (True, True)),
(10.0, 20.0, False, (False, False)),
(10.0, 10.0, True, (False, False)),
],
)
def test_recomputed_budget_flags(
cost: float,
max_budget: float | None,
interactive: bool,
expected: tuple[bool, bool],
) -> None:
assert recomputed_budget_flags(cost, max_budget, interactive=interactive) == expected
+46 -49
View File
@@ -62,6 +62,8 @@ def _cache_points(model_name: str) -> Any:
def test_make_model_settings_enables_prompt_cache_for_bedrock_claude() -> None:
# Bedrock Converse is the only route whose transform consumes the
# ``tool_config`` location, so it gets all three breakpoints.
assert _cache_points("bedrock/global.anthropic.claude-opus-4-8") == [
{"location": "message", "role": "system"},
{"location": "tool_config"},
@@ -78,6 +80,11 @@ def test_make_model_settings_enables_prompt_cache_for_bedrock_claude() -> None:
],
)
def test_make_model_settings_enables_prompt_cache_for_non_bedrock_claude(model_name: str) -> None:
# Non-Bedrock routes get system + tail only: LiteLLM implements the
# ``tool_config`` location solely in the Bedrock Converse transform, so
# sending it elsewhere would leak an unknown top-level field (see the
# dedicated no-leak test). Tools are already cached by the system breakpoint
# there — Anthropic orders tools ahead of the system prompt in the prefix.
assert _cache_points(model_name) == [
{"location": "message", "role": "system"},
{"location": "message", "index": -1},
@@ -85,14 +92,25 @@ def test_make_model_settings_enables_prompt_cache_for_non_bedrock_claude(model_n
def test_tool_config_point_not_leaked_to_non_bedrock_claude() -> None:
# LiteLLM only consumes tool_config on Bedrock; elsewhere it leaks onto the
# wire and native Anthropic 400s.
# Regression guard: the ``tool_config`` injection point must NEVER be sent on
# a non-Bedrock route. LiteLLM leaves it on the outgoing body as a top-level
# ``cache_control_injection_points`` field there, which native Anthropic
# would reject as an unknown field.
for model in ("anthropic/claude-sonnet-4-5", "openrouter/anthropic/claude-3.5-sonnet"):
points = _cache_points(model) or []
assert all(p.get("location") != "tool_config" for p in points)
@pytest.mark.parametrize("model_name", ["gpt-5", "vertex_ai/gemini-2.5-pro", "openai/o3"])
def test_make_model_settings_no_prompt_cache_for_non_claude(model_name: str) -> None:
# No injection points for non-Claude models: the LiteLLM cache hook never
# fires, so this stays a strict no-op (won't emit cache_control to strict
# OpenAI-compatible endpoints).
assert make_model_settings(None, model_name=model_name).extra_args is None
def test_prompt_cache_can_be_disabled() -> None:
# STRIX_PROMPT_CACHE=false kill switch: no injection points even for Claude.
assert (
make_model_settings(
None, model_name="anthropic/claude-sonnet-4-5", prompt_cache=False
@@ -101,27 +119,36 @@ def test_prompt_cache_can_be_disabled() -> None:
)
@pytest.mark.parametrize("model_name", ["gpt-5", "vertex_ai/gemini-2.5-pro", "openai/o3"])
def test_make_model_settings_no_prompt_cache_for_non_claude(model_name: str) -> None:
assert make_model_settings(None, model_name=model_name).extra_args is None
def test_no_prompt_cache_for_unmapped_bedrock_claude_model(monkeypatch: Any) -> None:
# A Bedrock Claude model LiteLLM hasn't mapped must run uncached, not crash.
"""A BEDROCK Claude route LiteLLM has NOT mapped (a new release, or any model
when LiteLLM can't refresh its model map and falls back to a stale local
copy) must run UNCACHED, not crash. Bedrock's Converse API rejects the
unknown field outright (ValidationException 'cache_control_injection_points:
Extra inputs are not permitted'); LiteLLM only strips the marker for models
it recognises as cache-capable, so an unmapped model would 500 the first
call and fail the whole run."""
unmapped = "bedrock/global.anthropic.claude-brand-new-9"
# Simulate a model LiteLLM doesn't know: no cost-map entry, checker says no.
monkeypatch.setattr(litellm, "model_cost", {}, raising=False)
if getattr(getattr(litellm, "utils", None), "supports_prompt_caching", None):
monkeypatch.setattr(litellm.utils, "supports_prompt_caching", lambda *_a, **_k: False)
# Bedrock Claude by name, but unmapped → no injection points, no crash.
assert make_model_settings(None, model_name=unmapped).extra_args is None
def test_prompt_cache_kept_for_non_bedrock_claude_even_if_unmapped(monkeypatch: Any) -> None:
# Only Bedrock hard-rejects unknown cache fields, so only Bedrock is guarded.
"""Non-Bedrock Claude routes must KEEP caching-by-family even when LiteLLM
can't confirm support — those providers tolerate/ignore the marker (or
LiteLLM maps them under keys we don't resolve, e.g. OpenRouter), so gating
them on confirmed support would DISABLE caching for capable models a
regression. Only Bedrock hard-rejects, so only Bedrock is guarded."""
monkeypatch.setattr(litellm, "model_cost", {}, raising=False)
if getattr(getattr(litellm, "utils", None), "supports_prompt_caching", None):
monkeypatch.setattr(litellm.utils, "supports_prompt_caching", lambda *_a, **_k: False)
# Even with LiteLLM knowing nothing, an Anthropic-native / OpenRouter Claude
# still gets the injection points (system + tail, no tool_config).
for model in ("anthropic/claude-brand-new-9", "openrouter/anthropic/claude-brand-new"):
assert _cache_points(model) == [
{"location": "message", "role": "system"},
@@ -130,8 +157,16 @@ def test_prompt_cache_kept_for_non_bedrock_claude_even_if_unmapped(monkeypatch:
def test_conversation_tail_breakpoint_moves_with_appended_transcript() -> None:
# LiteLLM must place the index=-1 cache_control on the last message however
# long the transcript grows.
"""The tail breakpoint's premise, end-to-end: LiteLLM's own message-injection
logic must place the cache_control on the LAST message for both a short and a
long transcript i.e. it tracks the growing (append-only) tail rather than a
fixed position so the immutable prefix-so-far is cached and re-read next
turn.
Driven through the hook's static ``_apply_message_injections`` primitive
(stable across LiteLLM versions) rather than the prompt-manager entrypoint
(whose signature drifts).
"""
hook_mod = pytest.importorskip("litellm.integrations.anthropic_cache_control_hook")
apply = hook_mod.AnthropicCacheControlHook._apply_message_injections
points = _cache_points("bedrock/global.anthropic.claude-opus-4-8")
@@ -255,44 +290,6 @@ def test_make_model_settings_forces_required_for_anyllm_routed_openai_model() ->
assert settings.tool_choice == "required"
def test_make_model_settings_auto_forces_required_on_custom_openai_endpoint() -> None:
# GLM / Kimi via cortecs: openai/-routed reasoning model on a custom base.
settings = make_model_settings(
None,
model_name="openai/glm-5.2",
custom_api_base=True,
)
assert settings.tool_choice == "required"
def test_make_model_settings_auto_skips_required_without_custom_endpoint() -> None:
settings = make_model_settings(None, model_name="openai/glm-5.2")
assert settings.tool_choice is None
def test_make_model_settings_explicit_false_overrides_custom_endpoint() -> None:
settings = make_model_settings(
None,
model_name="openai/glm-5.2",
force_required_tool_choice=False,
custom_api_base=True,
)
assert settings.tool_choice is None
def test_make_model_settings_auto_skips_required_for_non_openai_custom_endpoint() -> None:
settings = make_model_settings(
None,
model_name="anthropic/claude-3-7-sonnet-latest",
custom_api_base=True,
)
assert settings.tool_choice is None
def test_make_model_settings_sets_request_timeout() -> None:
settings = make_model_settings(
"none",
-154
View File
@@ -1,154 +0,0 @@
from __future__ import annotations
import stat
import subprocess
import sys
import tarfile
from pathlib import Path
import pytest
RELEASE_VERSION = "9.9.9"
RELEASE_TARGET = "linux-arm64"
pytestmark = pytest.mark.skipif(
sys.platform == "win32",
reason="scripts/install.sh is a POSIX shell installer",
)
def _write_executable(path: Path, content: str) -> None:
path.write_text(content, encoding="utf-8")
path.chmod(path.stat().st_mode | stat.S_IXUSR)
def _create_release_archive(tmp_path: Path) -> Path:
binary_name = f"strix-{RELEASE_VERSION}-{RELEASE_TARGET}"
binary_path = tmp_path / binary_name
_write_executable(binary_path, f"#!/bin/sh\nprintf 'strix {RELEASE_VERSION}\\n'\n")
archive_path = tmp_path / f"{binary_name}.tar.gz"
with tarfile.open(archive_path, "w:gz") as archive:
archive.add(binary_path, arcname=binary_name)
return archive_path
def _create_mock_commands(tmp_path: Path, machine: str) -> Path:
mock_bin = tmp_path / "mock-bin"
mock_bin.mkdir()
_write_executable(
mock_bin / "uname",
f"""#!/bin/sh
case "$1" in
-s) echo Linux ;;
-m) echo {machine} ;;
*) echo "unexpected uname argument: $*" >&2; exit 1 ;;
esac
""",
)
_write_executable(mock_bin / "docker", "#!/bin/sh\nexit 0\n")
_write_executable(
mock_bin / "curl",
"""#!/bin/sh
output=""
while [ "$#" -gt 0 ]; do
if [ "$1" = "-o" ]; then
output="$2"
shift 2
continue
fi
printf '%s\\n' "$1" >> "$STRIX_TEST_CURL_LOG"
shift
done
cp "$STRIX_TEST_ARCHIVE" "$output"
""",
)
return mock_bin
def _create_installer_environment(
tmp_path: Path,
archive_path: Path,
mock_bin: Path,
) -> tuple[dict[str, str], Path, Path]:
"""Build the installer environment explicitly.
Every variable the installer reads is listed here, so no inherited value
(`XDG_CONFIG_HOME`, `GITHUB_ACTIONS`, `TMPDIR`, ...) can send a write
outside the sandbox or change the code path under test.
"""
home_path = tmp_path / "home"
home_path.mkdir()
download_path = tmp_path / "downloads"
download_path.mkdir()
curl_log_path = tmp_path / "curl.log"
environment = {
"HOME": str(home_path),
"XDG_CONFIG_HOME": str(home_path / ".config"),
"PATH": f"{mock_bin}:/usr/bin:/bin",
"SHELL": "/bin/bash",
"TMPDIR": str(download_path),
"STRIX_TEST_ARCHIVE": str(archive_path),
"STRIX_TEST_CURL_LOG": str(curl_log_path),
"VERSION": RELEASE_VERSION,
}
return environment, home_path, curl_log_path
def _run_installer(
repository_root: Path,
environment: dict[str, str],
) -> subprocess.CompletedProcess[str]:
return subprocess.run( # noqa: S603
["/bin/bash", str(repository_root / "scripts/install.sh")],
cwd=repository_root,
env=environment,
capture_output=True,
text=True,
check=False,
)
def test_installer_downloads_and_runs_linux_arm64_release(tmp_path: Path) -> None:
repository_root = Path(__file__).resolve().parents[1]
archive_path = _create_release_archive(tmp_path)
mock_bin = _create_mock_commands(tmp_path, machine="aarch64")
environment, home_path, curl_log_path = _create_installer_environment(
tmp_path,
archive_path,
mock_bin,
)
result = _run_installer(repository_root, environment)
assert result.returncode == 0, result.stderr
expected_filename = f"strix-{RELEASE_VERSION}-{RELEASE_TARGET}.tar.gz"
assert expected_filename in curl_log_path.read_text(encoding="utf-8")
installed_binary = home_path / ".strix/bin/strix"
installed_result = subprocess.run( # noqa: S603
[str(installed_binary), "--version"],
capture_output=True,
text=True,
check=True,
)
assert installed_result.stdout.strip() == f"strix {RELEASE_VERSION}"
def test_installer_rejects_unsupported_architecture(tmp_path: Path) -> None:
repository_root = Path(__file__).resolve().parents[1]
archive_path = _create_release_archive(tmp_path)
mock_bin = _create_mock_commands(tmp_path, machine="riscv64")
environment, home_path, curl_log_path = _create_installer_environment(
tmp_path,
archive_path,
mock_bin,
)
result = _run_installer(repository_root, environment)
assert result.returncode != 0
assert "Unsupported OS/Arch: linux/riscv64" in result.stdout
assert not curl_log_path.exists()
assert not (home_path / ".strix").exists()
+2 -4
View File
@@ -14,7 +14,6 @@ import strix.tools.notes.tools as notes_tools
import strix.tools.todo.tools as todo_tools
from strix.core import runner
from strix.core.agents import AgentCoordinator
from strix.runtime import session_manager
def _make_rate_limit_error() -> RateLimitError:
@@ -38,7 +37,6 @@ async def test_persistent_rate_limit_stops_gracefully(
model="openai/gpt-4o",
reasoning_effort="high",
force_required_tool_choice=False,
api_base=None,
timeout=300,
prompt_cache=True,
),
@@ -59,8 +57,8 @@ async def test_persistent_rate_limit_stops_gracefully(
async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
return None
monkeypatch.setattr(session_manager, "create_or_reuse", _create_or_reuse)
monkeypatch.setattr(session_manager, "cleanup", _cleanup)
monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse) # type: ignore[attr-defined]
monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup) # type: ignore[attr-defined]
monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: "")
-1
View File
@@ -46,7 +46,6 @@ def _patch_engine_scaffold(
model="openai/gpt-4o",
reasoning_effort="high",
force_required_tool_choice=False,
api_base=None,
timeout=300,
prompt_cache=True,
),
+9 -35
View File
@@ -159,40 +159,14 @@ def test_sha256_file(tmp_path: Path) -> None:
assert update_check._sha256_file(path) == hashlib.sha256(b"strix").hexdigest()
@pytest.mark.parametrize(
("system", "machine", "expected"),
[
("Linux", "x86_64", "linux-x86_64"),
("Linux", "aarch64", "linux-arm64"),
("Linux", "arm64", "linux-arm64"),
("Darwin", "arm64", "macos-arm64"),
("Darwin", "riscv64", None),
],
)
def test_release_target(
monkeypatch: pytest.MonkeyPatch,
system: str,
machine: str,
expected: str | None,
) -> None:
monkeypatch.setattr(platform, "system", lambda: system)
monkeypatch.setattr(platform, "machine", lambda: machine)
assert update_check._release_target() == expected
def test_self_update_uses_linux_arm64_release(monkeypatch: pytest.MonkeyPatch) -> None:
requested_update: list[tuple[str, str]] = []
def record_download(version: str, target: str, _console: Console) -> bool:
requested_update.append((version, target))
return True
monkeypatch.setattr(update_check, "is_binary_install", lambda: True)
monkeypatch.setattr(update_check, "get_version", lambda: "1.0.0")
def test_release_target(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(platform, "system", lambda: "Linux")
monkeypatch.setattr(platform, "machine", lambda: "aarch64")
monkeypatch.setattr(update_check, "_download_and_replace", record_download)
monkeypatch.setattr(platform, "machine", lambda: "x86_64")
assert update_check._release_target() == "linux-x86_64"
assert update_check.self_update(Console(file=io.StringIO()), version="1.1.0") is True
assert requested_update == [("1.1.0", "linux-arm64")]
monkeypatch.setattr(platform, "system", lambda: "Darwin")
monkeypatch.setattr(platform, "machine", lambda: "arm64")
assert update_check._release_target() == "macos-arm64"
monkeypatch.setattr(platform, "machine", lambda: "riscv64")
assert update_check._release_target() is None
+2 -125
View File
@@ -4,11 +4,9 @@ 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
@@ -88,75 +86,6 @@ 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
@@ -342,11 +271,6 @@ 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
@@ -387,7 +311,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 f"{_cookie_name(url)}=" in cookie
assert "strix_viewer_session=" in cookie
assert "HttpOnly" in cookie and "SameSite=Strict" in cookie
# Static assets never carry it.
@@ -420,7 +344,7 @@ def test_unauthorized_client_cannot_acquire_capability(
url,
"/api/agents/steer",
{"agent_id": "root", "message": "pwn"},
cookie=f"{_cookie_name(url)}=",
cookie="strix_viewer_session=",
)
assert status == 403
assert delivered == []
@@ -617,53 +541,6 @@ 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"
Generated
+1 -1
View File
@@ -2411,7 +2411,7 @@ wheels = [
[[package]]
name = "strix-agent"
version = "1.4.1"
version = "1.3.1"
source = { editable = "." }
dependencies = [
{ name = "caido-sdk-client" },