Compare commits

..
Author SHA1 Message Date
Ian 98707a785a fix(proxy): log all Caido SDK errors as warnings 2026-10-12 15:00:00 -04:00
Ian b58d3ac3c6 fix(proxy): log Caido timeouts as warnings 2026-10-12 14:51:10 -04:00
40 changed files with 75 additions and 1880 deletions

No files matched your search

-18
View File
@@ -33,18 +33,6 @@ Configure Strix using environment variables or a config file.
the LiteLLM and native OpenAI routing paths.
</ParamField>
<ParamField path="LLM_EXTRA_BODY" type="string">
Extra fields merged into the JSON body of every request to the main model, as a JSON object
(e.g. `{"service_tier":"flex","safety_identifier":"9f86d081..."}`). Use it
for provider parameters Strix doesn't expose, such as OpenAI's `service_tier`
or `safety_identifier`. Applies to both the LiteLLM and native OpenAI routing
paths. Fields are not validated, so the provider or gateway must accept them;
unknown fields are usually rejected. Extra fields can override fields Strix
sends and break requests, so don't set request-shaping fields such as `model`, `messages`, `input`,
`tools` or `stream`, and use `STRIX_REASONING_EFFORT` rather than
`reasoning_effort`.
</ParamField>
<ParamField path="LLM_TIMEOUT" default="300" type="integer">
Request timeout in seconds for LLM calls.
</ParamField>
@@ -100,12 +88,6 @@ affecting the agents that do the actual testing.
inherits `LLM_EXTRA_HEADERS`; set this when its endpoint needs custom headers.
</ParamField>
<ParamField path="DEDUPE_LLM_EXTRA_BODY" type="string">
Optional JSON object of extra request-body fields sent on every
deduplication-model request. A dedicated dedupe model never inherits
`LLM_EXTRA_BODY`; set this when its endpoint needs them too.
</ParamField>
<ParamField path="STRIX_DEDUPE_REASONING_EFFORT" type="string">
Reasoning effort for the deduplication model. Defaults to the model's own
baseline when unset.
-15
View File
@@ -68,21 +68,6 @@ export LLM_API_KEY="your-bearer-token" # sent as Authorization: Bearer
export LLM_EXTRA_HEADERS='{"X-Feature-Key":"value","X-Tenant":"acme"}'
```
### Extra request-body fields
Some provider parameters go in the request body rather than a header, and
Strix has no dedicated setting for them. Set them with `LLM_EXTRA_BODY` as a
JSON object — they are merged into the body of every request to the model. For example, to
use OpenAI's cheaper flex processing and attribute requests to a user with a
safety identifier (a hash, never a raw email):
```bash
export LLM_EXTRA_BODY='{"service_tier":"flex","safety_identifier":"9f86d081884c7d65"}'
```
Fields aren't validated, so use only fields your provider accepts, and don't
set fields Strix already sends, such as `model`, `messages` or `stream`.
For endpoints behind a private CA, point Strix at your certificate bundle with
the standard `SSL_CERT_FILE=/path/to/ca-bundle.pem` — never disable TLS
verification against a real endpoint.
+1 -49
View File
@@ -13,13 +13,11 @@ from agents.agent import ToolsToFinalOutputResult
from agents.sandbox import SandboxAgent
from agents.sandbox.capabilities import Filesystem, Shell
from agents.sandbox.errors import InvalidManifestPathError
from agents.sandbox.session.pty_types import PTY_YIELD_TIME_MS_MAX
from agents.tool import CustomTool, FunctionTool, Tool
from pydantic import ValidationError
from strix.agents.prompt import render_system_prompt
from strix.config import load_settings
from strix.config.tool_call_arguments import describe_malformed_arguments
from strix.tools.agents_graph.tools import (
agent_finish,
create_agent,
@@ -269,10 +267,6 @@ def _with_coerced_arguments(tool: FunctionTool) -> FunctionTool:
nullish = tool.name.startswith(_QUERY_TOOL_PREFIXES)
async def invoke(ctx: Any, raw_input: str) -> Any:
malformed = describe_malformed_arguments(tool.name, raw_input)
if malformed is not None:
logger.debug("Tool %s got malformed arguments; asking the model to re-issue", tool.name)
return malformed
return await invoke_tool(ctx, _coerce_arguments(raw_input, schema, nullish=nullish))
tool.on_invoke_tool = invoke
@@ -442,47 +436,8 @@ def _apply_shell_output_cap(parsed: dict[str, Any]) -> None:
)
_EXEC_YIELD_CAP_S = PTY_YIELD_TIME_MS_MAX // 1000
_EXEC_COMMAND_DESCRIPTION_NOTE = (
f"The wait for output is capped at {_EXEC_YIELD_CAP_S} s whatever yield_time_ms says. "
"A non-TTY command still running at the yield cannot be read later (write_stdin fails "
"on it and its output is lost): run anything longer in the background with its output "
"redirected to a file (`cmd > /tmp/out.log 2>&1 &`, then read the log), or start it with "
'tty=true and poll with write_stdin(session_id=<id>, chars="").'
)
_RUNNING_NON_TTY_NOTICE = (
"Note: this non-TTY process cannot be polled and its later output is lost. Rerun it in "
"the background (`cmd > /tmp/out.log 2>&1 &`, then read the log) or with tty=true."
)
_PROCESS_RUNNING_RE = re.compile(r"^Process running with session ID \d+$", re.MULTILINE)
_OUTPUT_HEADER_RE = re.compile(r"^Output:\n", re.MULTILINE)
def _annotate_running_non_tty(parsed: dict[str, Any], result: Any) -> Any:
if parsed.get("tty") or not isinstance(result, str):
return result
header = _OUTPUT_HEADER_RE.search(result)
head_end = len(result) if header is None else header.start()
if _PROCESS_RUNNING_RE.search(result, 0, head_end) is None:
return result
notice = _RUNNING_NON_TTY_NOTICE
requested = parsed.get("yield_time_ms")
if isinstance(requested, int | float) and requested > PTY_YIELD_TIME_MS_MAX:
notice = (
f"Note: yield_time_ms={int(requested)} was capped at {PTY_YIELD_TIME_MS_MAX} "
f"({_EXEC_YIELD_CAP_S} s). " + notice.removeprefix("Note: ")
)
if header is None:
return f"{result}\n{notice}"
return f"{result[: header.end()]}{notice}\n{result[header.end() :]}"
def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
invoke_tool = tool.on_invoke_tool
tool.description = f"{tool.description.rstrip()} {_EXEC_COMMAND_DESCRIPTION_NOTE}"
async def invoke(ctx: Any, raw_input: str) -> Any:
try:
@@ -495,7 +450,7 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
_apply_shell_output_cap(parsed)
raw_input = json.dumps(parsed)
try:
result = await invoke_tool(ctx, raw_input)
return await invoke_tool(ctx, raw_input)
except ValidationError as exc:
return _format_validation_error(tool.name, exc)
except InvalidManifestPathError as exc:
@@ -505,9 +460,6 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
"(or omitted to use the turn's cwd). "
f"Got: {rel!r}."
)
if isinstance(parsed, dict):
return _annotate_running_non_tty(parsed, result)
return result
tool.on_invoke_tool = invoke
return tool
+1 -2
View File
@@ -44,7 +44,7 @@ INTERACTIVE BEHAVIOR:
- If you end a turn on plain text and the nudge arrives, your words already reached the user. Do not repeat them: call wait_for_user.
- You can narrate while you work — plain text is shown to the user as you go. Narrating is free; wait_for_user is specifically the act of WAITING for the user, so do not call it just to give a status update.
- Respond naturally when the user asks questions or gives instructions.
- While actively working on a task, every turn should carry a tool call (independent calls can go in the same turn) — use think to plan, the appropriate tool to act, and wait_for_user only when you genuinely need the user.
- While actively working on a task, every turn should carry exactly one tool call — use think to plan, the appropriate tool to act, and wait_for_user only when you genuinely need the user.
- Never loop through think or other tools just to prepare, polish, confirm, or announce an answer. Once you know the answer, write it and call wait_for_user.
{% else %}
AUTONOMOUS BEHAVIOR:
@@ -55,7 +55,6 @@ AUTONOMOUS BEHAVIOR:
- There is no user attached to this run, so there is nobody to ask and nothing to yield to. If there is nothing left to execute: do NOT send filler/repetitive text — either call wait_for_agents (only if you are genuinely expecting another agent to message you) or finish your work (subagents: agent_finish; root: finish_scan)
- While the agent loop is running, almost every output MUST be a tool call. Do NOT send plain text messages; act via tools. If waiting on another agent, use wait_for_agents; when done, use agent_finish (subagents) or finish_scan (root)
- A text-only turn does nothing: it neither ends the run nor yields — it just wastes a turn and forces a retry. The lifecycle tools (``finish_scan`` for root, ``agent_finish`` for subagents) are the ONLY way to terminate, and the report flows through them. If you find yourself wanting to say "Done!" or "Scan complete" without a tool call, call the lifecycle tool instead.
- Read-only calls can go together in one turn: when several file reads or searches don't depend on each other's output, issue them as separate tool calls in the same turn rather than one per turn. Anything else — writing files, running commands that change state, sending requests, creating or messaging agents, reporting, or finishing — goes alone in its own turn.
{% endif %}
</communication_rules>
+2 -26
View File
@@ -6,8 +6,7 @@ import json
import logging
import os
from pathlib import Path
from types import UnionType
from typing import TYPE_CHECKING, Any, Union, get_args, get_origin
from typing import TYPE_CHECKING, Any
from pydantic import AliasChoices, BaseModel
@@ -127,36 +126,13 @@ def _read_json_overrides(path: Path) -> dict[str, dict[str, Any]]:
continue # env wins under some alias; skip the JSON file for this field
for alias in aliases:
if alias in env_block_upper:
sub_data[fname] = _decode_saved_value(finfo, env_block_upper[alias])
sub_data[fname] = env_block_upper[alias]
break
if sub_data:
nested[sub_name] = sub_data
return nested
def _decode_saved_value(finfo: FieldInfo, value: Any) -> Any:
"""Decode a JSON string saved for a dict field, as pydantic does for env vars.
``persist_current`` saves env values verbatim, so a dict setting such as
``LLM_EXTRA_HEADERS`` lands in the file as JSON text. Values that are
already objects, or aren't valid JSON, are left for validation to judge.
"""
if not isinstance(value, str) or not _is_dict_field(finfo):
return value
try:
return json.loads(value)
except ValueError:
return value
def _is_dict_field(finfo: FieldInfo) -> bool:
annotation = finfo.annotation
candidates = (
get_args(annotation) if get_origin(annotation) in (Union, UnionType) else (annotation,)
)
return any(c is dict or get_origin(c) is dict for c in candidates)
def _first_alias_value(aliases: list[str], source: Mapping[str, Any]) -> Any | None:
return next((source[alias] for alias in aliases if alias in source), None)
+2 -13
View File
@@ -41,7 +41,6 @@ from openai.types.shared import Reasoning
from strix.agents.prompt import CACHE_POINT
from strix.config import codex
from strix.config.loader import load_settings
from strix.config.tool_call_arguments import repair_input
from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input
from strix.config.tool_call_limits import TurnToolCallLimiter
from strix.llm import request_log
@@ -256,12 +255,6 @@ class _TurnGuardModel(Model):
Ids that collide with the history are rewritten before the turn is
recorded, and already-corrupted histories are repaired on the way out.
Tool-call arguments: a turn whose ``arguments`` are not valid JSON fails
that one call with a parse error, but the raw string is recorded and
strict providers then reject every request that replays it. Such
arguments are rewritten to a valid JSON object on the way out (see
:mod:`strix.config.tool_call_arguments`).
Tool-call volume: a degenerate response can queue hundreds of calls that
the run loop then honours one by one. Only the first
``LLM_MAX_TOOL_CALLS_PER_TURN`` calls of a response are kept.
@@ -320,7 +313,7 @@ class _TurnGuardModel(Model):
conversation_id: str | None,
prompt: ResponsePromptParam | None,
) -> ModelResponse:
sanitized = _sanitize_input(input)
sanitized = dedupe_input(input)
rewriter = TurnCallIdRewriter(sanitized)
system_instructions, sanitized = _split_cached_prefix(
system_instructions, sanitized, model_settings
@@ -356,7 +349,7 @@ class _TurnGuardModel(Model):
conversation_id: str | None,
prompt: ResponsePromptParam | None,
) -> AsyncIterator[TResponseStreamEvent]:
sanitized = _sanitize_input(input)
sanitized = dedupe_input(input)
rewriter = TurnCallIdRewriter(sanitized)
system_instructions, sanitized = _split_cached_prefix(
system_instructions, sanitized, model_settings
@@ -386,10 +379,6 @@ class _TurnGuardModel(Model):
self._log_dropped(limiter)
def _sanitize_input(model_input: str | list[TResponseInputItem]) -> str | list[Any]:
return repair_input(dedupe_input(model_input))
def _split_cached_prefix(
system_instructions: str | None,
model_input: str | list[Any],
+1 -11
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from typing import Any, Literal
from typing import Literal
from pydantic import AliasChoices, Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -55,11 +55,6 @@ class LlmSettings(BaseSettings):
alias="LLM_EXTRA_HEADERS",
repr=False,
)
extra_body: dict[str, Any] | None = Field(
default=None,
alias="LLM_EXTRA_BODY",
repr=False,
)
reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
force_required_tool_choice: bool = Field(
default=False,
@@ -115,11 +110,6 @@ class DedupeSettings(BaseSettings):
alias="DEDUPE_LLM_EXTRA_HEADERS",
repr=False,
)
extra_body: dict[str, Any] | None = Field(
default=None,
alias="DEDUPE_LLM_EXTRA_BODY",
repr=False,
)
@field_validator("reasoning_effort", mode="before")
@classmethod
-100
View File
@@ -1,100 +0,0 @@
"""Keep replayed tool-call arguments valid JSON.
A model occasionally emits a tool call whose ``arguments`` string is not a
JSON object (truncated, unbalanced, empty). The call itself fails safely: the
tool reports the parse error and the run continues. But the raw string is
recorded in the session as-is, and strict OpenAI-compatible servers (vLLM,
SGLang, ...) validate every assistant tool call in the request, so from then
on each turn is rejected with ``Assistant tool call function.arguments must
be valid JSON`` and the agent can never recover. Rewriting the replayed
arguments to a JSON object that carries the original text keeps the history
valid while the model still sees what it sent.
"""
from __future__ import annotations
import json
from typing import Any
from openai.types.responses import ResponseFunctionToolCall
MALFORMED_ARGUMENTS_KEY = "malformed_arguments"
def _reject_constant(name: str) -> Any:
raise ValueError(f"non-standard JSON constant {name}")
def _loads_strict(text: str) -> Any:
try:
return json.loads(text, parse_constant=_reject_constant)
except RecursionError as exc:
raise ValueError("arguments are nested too deeply to parse") from exc
def describe_malformed_arguments(tool_name: str, arguments: str) -> str | None:
"""Return a model-facing recovery message if ``arguments`` is not a JSON object.
A tool call whose arguments do not parse is almost always one the stream cut
off (the server flushed a prefix of the JSON, or the client closed early),
so instead of the SDK's generic parse-error result the model is told the
call never ran and must be re-issued whole.
"""
if not arguments.strip():
return None
try:
parsed = _loads_strict(arguments)
except ValueError as exc:
detail = str(exc)
else:
if isinstance(parsed, dict):
return None
detail = f"expected a JSON object, got {type(parsed).__name__}"
return (
f"{tool_name}: the tool call was not executed because its arguments were "
f"truncated or otherwise not valid JSON ({detail}). The response was likely "
"cut off mid-call. Re-issue the call with complete, valid JSON arguments."
)
def repair_arguments(arguments: object) -> str | None:
"""Return replacement arguments a strict server accepts, or ``None`` if already valid."""
if not isinstance(arguments, str) or not arguments.strip():
return "{}"
try:
parsed = _loads_strict(arguments)
except ValueError:
parsed = None
if isinstance(parsed, dict):
return None
return json.dumps({MALFORMED_ARGUMENTS_KEY: arguments}, ensure_ascii=False)
def repair_history_arguments(items: list[Any]) -> tuple[list[Any], bool]:
"""Rewrite function calls in a conversation history whose arguments are not a JSON object."""
rebuilt: list[Any] = []
changed = False
for item in items:
if isinstance(item, dict):
if item.get("type") == "function_call":
repaired = repair_arguments(item.get("arguments"))
if repaired is not None:
item = {**item, "arguments": repaired} # noqa: PLW2901
changed = True
elif isinstance(item, ResponseFunctionToolCall):
repaired = repair_arguments(item.arguments)
if repaired is not None:
item = item.model_copy(update={"arguments": repaired}) # noqa: PLW2901
changed = True
rebuilt.append(item)
return rebuilt, changed
def repair_input(model_input: str | list[Any]) -> str | list[Any]:
if isinstance(model_input, str):
return model_input
rebuilt, changed = repair_history_arguments(model_input)
return rebuilt if changed else model_input
+3 -19
View File
@@ -75,7 +75,6 @@ class AgentCoordinator:
self._budget_stopped = False
self._reserve_stopped = False
self._budget_paused = False
self._operator_paused = False
self._resume_epoch = 0
self._budget_policy: BudgetPolicy = "stop"
self._extend_budget: Callable[[], None] | None = None
@@ -106,10 +105,6 @@ class AgentCoordinator:
def budget_paused(self) -> bool:
return self._budget_paused
@property
def operator_paused(self) -> bool:
return self._operator_paused
@property
def resume_epoch(self) -> int:
"""Bumped by every ``resume_budget``; an agent parks against the value it read."""
@@ -155,7 +150,6 @@ class AgentCoordinator:
"""
async with self._lock:
self._budget_paused = True
self._operator_paused = True
logger.info("scan paused by the operator")
await self._maybe_snapshot()
@@ -171,7 +165,6 @@ class AgentCoordinator:
self._set_budget_limit(max_budget_usd)
async with self._lock:
self._budget_paused = False
self._operator_paused = False
self._resume_epoch += 1
woken = [aid for aid, status in self.statuses.items() if status == "budget_paused"]
for aid in woken:
@@ -239,7 +232,7 @@ class AgentCoordinator:
self._budget_stopped = budget_stopped
self._reserve_stopped = reserve_stopped
if not budget_paused:
self._budget_paused = self._operator_paused
self._budget_paused = False
for aid, status in self.statuses.items():
if status == "budget_paused":
self.statuses[aid] = "waiting"
@@ -431,12 +424,7 @@ class AgentCoordinator:
unknown, or it is terminal and its loop does not park for wake-ups.
"""
from_user = message.get("from") == "user"
if (
from_user
and self._budget_paused
and not self._operator_paused
and self._budget_policy != "pause"
):
if from_user and self._budget_paused and self._budget_policy != "pause":
await self.resume_from_budget_pause(exclude=target_agent_id)
async with self._lock:
if target_agent_id not in self.statuses:
@@ -459,9 +447,7 @@ class AgentCoordinator:
self.recovery_counts.pop(target_agent_id, None)
self.idle_resume_counts.pop(target_agent_id, None)
self._parent_notified.discard(target_agent_id)
held = self._operator_paused and self.statuses[target_agent_id] == "budget_paused"
if not held:
self.statuses[target_agent_id] = "waiting"
self.statuses[target_agent_id] = "waiting"
runtime.wake.set()
stream = runtime.stream
interrupt_on_message = runtime.interrupt_on_message
@@ -654,7 +640,6 @@ class AgentCoordinator:
"budget_stopped": self._budget_stopped,
"reserve_stopped": self._reserve_stopped,
"budget_paused": self._budget_paused,
"operator_paused": self._operator_paused,
}
async def restore(self, snap: dict[str, Any]) -> None:
@@ -677,7 +662,6 @@ class AgentCoordinator:
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))
self._operator_paused = bool(snap.get("operator_paused", False))
for aid in self.statuses:
self.runtimes.setdefault(aid, AgentRuntime())
+4 -11
View File
@@ -24,7 +24,6 @@ from openai import (
from strix.config import codex
from strix.core.hooks import (
WRAP_UP_AFTER_TURNS_KEY,
BudgetExceededError,
BudgetPausedError,
SubagentBudgetReservedError,
@@ -40,7 +39,6 @@ from strix.core.sessions import (
)
from strix.llm import request_log
from strix.llm.compaction import is_context_overflow, maybe_compact
from strix.skills import skill_wrap_up_after_turns
if TYPE_CHECKING:
@@ -436,7 +434,6 @@ async def spawn_child_agent(
task=task,
parent_history=parent_history,
),
wrap_up_after_turns=skill_wrap_up_after_turns(skills),
event_sink=event_sink,
hooks=hooks,
)
@@ -515,7 +512,6 @@ async def respawn_subagents(
task=str(md.get("task", "")),
initial_input=[],
start_parked=start_parked,
wrap_up_after_turns=skill_wrap_up_after_turns(child_skills),
event_sink=event_sink,
hooks=hooks,
)
@@ -603,7 +599,7 @@ async def _run_until_lifecycle(
hooks=hooks,
)
except BudgetPausedError as exc:
if coordinator.budget_policy != "pause" and not exc.operator_pause:
if coordinator.budget_policy != "pause":
raise
# The agent parked right before an LLM call; everything up to that
# point is already in its session. Once resumed, the same call goes
@@ -616,7 +612,6 @@ async def _run_until_lifecycle(
# Stopped while parked (operator stop or a parent's stop_agent).
await coordinator.reset_recovery(agent_id)
return result
await coordinator.consume_pending(agent_id)
if session is not None:
input_data = []
continue
@@ -836,7 +831,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
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)
if coordinator.budget_policy == "pause" or exc.operator_pause:
if coordinator.budget_policy == "pause":
await coordinator.park_for_budget(agent_id)
else:
await coordinator.pause_for_budget(agent_id)
@@ -1008,7 +1003,7 @@ async def _append_tool_required_message(
message = (
"Your previous message ended a turn without a tool call. Plain text never ends "
"execution and never hands control to the user: it is shown to the user, and the "
"run continues. Continue immediately and call a tool. "
"run continues. Continue immediately and call exactly one tool. "
"If you have nothing to do until the user replies, call wait_for_user; your "
"text already reached them, so do not repeat it. "
"If you are blocked waiting for another agent, call wait_for_agents. "
@@ -1020,7 +1015,7 @@ async def _append_tool_required_message(
message = (
"Your previous response ended the autonomous run without a lifecycle tool "
"call. That is invalid in non-interactive mode; plain text final answers are "
"ignored. Continue immediately and call a tool. "
"ignored. Continue immediately and call exactly one tool. "
f"If your work is complete, call {finish_tool}. "
"If you are blocked waiting for another agent, call wait_for_agents. "
"Otherwise use the appropriate execution or planning tool. "
@@ -1167,7 +1162,6 @@ async def _start_child_runner(
task: str,
initial_input: Any,
start_parked: bool = False,
wrap_up_after_turns: int | None = None,
event_sink: StreamEventSink | None = None,
hooks: RunHooks[dict[str, Any]] | None = None,
) -> None:
@@ -1179,7 +1173,6 @@ async def _start_child_runner(
child_ctx["agent_id"] = child_id
child_ctx["parent_id"] = parent_id
child_ctx["task"] = task
child_ctx[WRAP_UP_AFTER_TURNS_KEY] = wrap_up_after_turns
async def _child_loop() -> None:
# A budget stop is a clean scan-wide shutdown, not a child failure: the
+1 -39
View File
@@ -22,9 +22,6 @@ logger = logging.getLogger(__name__)
LLM_TURN_KEY = "llm_turn"
# Set on a sub-agent's context from its skills (see ``skill_wrap_up_after_turns``).
WRAP_UP_AFTER_TURNS_KEY = "wrap_up_after_turns"
_WRAP_UP_REPEAT_TURNS = 5
# ``BudgetPolicy`` decides what happens when the accumulated LLM cost reaches
# ``max_budget_usd``.
@@ -38,7 +35,6 @@ _WRAP_UP_REPEAT_TURNS = 5
# same call when the operator raises the limit or resumes.
__all__ = [
"LLM_TURN_KEY",
"WRAP_UP_AFTER_TURNS_KEY",
"BudgetExceededError",
"BudgetPausedError",
"BudgetPolicy",
@@ -69,12 +65,9 @@ class BudgetPausedError(RuntimeError):
decided to park; the agent waits for a resume newer than that.
"""
def __init__(
self, message: str, *, resume_epoch: int = 0, operator_pause: bool = False
) -> None:
def __init__(self, message: str, *, resume_epoch: int = 0) -> None:
super().__init__(message)
self.resume_epoch = resume_epoch
self.operator_pause = operator_pause
def _validate_budget(max_budget_usd: float | None) -> None:
@@ -205,26 +198,15 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
system_prompt: str | None, # noqa: ARG002
input_items: list[TResponseInputItem],
) -> None:
self._pause_if_operator_paused(context)
if self._budget_policy == "pause":
self._pause_if_limited(context)
context.context[LLM_TURN_KEY] = int(context.context.get(LLM_TURN_KEY, 0)) + 1
try:
self._maybe_warn_turns(context, input_items)
self._maybe_nudge_wrap_up(context, input_items)
self._maybe_warn_budget(context, input_items)
except Exception:
logger.exception("budget/turn warning injection failed")
def _pause_if_operator_paused(self, context: RunContextWrapper[dict[str, Any]]) -> None:
coordinator = coordinator_from_context(context.context)
if coordinator is not None and coordinator.operator_paused:
raise BudgetPausedError(
"scan paused; waiting for the operator to resume",
resume_epoch=coordinator.resume_epoch,
operator_pause=True,
)
def _pause_if_limited(self, context: RunContextWrapper[dict[str, Any]]) -> None:
"""Park the agent before a paid call when the scan is at its limit or paused.
@@ -275,26 +257,6 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
)
input_items.append({"role": "user", "content": content})
def _maybe_nudge_wrap_up(
self,
context: RunContextWrapper[dict[str, Any]],
input_items: list[TResponseInputItem],
) -> None:
"""A soft nudge, never a stop: at the turn a skill asks for, then every few turns."""
after = context.context.get(WRAP_UP_AFTER_TURNS_KEY)
requests = getattr(getattr(context, "usage", None), "requests", None)
if not isinstance(after, int) or not isinstance(requests, int):
return
turns_used = requests + 1
if turns_used < after or (turns_used - after) % _WRAP_UP_REPEAT_TURNS:
return
content = (
f"[NOTICE] You have used {turns_used} turns. Wrap up soon: take only the steps "
"you still need to answer your task, then call agent_finish with what you have, "
"listing anything unresolved in open_items instead of tracing it further."
)
input_items.append({"role": "user", "content": content})
def _maybe_warn_budget(
self,
context: RunContextWrapper[dict[str, Any]],
+5 -13
View File
@@ -250,26 +250,22 @@ def make_model_settings(
request_timeout: float | None = None,
prompt_cache: bool = True,
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, Any] | None = None,
has_tools: bool = True,
) -> ModelSettings:
headers = _request_headers(model_name, extra_headers)
model_settings = ModelSettings(
parallel_tool_calls=True if has_tools else None,
parallel_tool_calls=False if has_tools else None,
retry=DEFAULT_MODEL_RETRY,
include_usage=True,
extra_args=request_timeout_extra_args(request_timeout),
extra_headers=headers,
extra_body=dict(extra_body) if extra_body else None,
)
if (
reasoning_effort is not None
and reasoning_effort != "none"
and model_supports_reasoning(model_name)
):
model_settings = model_settings.resolve(
_reasoning_settings(reasoning_effort, extra_body),
)
model_settings = model_settings.resolve(_reasoning_settings(reasoning_effort))
if force_required_tool_choice and _accepts_required_tool_choice(model_name):
model_settings = model_settings.resolve(ModelSettings(tool_choice="required"))
@@ -294,9 +290,7 @@ def _request_headers(
return headers or None
def _reasoning_settings(
effort: ReasoningEffort, extra_body: dict[str, Any] | None = None
) -> ModelSettings:
def _reasoning_settings(effort: ReasoningEffort) -> ModelSettings:
"""``max`` is not in the OpenAI SDK's ``Reasoning.effort`` enum, so send it as
a raw body field instead — also keeping it clear of LiteLLM's DeepSeek mapping,
which collapses every ``reasoning_effort`` level to plain thinking-enabled.
@@ -304,13 +298,11 @@ def _reasoning_settings(
It goes in ``extra_body``, the field every model implementation forwards as the
request's ``extra_body``; the same value under ``extra_args`` collides with that
keyword and raises before a request is ever sent. ``resolve`` replaces
``extra_body`` wholesale, so the user's ``LLM_EXTRA_BODY`` fields are carried
over, with ``reasoning_effort`` taking precedence.
keyword and raises before a request is ever sent.
"""
if effort != "max":
return ModelSettings(reasoning=Reasoning(effort=effort))
return ModelSettings(extra_body={**(extra_body or {}), "reasoning_effort": "max"})
return ModelSettings(extra_body={"reasoning_effort": "max"})
def _prompt_cache_extra_args(model_name: str) -> dict[str, Any] | None:
+2 -3
View File
@@ -8,7 +8,7 @@ import io
import json
import logging
import uuid
from collections.abc import Awaitable, Callable
from collections.abc import Callable
from pathlib import Path
from typing import TYPE_CHECKING, Any
@@ -203,7 +203,7 @@ async def run_strix_scan(
scan_id: str | None = None,
image: str,
local_sources: list[dict[str, Any]] | None = None,
extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None = None,
extra_files: list[dict[str, Any]] | None = None,
coordinator: AgentCoordinator | None = None,
interactive: bool = False,
max_turns: int = DEFAULT_MAX_TURNS,
@@ -401,7 +401,6 @@ async def run_strix_scan(
request_timeout=settings.llm.timeout,
prompt_cache=settings.llm.prompt_cache,
extra_headers=settings.llm.extra_headers,
extra_body=settings.llm.extra_body,
)
run_config = RunConfig(
model=resolved_model,
+2 -3
View File
@@ -164,13 +164,12 @@ async def warm_up_llm() -> None:
raw_model = dedupe_model
deduper = resolve_dedupe_model(settings.dedupe, dedupe_model)
# A dedicated dedupe model may route to another provider, which must
# never receive the main endpoint's headers or body fields; it has
# its own DEDUPE_LLM_EXTRA_HEADERS and DEDUPE_LLM_EXTRA_BODY.
# never receive the main endpoint's headers; it has its own
# DEDUPE_LLM_EXTRA_HEADERS.
await preflight_request(
deduper,
model_name=dedupe_model,
extra_headers=settings.dedupe.extra_headers,
extra_body=settings.dedupe.extra_body,
timeout=llm.preflight_timeout,
api_base_setting="DEDUPE_LLM_API_BASE",
)
-3
View File
@@ -116,7 +116,6 @@ async def preflight_model_connection(
model,
model_name=model_name,
extra_headers=resolved_settings.llm.extra_headers,
extra_body=resolved_settings.llm.extra_body,
timeout=resolved_settings.llm.preflight_timeout,
api_base_setting="LLM_API_BASE",
)
@@ -127,7 +126,6 @@ async def preflight_request(
*,
model_name: str,
extra_headers: dict[str, str] | None,
extra_body: dict[str, Any] | None,
timeout: int,
api_base_setting: str,
) -> None:
@@ -146,7 +144,6 @@ async def preflight_request(
request_timeout=timeout,
prompt_cache=False,
extra_headers=extra_headers,
extra_body=extra_body,
has_tools=False,
)
try:
-1
View File
@@ -307,7 +307,6 @@ async def _summarize(model: str, prompt: str, max_tokens: int) -> str | None:
request_timeout=llm.timeout,
prompt_cache=False,
extra_headers=llm.extra_headers,
extra_body=llm.extra_body,
has_tools=False,
).resolve(ModelSettings(max_tokens=max_tokens))
try:
+4 -6
View File
@@ -39,13 +39,11 @@ def _dedupe_model_settings(
model_name=model_name,
force_required_tool_choice=False,
request_timeout=request_timeout,
# The main model's headers and body fields apply only when dedupe falls
# back to the main model; a dedicated dedupe model may route to another
# provider, which must never receive the main endpoint's credentials or
# identifiers. A dedicated model gets its own DEDUPE_LLM_EXTRA_HEADERS
# and DEDUPE_LLM_EXTRA_BODY instead.
# The main model's headers apply only when dedupe falls back to the main
# model; a dedicated dedupe model may route to another provider, which
# must never receive the main endpoint's credentials. A dedicated model
# gets its own DEDUPE_LLM_EXTRA_HEADERS instead.
extra_headers=dedupe.extra_headers if dedupe.model else llm.extra_headers,
extra_body=dedupe.extra_body if dedupe.model else llm.extra_body,
has_tools=False,
)
-255
View File
@@ -1,255 +0,0 @@
from __future__ import annotations
import asyncio
import contextlib
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any
from agents.sandbox.session.manager import Instrumentation
from agents.sandbox.session.sandbox_session import SandboxSession
from agents.sandbox.session.sandbox_session_state import SandboxSessionState
from agents.sandbox.snapshot import NoopSnapshot
if TYPE_CHECKING:
import io
from collections.abc import Sequence
from agents.sandbox.entries import BaseEntry
from agents.sandbox.manifest import Manifest
from agents.sandbox.materialization import MaterializedFile
from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
from agents.sandbox.types import ExecResult, ExposedPortEndpoint, User
logger = logging.getLogger(__name__)
class _PendingSessionState(SandboxSessionState):
type: str = "pending"
class DeferredSandboxSession(SandboxSession):
def __init__(
self,
bring_up: asyncio.Task[tuple[Any, BaseSandboxSession]],
*,
manifest: Manifest,
exposed_ports: tuple[int, ...] = (),
) -> None:
self._bring_up = bring_up
self._pending_state: SandboxSessionState = _PendingSessionState(
snapshot=NoopSnapshot(id="pending"),
manifest=manifest,
exposed_ports=exposed_ports,
)
self._pending_limits: Any = None
self._pending_archive_limits: Any = None
self._resolved: BaseSandboxSession | None = None
self._client: Any = None
self._instrumentation: Instrumentation = Instrumentation()
self._seq = 0
@property
def pending(self) -> bool:
return self._resolved is None
@property
def bring_up(self) -> asyncio.Task[tuple[Any, BaseSandboxSession]]:
return self._bring_up
async def resolve(self) -> BaseSandboxSession:
if self._resolved is not None:
return self._resolved
client, session = await asyncio.shield(self._bring_up)
self._adopt(client, session)
return session
def _adopt(self, client: Any, session: BaseSandboxSession) -> None:
if self._resolved is not None:
return
self._client = client
self._resolved = session
self._inner = session
if self._pending_limits is not None:
session._set_concurrency_limits(self._pending_limits)
if self._pending_archive_limits is not None:
session._set_archive_limits(self._pending_archive_limits)
if self._pending_state.manifest != session.state.manifest:
session.state = session.state.model_copy(
update={"manifest": self._pending_state.manifest}
)
@property
def client(self) -> Any:
return self._client
@property
def state(self) -> SandboxSessionState:
if self._resolved is None:
return self._pending_state
return self._resolved.state
@state.setter
def state(self, value: SandboxSessionState) -> None:
if self._resolved is None:
self._pending_state = value
else:
self._resolved.state = value
def _set_concurrency_limits(self, limits: Any) -> None:
if self._resolved is None:
self._pending_limits = limits
else:
self._resolved._set_concurrency_limits(limits)
def _set_archive_limits(self, limits: Any) -> None:
if self._resolved is None:
self._pending_archive_limits = limits
else:
self._resolved._set_archive_limits(limits)
def _manifest_base_dir(self) -> Path:
if self._resolved is None:
return Path.cwd()
return self._resolved._manifest_base_dir()
def normalize_path(self, path: Path | str, *, for_write: bool = False) -> Path:
if self._resolved is None:
return Path(path)
return self._resolved.normalize_path(path, for_write=for_write)
def register_persist_workspace_skip_path(self, path: Path | str) -> Path:
if self._resolved is None:
return Path(path)
return self._resolved.register_persist_workspace_skip_path(path)
def supports_pty(self) -> bool:
return self._resolved.supports_pty() if self._resolved is not None else False
async def running(self) -> bool:
if self._resolved is None:
if not self._bring_up.done():
return True
return not self._bring_up.cancelled() and self._bring_up.exception() is None
return await self._resolved.running()
async def start(self) -> None:
if self._resolved is None:
return
await self._resolved.start()
async def stop(self) -> None:
if self._resolved is None:
return
await self._resolved.stop()
async def shutdown(self) -> None:
if self._resolved is None:
return
await self._resolved.shutdown()
async def aclose(self) -> None:
if self._resolved is None:
return
await self._resolved.aclose()
async def _validate_manifest_application(self, *, only_ephemeral: bool = False) -> None:
session = await self.resolve()
await session._validate_manifest_application(only_ephemeral=only_ephemeral)
async def apply_manifest(self, *, only_ephemeral: bool = False) -> Any:
session = await self.resolve()
return await session.apply_manifest(only_ephemeral=only_ephemeral)
async def _apply_entry_batch(
self,
entries: Sequence[tuple[Path, BaseEntry]],
*,
base_dir: Path,
) -> list[MaterializedFile]:
del base_dir
session = await self.resolve()
return await session._apply_entry_batch(entries, base_dir=session._manifest_base_dir())
async def exec(self, *command: str | Path, **kwargs: Any) -> ExecResult:
await self.resolve()
return await super().exec(*command, **kwargs)
async def ls(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().ls(*args, **kwargs)
async def rm(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().rm(*args, **kwargs)
async def mkdir(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().mkdir(*args, **kwargs)
async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase:
await self.resolve()
return await super().read(path, user=user)
async def write(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().write(*args, **kwargs)
async def resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
await self.resolve()
return await super().resolve_exposed_port(port)
async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
session = await self.resolve()
return await session._resolve_exposed_port(port)
async def pty_exec_start(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().pty_exec_start(*args, **kwargs)
async def pty_write_stdin(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().pty_write_stdin(*args, **kwargs)
async def pty_terminate_all(self) -> None:
if self._resolved is None:
return
await super().pty_terminate_all()
async def persist_workspace(self) -> io.IOBase:
await self.resolve()
return await super().persist_workspace()
async def hydrate_workspace(self, data: io.IOBase) -> None:
await self.resolve()
await super().hydrate_workspace(data)
async def teardown(self) -> None:
if self._resolved is None:
self._bring_up.cancel()
with contextlib.suppress(BaseException):
client, session = await self._bring_up
self._adopt(client, session)
if self._resolved is None:
return
await self._client.delete(self._resolved)
class DeferredSandboxClient:
def __init__(self, session: DeferredSandboxSession) -> None:
self._session = session
def __getattr__(self, name: str) -> Any:
client = self._session.client
if client is None:
raise AttributeError(f"sandbox client unavailable before bring-up finishes ({name})")
return getattr(client, name)
async def delete(self, session: Any) -> None:
if isinstance(session, DeferredSandboxSession):
await session.teardown()
return
real = await self._session.resolve()
await self._session.client.delete(real if session is self._session else session)
+32 -93
View File
@@ -3,8 +3,6 @@
from __future__ import annotations
import asyncio
import contextlib
import inspect
import io
import logging
import os
@@ -21,12 +19,9 @@ from strix.config import load_settings
from strix.runtime.backends import backend_supports_bind_mounts, get_backend
from strix.runtime.caido_bootstrap import bootstrap_caido
from strix.runtime.caido_handle import CaidoBootstrapHandle
from strix.runtime.deferred_session import DeferredSandboxClient, DeferredSandboxSession
if TYPE_CHECKING:
from collections.abc import Awaitable
from agents.sandbox.session import BaseSandboxSession
from strix.runtime.status import StatusSink
@@ -260,22 +255,13 @@ def _gitdir_from_pointer(git_file: Path) -> Path | None:
return None
async def _resolve_extra_file_archive(
extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None,
local_sources: list[dict[str, Any]],
) -> bytes | None:
files = await extra_files if inspect.isawaitable(extra_files) else extra_files
return build_extra_file_archive(files, local_sources) if files else None
async def create_or_reuse(
scan_id: str,
*,
image: str,
local_sources: list[dict[str, Any]],
extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None = None,
extra_files: list[dict[str, Any]] | None = None,
status_sink: StatusSink | None = None,
defer: bool = False,
) -> dict[str, Any]:
"""Return the existing session bundle for ``scan_id`` or create a new one.
@@ -285,9 +271,7 @@ async def create_or_reuse(
Each ``extra_files`` entry (``{"workspace_path": "/workspace/<rel>",
"content": bytes | str}``) lands as a regular, agent-writable file at its
``workspace_path`` on every backend: the files are uploaded as one archive
and unpacked inside the sandbox right after bring-up. ``extra_files`` may
also be an awaitable resolving to that list: provisioning starts at once and
the files are awaited only when the sandbox is ready to receive them.
and unpacked inside the sandbox right after bring-up.
"""
def report(phase: str) -> None:
@@ -308,6 +292,9 @@ async def create_or_reuse(
else:
bind_mounts = []
entries = build_manifest_entries(local_sources)
extra_file_archive = (
build_extra_file_archive(extra_files, local_sources) if extra_files else None
)
# Caido runs as an in-container sidecar; HTTP(S) traffic from any
# process started via ``session.exec`` (the SDK's Shell tool, etc.)
@@ -336,69 +323,41 @@ async def create_or_reuse(
backend_name,
image,
)
report("Starting sandbox container")
client, session = await backend(
image=image,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
)
async def _bring_up() -> tuple[Any, Any]:
report("Starting sandbox container")
client, session = await _create_or_discard(
backend(
image=image,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
)
)
if extra_file_archive is not None:
report("Placing workspace files")
try:
extra_file_archive = await _resolve_extra_file_archive(extra_files, local_sources)
if extra_file_archive is not None:
report("Placing workspace files")
await stage_extra_files(session, extra_file_archive)
await stage_extra_files(session, extra_file_archive)
except BaseException:
await _discard_session(client, session)
raise
return client, session
async def _caido_url(session: Any) -> str:
report("Setting up the proxy")
caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
scheme = "https" if caido_endpoint.tls else "http"
host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
logger.debug("Caido host endpoint resolved: %s", host_caido_url)
return host_caido_url
report("Setting up the proxy")
caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
scheme = "https" if caido_endpoint.tls else "http"
host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
logger.debug("Caido host endpoint resolved: %s", host_caido_url)
def _caido(session: Any, host_caido_url: str) -> Any:
return bootstrap_caido(
session,
host_url=host_caido_url,
container_url=container_caido_url,
)
if defer:
bring_up = asyncio.create_task(_bring_up(), name=f"sandbox-bring-up-{scan_id}")
deferred = DeferredSandboxSession(
bring_up,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
)
async def _caido_when_up() -> Any:
session = await deferred.resolve()
return await _caido(session, await _caido_url(session))
bundle = {
"client": DeferredSandboxClient(deferred),
"session": deferred,
"caido_client": CaidoBootstrapHandle(
asyncio.create_task(_caido_when_up(), name=f"caido-bootstrap-{scan_id}")
),
}
_SESSION_CACHE[scan_id] = bundle
logger.info("Sandbox session for scan %s deferred and cached", scan_id)
return bundle
client, session = await _bring_up()
host_caido_url = await _caido_url(session)
# The Caido login + project setup polls the guest for a couple of seconds
# and nothing needs the client before the first proxy tool call, so it
# runs concurrently with the rest of scan start; consumers resolve the
# handle at first use (see CaidoBootstrapHandle).
caido_client = CaidoBootstrapHandle(
asyncio.create_task(_caido(session, host_caido_url), name=f"caido-bootstrap-{scan_id}")
asyncio.create_task(
bootstrap_caido(
session,
host_url=host_caido_url,
container_url=container_caido_url,
),
name=f"caido-bootstrap-{scan_id}",
)
)
bundle = {
@@ -411,26 +370,6 @@ async def create_or_reuse(
return bundle
_DISCARD_TASKS: set[asyncio.Task[None]] = set()
async def _create_or_discard(create: Awaitable[tuple[Any, Any]]) -> tuple[Any, Any]:
pending = asyncio.ensure_future(create)
try:
return await asyncio.shield(pending)
except asyncio.CancelledError:
discard = asyncio.create_task(_discard_when_created(pending))
_DISCARD_TASKS.add(discard)
discard.add_done_callback(_DISCARD_TASKS.discard)
raise
async def _discard_when_created(create: asyncio.Future[tuple[Any, Any]]) -> None:
with contextlib.suppress(BaseException):
client, session = await create
await _discard_session(client, session)
async def _discard_session(client: Any, session: Any) -> None:
"""Best-effort teardown of a session that never made it into the cache."""
try:
-13
View File
@@ -270,19 +270,6 @@ def _candidate_skill_files(skill_name: str) -> list[Path]:
return _bare_skill_files(skill_name)
def skill_wrap_up_after_turns(skill_names: list[str]) -> int | None:
"""Smallest ``wrap_up_after_turns`` set in these skills' frontmatter, if any."""
turns = []
for skill_name in skill_names:
candidates = _candidate_skill_files(skill_name)
if len(candidates) != 1:
continue
value = _read_skill_metadata(candidates[0]).get("wrap_up_after_turns", "")
if value.isdigit() and int(value) > 0:
turns.append(int(value))
return min(turns, default=None)
def load_skills(skill_names: list[str], *, supports_images: bool = True) -> dict[str, str]:
"""Load skill markdown bodies (frontmatter stripped) by name.
+4
View File
@@ -1876,6 +1876,10 @@ def _build_dependency_evidence(
evidence += f"\n\n**Usage analysis:** {label}."
if reachability_evidence and reachability_evidence.strip():
evidence += f" {reachability_evidence.strip()}"
evidence += (
" This is a prioritization signal from static analysis, not a"
" proof of exploitability or of safety."
)
return evidence
-88
View File
@@ -80,94 +80,6 @@ async def test_wrap_exec_command_preserves_explicit_shell(shell: str) -> None:
assert json.loads(captured["raw_input"])["shell"] == shell
def _running_exec_tool(result: str) -> FunctionTool:
async def invoke(_ctx: Any, _raw_input: str) -> str:
return result
return FunctionTool(
name="exec_command",
description="Runs a command.",
params_json_schema={"type": "object", "properties": {}},
on_invoke_tool=invoke,
)
_RUNNING_RESULT = (
"Chunk ID: abc123\nWall time: 30.0012 seconds\nProcess running with session ID 7\nOutput:\n"
)
def test_wrap_exec_command_describes_the_yield_cap() -> None:
wrapped = factory._wrap_exec_command(_running_exec_tool("ok"))
assert wrapped.description.startswith("Runs a command. The wait for output is capped at 30 s")
assert "cannot be read later" in wrapped.description
assert "tty=true" in wrapped.description
@pytest.mark.asyncio
async def test_wrap_exec_command_flags_a_running_non_tty_process() -> None:
wrapped = factory._wrap_exec_command(_running_exec_tool(_RUNNING_RESULT))
result = await wrapped.on_invoke_tool(
cast("Any", None), json.dumps({"cmd": "sleep 60; curl x", "yield_time_ms": 90000})
)
assert result == (
"Chunk ID: abc123\nWall time: 30.0012 seconds\n"
"Process running with session ID 7\nOutput:\n"
"Note: yield_time_ms=90000 was capped at 30000 (30 s). this non-TTY process cannot be "
"polled and its later output is lost. Rerun it in the background "
"(`cmd > /tmp/out.log 2>&1 &`, then read the log) or with tty=true.\n"
)
@pytest.mark.asyncio
async def test_wrap_exec_command_flags_a_running_non_tty_process_within_the_cap() -> None:
wrapped = factory._wrap_exec_command(_running_exec_tool(_RUNNING_RESULT))
result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "sleep 60"}))
assert "Note: this non-TTY process cannot be polled" in result
assert "was capped" not in result
@pytest.mark.asyncio
async def test_wrap_exec_command_leaves_a_running_tty_process_alone() -> None:
wrapped = factory._wrap_exec_command(_running_exec_tool(_RUNNING_RESULT))
result = await wrapped.on_invoke_tool(
cast("Any", None), json.dumps({"cmd": "python3 -m http.server", "tty": True})
)
assert result == _RUNNING_RESULT
@pytest.mark.asyncio
async def test_wrap_exec_command_leaves_an_exited_process_alone() -> None:
exited = "Chunk ID: abc123\nWall time: 0.1 seconds\nProcess exited with code 0\nOutput:\nhi"
wrapped = factory._wrap_exec_command(_running_exec_tool(exited))
result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "echo hi"}))
assert result == exited
@pytest.mark.asyncio
async def test_wrap_exec_command_ignores_the_running_marker_inside_command_output() -> None:
exited = (
"Chunk ID: abc123\nWall time: 0.2 seconds\nProcess exited with code 0\nOutput:\n"
"Process running with session ID 7\nOutput:\n"
)
wrapped = factory._wrap_exec_command(_running_exec_tool(exited))
result = await wrapped.on_invoke_tool(
cast("Any", None), json.dumps({"cmd": "cat transcript.txt"})
)
assert result == exited
@pytest.mark.asyncio
async def test_responses_filesystem_custom_tool_output_is_bounded() -> None:
async def invoke(_ctx: Any, _inp: str) -> str:
+3 -21
View File
@@ -38,26 +38,11 @@ async def _roundtrip(
_STRING = {"todos": {"type": "string"}}
@pytest.mark.asyncio
async def test_truncated_arguments_short_circuit_with_a_reissue_message() -> None:
captured: dict[str, str] = {}
wrapped = factory._with_coerced_arguments(_capturing_tool(captured, _STRING))
result = await wrapped.on_invoke_tool(cast("Any", None), '{"todos": "a, b')
assert "not executed" in result
assert "Re-issue the call" in result
assert "raw_input" not in captured
_ARRAY = {"tags": {"type": "array", "items": {"type": "string"}}}
_NULLABLE_ARRAY = {
"tags": {"anyOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}]}
}
_OBJECT = {"modifications": {"type": "object"}}
OVER_NESTED = '{"a":' + "[" * 100_000
@pytest.mark.asyncio
@@ -151,15 +136,12 @@ async def test_unknown_and_null_arguments_are_untouched() -> None:
@pytest.mark.asyncio
@pytest.mark.parametrize("arguments", ["not json", '{"n": NaN}', OVER_NESTED])
async def test_non_object_payloads_are_reported_instead_of_invoked(arguments: str) -> None:
async def test_non_object_payloads_pass_through_unchanged() -> None:
captured: dict[str, str] = {}
wrapped = factory._with_coerced_arguments(_capturing_tool(captured, _ARRAY))
result = await wrapped.on_invoke_tool(cast("Any", None), arguments)
assert result.startswith("probe: the tool call was not executed")
assert "raw_input" not in captured
assert await wrapped.on_invoke_tool(cast("Any", None), "not json") == "ok"
assert captured["raw_input"] == "not json"
@pytest.mark.asyncio
+3 -146
View File
@@ -21,7 +21,7 @@ from strix.core.sessions import open_agent_session
if TYPE_CHECKING:
from collections.abc import AsyncIterator, Awaitable, Callable
from collections.abc import AsyncIterator, Callable
from pathlib import Path
@@ -31,7 +31,6 @@ _CALL_LATENCY_S = 0.005
class _FakeLedger:
def __init__(self) -> None:
self.on_pause: Callable[[], Awaitable[None]] | None = None
self.cost = 0.0
self.calls: list[str] = []
self.remaining: dict[str, int] = {}
@@ -81,8 +80,6 @@ class _FakeStream:
await self._hooks.on_llm_start(ctx_wrapper, self._agent, None, input_items)
except BudgetPausedError as exc:
self.run_loop_exception = exc
if self._ledger.on_pause is not None:
await self._ledger.on_pause()
return
self._ledger.warned_inputs.append(input_items)
if self._ledger.gate is not None:
@@ -153,16 +150,13 @@ class _Scan:
monkeypatch: pytest.MonkeyPatch,
*,
max_budget_usd: float | None,
budget_policy: str = "pause",
) -> None:
self.ledger = _FakeLedger()
self.hooks = ReportUsageHooks(
model="test-model",
max_budget_usd=max_budget_usd,
budget_policy=budget_policy, # type: ignore[arg-type]
model="test-model", max_budget_usd=max_budget_usd, budget_policy="pause"
)
self.coordinator = AgentCoordinator()
self.coordinator.set_budget_policy(budget_policy) # type: ignore[arg-type]
self.coordinator.set_budget_policy("pause")
self.coordinator.set_budget_limit_setter(self.hooks.set_max_budget_usd)
monkeypatch.setattr(execution, "Runner", _fake_runner(self.ledger, self.coordinator))
monkeypatch.setattr(execution, "_compact_session", _noop_compact)
@@ -544,140 +538,3 @@ async def test_stop_policy_is_unchanged() -> None:
def test_budget_policy_is_validated() -> None:
with pytest.raises(ValueError, match="budget_policy"):
ReportUsageHooks(model="m", budget_policy="later") # type: ignore[arg-type]
def _pause_flags(coordinator: AgentCoordinator) -> tuple[bool, bool]:
return coordinator.budget_paused, coordinator.operator_paused
@pytest.mark.asyncio
async def test_operator_pause_holds_under_the_stop_policy(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
scan = _Scan(tmp_path, monkeypatch, max_budget_usd=None, budget_policy="stop")
agents = ("root", "child-a", "child-b")
with patch("strix.core.hooks.get_global_report_state", return_value=scan.ledger):
await scan.start_root(calls=6)
await scan.start_child("child-a", calls=6)
await scan.start_child("child-b", calls=6)
await _wait_until(lambda: scan.ledger.cost >= 3.0)
await scan.coordinator.pause_budget()
spent_at_pause = scan.ledger.cost
await _wait_until(lambda: _all_parked(scan.coordinator, *agents))
assert _pause_flags(scan.coordinator) == (True, True)
assert all(not task.done() for task in scan.tasks())
await scan.coordinator.send("child-a", {"from": "root", "content": "status?"})
await scan.coordinator.send("root", {"from": "user", "content": "keep going"})
await asyncio.sleep(0.05)
assert _pause_flags(scan.coordinator) == (True, True)
assert scan.ledger.cost == pytest.approx(spent_at_pause)
assert _all_parked(scan.coordinator, *agents)
woken = await scan.coordinator.resume_budget()
assert sorted(woken) == sorted(agents)
assert _pause_flags(scan.coordinator) == (False, False)
await asyncio.wait_for(asyncio.gather(*scan.tasks(), return_exceptions=True), timeout=5.0)
assert scan.ledger.cost == pytest.approx(18.0)
assert {aid: str(s) for aid, s in scan.coordinator.statuses.items()} == {
"root": "completed",
"child-a": "completed",
"child-b": "completed",
}
root_items = await scan.sessions[0].get_items()
assert any("keep going" in str(item) for item in root_items)
for session in scan.sessions:
session.close()
@pytest.mark.asyncio
async def test_a_resume_that_lands_before_the_park_still_continues(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
scan = _Scan(tmp_path, monkeypatch, max_budget_usd=None, budget_policy="stop")
resumed = 0
async def resume_at_once() -> None:
nonlocal resumed
resumed += 1
await scan.coordinator.resume_budget()
scan.ledger.on_pause = resume_at_once
with patch("strix.core.hooks.get_global_report_state", return_value=scan.ledger):
await scan.start_root(calls=6)
await scan.start_child("child-a", calls=6)
await _wait_until(lambda: scan.ledger.cost >= 2.0)
await scan.coordinator.pause_budget()
await asyncio.wait_for(asyncio.gather(*scan.tasks(), return_exceptions=True), timeout=5.0)
assert resumed >= 1
assert _pause_flags(scan.coordinator) == (False, False)
assert scan.ledger.cost == pytest.approx(12.0)
assert {aid: str(s) for aid, s in scan.coordinator.statuses.items()} == {
"root": "completed",
"child-a": "completed",
}
for session in scan.sessions:
session.close()
@pytest.mark.asyncio
async def test_operator_pause_gates_the_stop_policy_hooks() -> None:
coordinator = AgentCoordinator()
hooks = ReportUsageHooks(model="m", max_budget_usd=None)
assert hooks.budget_policy == "stop"
await coordinator.pause_budget()
with pytest.raises(BudgetPausedError):
await hooks.on_llm_start(_ctx(coordinator), MagicMock(), None, [])
await coordinator.resume_budget()
await hooks.on_llm_start(_ctx(coordinator), MagicMock(), None, [])
@pytest.mark.asyncio
async def test_a_user_message_lifts_only_the_limit_pause(tmp_path: Path) -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
session = open_agent_session("root", tmp_path / "agents.db")
await coordinator.attach_runtime("root", session=session)
extensions: list[int] = []
coordinator.set_budget_extender(lambda: extensions.append(1))
await coordinator.pause_budget()
await coordinator.park_for_budget("root")
assert coordinator.statuses["root"] == "budget_paused"
await coordinator.send("root", {"from": "user", "content": "keep going"})
assert _pause_flags(coordinator) == (True, True)
assert extensions == []
assert coordinator.pending_counts["root"] == 1
await coordinator.resume_budget()
await coordinator.pause_for_budget("root")
await coordinator.send("root", {"from": "user", "content": "keep going"})
assert _pause_flags(coordinator) == (False, False)
assert extensions == [1]
session.close()
@pytest.mark.asyncio
async def test_snapshot_and_reset_keep_an_operator_pause() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.pause_budget()
snap = await coordinator.snapshot()
restored = AgentCoordinator()
await restored.restore(snap)
assert _pause_flags(restored) == (True, True)
await restored.reset_budget_stops(budget_stopped=False, reserve_stopped=False)
assert _pause_flags(restored) == (True, True)
await restored.resume_budget()
await restored.pause_for_budget("root")
await restored.reset_budget_stops(budget_stopped=False, reserve_stopped=False)
assert _pause_flags(restored) == (False, False)
assert restored.statuses["root"] == "waiting"
+2 -9
View File
@@ -147,9 +147,7 @@ def _patch_budget(monkeypatch: pytest.MonkeyPatch, *, keep_tokens: int, window:
context.auto_compact = True
settings = SimpleNamespace(
context=context,
llm=SimpleNamespace(
api_key=None, api_base=None, timeout=1, extra_headers=None, extra_body=None
),
llm=SimpleNamespace(api_key=None, api_base=None, timeout=1, extra_headers=None),
)
monkeypatch.setattr(compaction, "load_settings", lambda: settings)
@@ -230,11 +228,7 @@ async def test_summarize_routes_through_provider_with_settings(
"load_settings",
lambda: SimpleNamespace(
llm=SimpleNamespace(
api_key=None,
api_base=None,
timeout=1,
extra_headers={"X-Feature-Key": "svc"},
extra_body={"service_tier": "flex"},
api_key=None, api_base=None, timeout=1, extra_headers={"X-Feature-Key": "svc"}
)
),
)
@@ -245,7 +239,6 @@ async def test_summarize_routes_through_provider_with_settings(
assert captured["model"] == "litellm/openai/some-model"
settings = captured["model_settings"]
assert settings.extra_headers == {"X-Feature-Key": "svc"}
assert settings.extra_body == {"service_tier": "flex"}
assert settings.max_tokens == 64
-57
View File
@@ -37,12 +37,6 @@ _LLM_ENV_KEYS = [
"STRIX_RUNTIME_BACKEND",
# TelemetrySettings
"STRIX_TELEMETRY",
# JSON-object settings
"LLM_EXTRA_HEADERS",
"LLM_EXTRA_BODY",
"STRIX_DEDUPE_MODEL",
"DEDUPE_LLM_EXTRA_HEADERS",
"DEDUPE_LLM_EXTRA_BODY",
]
@@ -412,57 +406,6 @@ def test_persist_current_replaces_corrupt_file(
assert json.loads(target.read_text(encoding="utf-8")) == {"env": {"STRIX_LLM": "env-model"}}
@pytest.mark.parametrize(
("env_var", "section", "field"),
[
("LLM_EXTRA_HEADERS", "llm", "extra_headers"),
("LLM_EXTRA_BODY", "llm", "extra_body"),
("DEDUPE_LLM_EXTRA_HEADERS", "dedupe", "extra_headers"),
("DEDUPE_LLM_EXTRA_BODY", "dedupe", "extra_body"),
],
)
def test_persisted_json_object_setting_reloads_without_env(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, env_var: str, section: str, field: str
) -> None:
loader.apply_config_override(tmp_path / "cli-config.json")
monkeypatch.setenv(env_var, '{"X-Tenant": "acme"}')
loader.load_settings()
loader.persist_current()
monkeypatch.delenv(env_var)
loader.apply_config_override(tmp_path / "cli-config.json")
settings = loader.load_settings()
assert getattr(getattr(settings, section), field) == {"X-Tenant": "acme"}
def test_json_object_setting_saved_as_object_still_loads(tmp_path: Path) -> None:
target = tmp_path / "cli-config.json"
target.write_text(
json.dumps({"env": {"LLM_EXTRA_BODY": {"service_tier": "flex"}}}), encoding="utf-8"
)
loader.apply_config_override(target)
assert loader.load_settings().llm.extra_body == {"service_tier": "flex"}
def test_invalid_saved_json_object_setting_is_rejected(tmp_path: Path) -> None:
target = tmp_path / "cli-config.json"
target.write_text(json.dumps({"env": {"LLM_EXTRA_BODY": "not json"}}), encoding="utf-8")
loader.apply_config_override(target)
with pytest.raises(ValidationError):
loader.load_settings()
def test_saved_string_setting_is_not_json_decoded(tmp_path: Path) -> None:
target = tmp_path / "cli-config.json"
target.write_text(json.dumps({"env": {"STRIX_LLM": '{"not": "decoded"}'}}), encoding="utf-8")
loader.apply_config_override(target)
assert loader.load_settings().llm.model == '{"not": "decoded"}'
def test_persist_current_sets_0600_mode(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("STRIX_LLM", "persisted-model")
target = tmp_path / "cli-config.json"
-1
View File
@@ -136,7 +136,6 @@ def test_cost_callback_estimates_cost_with_provider_prefixed_model() -> None:
with (
patch("strix.report.state.get_global_report_state", return_value=report_state),
patch("strix.report.state.resolve_litellm_model", side_effect=lambda model: model),
patch("litellm.completion_cost", side_effect=fake_completion_cost),
):
litellm_cost_callback(kwargs, response)
-39
View File
@@ -148,42 +148,3 @@ def test_config_file_loads_dedupe_model(
assert settings.dedupe.reasoning_effort == "minimal"
# Main model stays independent of the dedupe override.
assert settings.llm.model == "openai/root"
def test_dedicated_dedupe_model_uses_own_body_fields_not_main(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"safety_identifier": "main"}))
loader._cached = None
try:
dedupe = DedupeSettings(
STRIX_DEDUPE_MODEL="deepseek/cheap",
DEDUPE_LLM_EXTRA_BODY={"service_tier": "flex"},
)
settings = _dedupe_model_settings(dedupe, "deepseek/cheap", 300)
assert settings.extra_body == {"service_tier": "flex"}
finally:
loader._cached = None
def test_dedicated_dedupe_model_gets_no_main_body_fields_by_default(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"safety_identifier": "main"}))
loader._cached = None
try:
dedupe = DedupeSettings(STRIX_DEDUPE_MODEL="deepseek/cheap")
settings = _dedupe_model_settings(dedupe, "deepseek/cheap", 300)
assert settings.extra_body is None
finally:
loader._cached = None
def test_fallback_dedupe_inherits_main_body_fields(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"safety_identifier": "main"}))
loader._cached = None
try:
settings = _dedupe_model_settings(DedupeSettings(), "openai/main-model", 300)
assert settings.extra_body == {"safety_identifier": "main"}
finally:
loader._cached = None
-316
View File
@@ -1,316 +0,0 @@
from __future__ import annotations
import asyncio
import io
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any
import pytest
from agents.run_config import SandboxRunConfig
from agents.sandbox import SandboxAgent
from agents.sandbox.capabilities import Filesystem, Shell
from agents.sandbox.runtime_session_manager import SandboxRuntimeSessionManager
from strix.runtime import session_manager
from strix.runtime.backends import _BACKENDS, _BIND_MOUNT_BACKENDS, register_backend
from strix.runtime.deferred_session import DeferredSandboxClient, DeferredSandboxSession
if TYPE_CHECKING:
from collections.abc import Iterator
@dataclass
class _RuntimeSettings:
backend: str
@dataclass
class _Settings:
runtime: _RuntimeSettings
@dataclass
class _Endpoint:
host: str = "127.0.0.1"
port: int = 8080
tls: bool = False
@dataclass
class _ExecResult:
exit_code: int = 0
stdout: bytes = b""
stderr: bytes = b""
def ok(self) -> bool:
return self.exit_code == 0
class _State:
def __init__(self) -> None:
self.manifest: Any = None
self.session_id = "00000000-0000-4000-8000-000000000001"
def model_copy(self, *, update: dict[str, Any]) -> _State:
copy = _State()
copy.manifest = update.get("manifest", self.manifest)
return copy
class _Session:
def __init__(self) -> None:
self.state = _State()
self.execs: list[tuple[str, ...]] = []
self.writes: list[tuple[Path, bytes]] = []
self.limits: list[Any] = []
def _set_concurrency_limits(self, limits: Any) -> None:
self.limits.append(limits)
def _set_archive_limits(self, limits: Any) -> None:
self.limits.append(limits)
async def running(self) -> bool:
return True
async def resolve_exposed_port(self, _port: int) -> _Endpoint:
return _Endpoint()
async def exec(self, *argv: str, timeout: float | None = None, **_kw: Any) -> _ExecResult:
del timeout
self.execs.append(argv)
return _ExecResult()
async def write(self, path: Path, data: io.IOBase, **_kw: Any) -> None:
self.writes.append((path, data.read()))
class _Client:
def __init__(self) -> None:
self.deleted: list[Any] = []
async def delete(self, session: Any) -> None:
self.deleted.append(session)
async def _no_caido(*_args: Any, **_kwargs: Any) -> None:
return None
class _Backend:
def __init__(self, *, fail: bool = False) -> None:
self.gate = asyncio.Event()
self.calls = 0
self.fail = fail
self.client = _Client()
self.session = _Session()
async def __call__(self, **_kwargs: Any) -> tuple[Any, Any]:
self.calls += 1
await self.gate.wait()
if self.fail:
raise RuntimeError("bring-up exploded")
return self.client, self.session
@pytest.fixture
def backend(monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest) -> Iterator[_Backend]:
name = f"deferred-{request.node.name}"
fake = _Backend(fail="fail" in request.node.name)
register_backend(name, fake, supports_bind_mounts=True)
monkeypatch.setattr(session_manager, "bootstrap_caido", _no_caido)
monkeypatch.setattr(
session_manager, "load_settings", lambda: _Settings(runtime=_RuntimeSettings(backend=name))
)
yield fake
_BACKENDS.pop(name, None)
_BIND_MOUNT_BACKENDS.discard(name)
async def _settle() -> None:
for _ in range(10):
await asyncio.sleep(0)
async def _create(scan_id: str, **kwargs: Any) -> dict[str, Any]:
return await session_manager.create_or_reuse(
scan_id, image="img", local_sources=[], defer=True, **kwargs
)
@pytest.mark.asyncio
async def test_deferred_bundle_returns_before_the_backend_finishes(backend: _Backend) -> None:
scan_id = "deferred-returns"
try:
bundle = await asyncio.wait_for(_create(scan_id), timeout=1)
session = bundle["session"]
assert isinstance(session, DeferredSandboxSession)
assert isinstance(bundle["client"], DeferredSandboxClient)
assert session.pending
await _settle()
assert backend.calls == 1
assert await session.running()
assert session.state.manifest.environment.value["PYTHONUNBUFFERED"] == "1"
again = await session_manager.create_or_reuse(scan_id, image="img", local_sources=[])
assert again is bundle
assert backend.calls == 1
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_first_io_resolves_the_sandbox_then_delegates(backend: _Backend) -> None:
scan_id = "deferred-io"
try:
bundle = await _create(scan_id)
session = bundle["session"]
session._set_concurrency_limits("limits")
exec_task = asyncio.create_task(session.exec("echo", "hi"))
await asyncio.sleep(0.05)
assert not exec_task.done()
backend.gate.set()
result = await asyncio.wait_for(exec_task, timeout=1)
assert result.ok()
assert backend.session.execs == [("echo", "hi")]
assert not session.pending
assert backend.session.limits == ["limits"]
assert backend.session.state.manifest is session.state.manifest
await session.write(Path("/workspace/x"), io.BytesIO(b"data"))
assert backend.session.writes == [(Path("/workspace/x"), b"data")]
finally:
await session_manager.cleanup(scan_id)
assert backend.client.deleted == [backend.session]
@pytest.mark.asyncio
async def test_failed_bring_up_surfaces_on_first_io(backend: _Backend) -> None:
scan_id = "deferred-fail"
try:
bundle = await _create(scan_id)
session = bundle["session"]
backend.gate.set()
with pytest.raises(RuntimeError, match="bring-up exploded"):
await asyncio.wait_for(session.exec("true"), timeout=1)
assert not await session.running()
with pytest.raises(RuntimeError, match="bring-up exploded"):
await bundle["caido_client"].get()
finally:
await session_manager.cleanup(scan_id)
assert backend.client.deleted == []
@pytest.mark.asyncio
async def test_cleanup_while_pending_cancels_the_bring_up(backend: _Backend) -> None:
scan_id = "deferred-cancel"
bundle = await _create(scan_id)
session = bundle["session"]
await asyncio.wait_for(session_manager.cleanup(scan_id), timeout=1)
assert session.bring_up.cancelled()
assert backend.client.deleted == []
assert session_manager._SESSION_CACHE.get(scan_id) is None
backend.gate.set()
await _settle()
assert backend.client.deleted == [backend.session]
@pytest.mark.asyncio
async def test_cleanup_while_pending_ignores_a_failed_late_bring_up(backend: _Backend) -> None:
scan_id = "deferred-cancel-fail"
await _create(scan_id)
await asyncio.wait_for(session_manager.cleanup(scan_id), timeout=1)
backend.gate.set()
await _settle()
assert backend.client.deleted == []
@pytest.mark.asyncio
async def test_extra_files_are_staged_inside_the_deferred_bring_up(backend: _Backend) -> None:
scan_id = "deferred-extra"
try:
bundle = await _create(
scan_id,
extra_files=[{"workspace_path": "/workspace/.strix/notes.md", "content": "hi"}],
)
assert backend.session.writes == []
backend.gate.set()
await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
assert len(backend.session.writes) == 1
assert backend.session.execs
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_awaitable_extra_files_do_not_delay_the_backend_call(backend: _Backend) -> None:
scan_id = "deferred-extra-awaitable"
files: asyncio.Future[list[dict[str, Any]] | None] = asyncio.get_running_loop().create_future()
try:
bundle = await _create(scan_id, extra_files=files)
backend.gate.set()
await _settle()
assert backend.calls == 1
assert backend.session.writes == []
files.set_result([{"workspace_path": "/workspace/.strix/notes.md", "content": "hi"}])
await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
assert len(backend.session.writes) == 1
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_awaitable_extra_files_resolving_to_none_stage_nothing(backend: _Backend) -> None:
scan_id = "deferred-extra-none"
async def _nothing() -> list[dict[str, Any]] | None:
return None
try:
bundle = await _create(scan_id, extra_files=_nothing())
backend.gate.set()
await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
assert backend.session.writes == []
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_default_path_still_blocks_until_the_sandbox_is_up(backend: _Backend) -> None:
scan_id = "deferred-off"
backend.gate.set()
try:
bundle = await session_manager.create_or_reuse(scan_id, image="img", local_sources=[])
assert bundle["session"] is backend.session
assert bundle["client"] is backend.client
finally:
await session_manager.cleanup(scan_id)
assert backend.client.deleted == [backend.session]
@pytest.mark.asyncio
async def test_sdk_session_prep_does_not_wait_for_the_bring_up(backend: _Backend) -> None:
scan_id = "deferred-sdk-prep"
try:
bundle = await _create(scan_id)
agent = SandboxAgent(
name="root", instructions="x", tools=[], capabilities=[Filesystem(), Shell()]
)
manager = SandboxRuntimeSessionManager(
starting_agent=agent,
sandbox_config=SandboxRunConfig(client=bundle["client"], session=bundle["session"]),
run_state=None,
)
session = await asyncio.wait_for(
manager.ensure_session(
agent=agent, capabilities=list(agent.capabilities), is_resumed_state=False
),
timeout=1,
)
assert session is bundle["session"]
assert session.pending
backend.gate.set()
await asyncio.wait_for(session.resolve(), timeout=1)
assert not session.pending
finally:
await session_manager.cleanup(scan_id)
+2 -54
View File
@@ -310,8 +310,8 @@ def test_make_model_settings_forces_required_for_anyllm_routed_openai_model() ->
assert settings.tool_choice == "required"
def test_make_model_settings_enables_parallel_tool_calls_by_default() -> None:
assert make_model_settings("none", model_name="gpt-4o").parallel_tool_calls is True
def test_make_model_settings_disables_parallel_tool_calls_by_default() -> None:
assert make_model_settings("none", model_name="gpt-4o").parallel_tool_calls is False
def test_make_model_settings_omits_parallel_tool_calls_without_tools() -> None:
@@ -361,58 +361,6 @@ def test_make_model_settings_extra_headers_survive_reasoning_resolve() -> None:
assert settings.extra_headers == {"X-Feature-Key": "svc"}
def test_make_model_settings_sets_extra_body() -> None:
settings = make_model_settings(
"none",
model_name="openai/gpt-5",
extra_body={"service_tier": "flex", "safety_identifier": "user-hash"},
)
assert settings.extra_body == {"service_tier": "flex", "safety_identifier": "user-hash"}
def test_make_model_settings_omits_extra_body_when_unset() -> None:
assert make_model_settings("none", model_name="gpt-4o").extra_body is None
def test_make_model_settings_extra_body_survives_reasoning_resolve() -> None:
settings = make_model_settings(
"high",
model_name="openai/o3",
extra_body={"service_tier": "flex"},
)
assert settings.extra_body == {"service_tier": "flex"}
def test_make_model_settings_extra_body_merges_with_max_reasoning_field() -> None:
settings = make_model_settings(
"max",
model_name="deepseek/deepseek-v4-flash",
extra_body={"safety_identifier": "user-hash"},
)
assert settings.extra_body == {"safety_identifier": "user-hash", "reasoning_effort": "max"}
def test_make_model_settings_max_reasoning_overrides_extra_body_reasoning_field() -> None:
settings = make_model_settings(
"max",
model_name="deepseek/deepseek-v4-flash",
extra_body={"reasoning_effort": "low"},
)
assert settings.extra_body == {"reasoning_effort": "max"}
def test_make_model_settings_does_not_mutate_caller_extra_body() -> None:
extra_body = {"safety_identifier": "user-hash"}
make_model_settings("max", model_name="deepseek/deepseek-v4-flash", extra_body=extra_body)
assert extra_body == {"safety_identifier": "user-hash"}
def test_make_model_settings_timeout_survives_reasoning_resolve() -> None:
# Reasoning is resolved via ModelSettings.resolve(); the timeout in extra_args
# must not be dropped when a reasoning override is merged in.
-97
View File
@@ -1,97 +0,0 @@
"""Tests for LLM_EXTRA_BODY: extra fields merged into every LLM request body."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
import httpx
import litellm
import openai
import pytest
from agents.extensions.models.litellm_model import LitellmModel
from agents.models.interface import ModelTracing
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
from agents.models.openai_responses import OpenAIResponsesModel
from strix.config import loader
from strix.config.loader import load_settings
from strix.core.inputs import make_model_settings
if TYPE_CHECKING:
from collections.abc import Iterator
from agents.models.interface import Model
_EXTRA_BODY = {"service_tier": "flex", "safety_identifier": "user-hash"}
@pytest.fixture(autouse=True)
def _reset(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
monkeypatch.delenv("LLM_EXTRA_BODY", raising=False)
monkeypatch.setattr(loader, "_cached", None)
monkeypatch.setattr(loader, "_override", None)
yield
def test_extra_body_parsed_from_json_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"service_tier": "flex", "n": 1}))
assert load_settings().llm.extra_body == {"service_tier": "flex", "n": 1}
async def _send(model: Model) -> None:
await model.get_response(
system_instructions=None,
input="Reply with just 'OK'.",
model_settings=make_model_settings(None, model_name="gpt-5", extra_body=_EXTRA_BODY),
tools=[],
output_schema=None,
handoffs=[],
tracing=ModelTracing.DISABLED,
previous_response_id=None,
conversation_id=None,
prompt=None,
)
def _capturing_client(captured: dict[str, Any]) -> openai.AsyncOpenAI:
def handler(request: httpx.Request) -> httpx.Response:
captured.update(json.loads(request.content))
return httpx.Response(400, json={"error": {"message": "captured"}})
return openai.AsyncOpenAI(
api_key="test",
base_url="https://gateway.example/v1",
max_retries=0,
http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
)
@pytest.mark.asyncio
@pytest.mark.parametrize("model_cls", [OpenAIChatCompletionsModel, OpenAIResponsesModel])
async def test_extra_body_reaches_native_openai_request_body(model_cls: type) -> None:
captured: dict[str, Any] = {}
with pytest.raises(openai.BadRequestError):
await _send(model_cls(model="gpt-5", openai_client=_capturing_client(captured)))
assert {k: captured.get(k) for k in _EXTRA_BODY} == _EXTRA_BODY
@pytest.mark.asyncio
async def test_extra_body_passed_to_litellm(monkeypatch: pytest.MonkeyPatch) -> None:
captured: dict[str, Any] = {}
async def fake_acompletion(**kwargs: Any) -> Any:
captured.update(kwargs)
raise RuntimeError("captured")
monkeypatch.setattr(litellm, "acompletion", fake_acompletion)
with pytest.raises(RuntimeError, match="captured"):
await _send(LitellmModel(model="openai/gpt-5", base_url="https://gateway.example/v1"))
assert captured["extra_body"] == _EXTRA_BODY
-50
View File
@@ -60,7 +60,6 @@ def test_preflight_request_fails_after_its_own_timeout_with_a_clear_message() ->
model, # type: ignore[arg-type]
model_name="openai/gpt-4o",
extra_headers=None,
extra_body=None,
timeout=1,
api_base_setting="DEDUPE_LLM_API_BASE",
)
@@ -131,52 +130,3 @@ def test_warm_up_checks_the_dedupe_model_with_its_own_headers_and_the_preflight_
"DEDUPE_LLM_API_BASE",
]
assert calls[1][0] is dedupe_model
def test_warm_up_checks_each_model_with_its_own_body_fields(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.setenv("STRIX_LLM", "openai/gpt-4o")
monkeypatch.setenv("LLM_API_KEY", "sk-test")
monkeypatch.setenv("LLM_EXTRA_BODY", '{"safety_identifier": "main"}')
monkeypatch.setenv("STRIX_DEDUPE_MODEL", "anthropic/claude-sonnet-4-5")
monkeypatch.setenv("DEDUPE_LLM_EXTRA_BODY", '{"service_tier": "flex"}')
_fresh_settings(monkeypatch, tmp_path)
calls: list[dict[str, Any]] = []
async def record(_model: Any, **kwargs: Any) -> None:
calls.append(kwargs)
monkeypatch.setattr(scan_setup, "preflight_request", record)
monkeypatch.setattr(cli_main, "preflight_request", record)
monkeypatch.setattr("strix.config.models.configure_sdk_model_defaults", lambda _settings: None)
monkeypatch.setattr("strix.report.dedupe.resolve_dedupe_model", lambda _dedupe, _name: object())
asyncio.run(cli_main.warm_up_llm())
assert [kwargs["extra_body"] for kwargs in calls] == [
{"safety_identifier": "main"},
{"service_tier": "flex"},
]
def test_preflight_request_sends_the_extra_body() -> None:
seen: list[Any] = []
class _Recorder:
async def get_response(self, *, model_settings: Any, **_: Any) -> None:
seen.append(model_settings.extra_body)
asyncio.run(
preflight_request(
_Recorder(), # type: ignore[arg-type]
model_name="openai/gpt-4o",
extra_headers=None,
extra_body={"safety_identifier": "main"},
timeout=5,
api_base_setting="LLM_API_BASE",
)
)
assert seen == [{"safety_identifier": "main"}]
+1
View File
@@ -663,6 +663,7 @@ async def test_dependency_report_records_reachability(report_state: ReportState)
== "src/render.ts:14 calls `_.template()`."
)
assert "**Usage analysis:**" in report["evidence"]
assert "not a proof of exploitability or of safety" in report["evidence"]
# The level must never influence the rating — that comes from the contextual
# breakdown, or from advisory_cvss when no breakdown applies.
assert report["severity"] == "high"
-4
View File
@@ -263,11 +263,7 @@ def test_narrow_terminals_never_wrap_a_row() -> None:
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX key reader")
def test_posix_key_reader_handles_multibyte_and_arrows() -> None:
tty = pytest.importorskip("tty")
master, slave = os.openpty()
# Without this the pty echoes the writes into an output queue nobody reads,
# and read_raw_key's TCSADRAIN restore blocks forever on macOS.
tty.setcbreak(slave)
try:
with os.fdopen(slave, "r+b", buffering=0) as stream:
os.write(master, "\u00e9".encode())
-1
View File
@@ -28,7 +28,6 @@ def _wire_runner(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
timeout=300,
prompt_cache=True,
extra_headers=None,
extra_body=None,
),
runtime=types.SimpleNamespace(max_context_images=3),
)
-1
View File
@@ -44,7 +44,6 @@ def _settings() -> Any:
timeout=300,
prompt_cache=True,
extra_headers=None,
extra_body=None,
),
runtime=types.SimpleNamespace(max_context_images=3),
)
-1
View File
@@ -42,7 +42,6 @@ async def test_persistent_rate_limit_stops_gracefully(
timeout=300,
prompt_cache=True,
extra_headers=None,
extra_body=None,
),
runtime=types.SimpleNamespace(max_context_images=3),
)
-1
View File
@@ -64,7 +64,6 @@ def _patch_engine_scaffold(
timeout=300,
prompt_cache=True,
extra_headers=None,
extra_body=None,
),
runtime=types.SimpleNamespace(max_context_images=3),
)
-1
View File
@@ -52,7 +52,6 @@ def _settings() -> Any:
timeout=300,
prompt_cache=True,
extra_headers=None,
extra_body=None,
),
runtime=types.SimpleNamespace(max_context_images=3),
)
-300
View File
@@ -1,300 +0,0 @@
"""Tests for keeping replayed tool-call arguments valid JSON.
A model that emits a tool call with malformed ``arguments`` fails that one
call, but the raw string is recorded in the session. Strict OpenAI-compatible
servers validate every assistant tool call in the request, so each later turn
is rejected and the agent can never recover. A gateway that validates
arguments the way those servers do proves both the failure and the fix.
"""
from __future__ import annotations
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import TYPE_CHECKING, Any
import pytest
from agents import Agent, Runner, function_tool
from agents.models.interface import Model, ModelProvider
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
from agents.run import RunConfig
from openai import AsyncOpenAI, BadRequestError
from openai.types.responses import ResponseFunctionToolCall
from strix.config.models import _NonStreamingModel, _TurnGuardModel
from strix.config.tool_call_arguments import (
MALFORMED_ARGUMENTS_KEY,
_loads_strict,
describe_malformed_arguments,
repair_arguments,
repair_history_arguments,
repair_input,
)
if TYPE_CHECKING:
from collections.abc import Iterator
TRUNCATED = '{"n": 1'
OVER_NESTED = '{"a":' + "[" * 100_000
def _tool_call_completion(arguments: str) -> dict[str, Any]:
return {
"id": "chatcmpl-1",
"object": "chat.completion",
"created": 0,
"model": "gw-model",
"choices": [
{
"index": 0,
"finish_reason": "tool_calls",
"message": {
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "do_thing", "arguments": arguments},
}
],
},
}
],
"usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7},
}
def _text_completion(text: str) -> dict[str, Any]:
return {
"id": "chatcmpl-2",
"object": "chat.completion",
"created": 0,
"model": "gw-model",
"choices": [
{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": text}}
],
"usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8},
}
_REQUESTS: list[list[dict[str, Any]]] = []
def _assistant_arguments(messages: list[dict[str, Any]]) -> list[str]:
return [
str(call["function"]["arguments"])
for message in messages
for call in message.get("tool_calls") or []
]
def _tool_results(messages: list[dict[str, Any]]) -> list[str]:
return [str(m.get("content")) for m in messages if m.get("role") == "tool"]
class _StrictHandler(BaseHTTPRequestHandler):
"""Gateway that rejects malformed assistant tool-call arguments, like vLLM/SGLang do."""
def log_message(self, *args: Any) -> None:
pass
def do_POST(self) -> None:
length = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(length) or b"{}")
messages = body.get("messages", [])
_REQUESTS.append(messages)
for arguments in _assistant_arguments(messages):
try:
json.loads(arguments)
except ValueError:
self._respond(
400,
{
"object": "error",
"message": "Assistant tool call function.arguments must be valid JSON.",
"type": "BadRequest",
"param": None,
"code": 400,
},
)
return
if len(_REQUESTS) == 1:
self._respond(200, _tool_call_completion(TRUNCATED))
else:
self._respond(200, _text_completion("all done"))
def _respond(self, status: int, payload: dict[str, Any]) -> None:
encoded = json.dumps(payload).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(encoded)))
self.end_headers()
self.wfile.write(encoded)
@pytest.fixture
def strict_gateway() -> Iterator[str]:
_REQUESTS.clear()
server = HTTPServer(("127.0.0.1", 0), _StrictHandler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
yield f"http://127.0.0.1:{server.server_address[1]}/v1"
finally:
server.shutdown()
server.server_close()
def _model(base_url: str) -> Model:
client = AsyncOpenAI(api_key="tok", base_url=base_url, max_retries=0)
return _NonStreamingModel(OpenAIChatCompletionsModel(model="gw-model", openai_client=client))
async def _run_agent(base_url: str, *, wrap: bool) -> Any:
@function_tool
def do_thing(n: int) -> str:
return f"did {n}"
class _Provider(ModelProvider):
def get_model(self, model_name: str | None) -> Model: # noqa: ARG002
model = _model(base_url)
return _TurnGuardModel(model) if wrap else model
agent = Agent(name="t", instructions="use the tool", tools=[do_thing], model="gw-model")
result = Runner.run_streamed(
agent, input="please", run_config=RunConfig(model_provider=_Provider())
)
async for _ in result.stream_events():
pass
return result
@pytest.mark.asyncio
async def test_malformed_arguments_poison_every_later_turn_without_the_wrapper(
strict_gateway: str,
) -> None:
# Repro: the model truncates a tool call's arguments once. The tool fails
# that call gracefully, but the next request replays the raw string and
# the provider rejects the whole conversation from then on.
with pytest.raises(BadRequestError, match="must be valid JSON"):
await _run_agent(strict_gateway, wrap=False)
assert _assistant_arguments(_REQUESTS[-1]) == [TRUNCATED]
@pytest.mark.asyncio
async def test_malformed_arguments_are_replayed_as_valid_json(strict_gateway: str) -> None:
result = await _run_agent(strict_gateway, wrap=True)
assert result.final_output == "all done"
(arguments,) = _assistant_arguments(_REQUESTS[-1])
assert json.loads(arguments) == {MALFORMED_ARGUMENTS_KEY: TRUNCATED}
(tool_result,) = _tool_results(_REQUESTS[-1])
assert "JSON" in tool_result
@pytest.mark.parametrize(
("arguments", "expected"),
[
('{"cmd": "ls', json.dumps({MALFORMED_ARGUMENTS_KEY: '{"cmd": "ls'})),
("[1, 2]", json.dumps({MALFORMED_ARGUMENTS_KEY: "[1, 2]"})),
("", "{}"),
(" ", "{}"),
(None, "{}"),
],
)
def test_repair_arguments_rewrites_anything_but_a_json_object(
arguments: str | None, expected: str
) -> None:
assert repair_arguments(arguments) == expected
@pytest.mark.parametrize(
"arguments", ["{}", '{"cmd": "ls -la"}', '{"nested": {"a": [1]}}', '{"n": "NaN"}']
)
def test_repair_arguments_leaves_json_objects_alone(arguments: str) -> None:
assert repair_arguments(arguments) is None
@pytest.mark.parametrize("arguments", ['{"n": NaN}', '{"n": Infinity}', '{"n": -Infinity}'])
def test_repair_arguments_wraps_non_standard_json_constants(arguments: str) -> None:
repaired = repair_arguments(arguments)
assert repaired is not None
assert _loads_strict(repaired) == {MALFORMED_ARGUMENTS_KEY: arguments}
def test_repair_arguments_wraps_over_nested_json() -> None:
repaired = repair_arguments(OVER_NESTED)
assert repaired is not None
assert _loads_strict(repaired) == {MALFORMED_ARGUMENTS_KEY: OVER_NESTED}
def test_history_repair_rewrites_only_malformed_calls() -> None:
items = [
{"role": "user", "content": "go"},
{"type": "function_call", "call_id": "a", "name": "x", "arguments": '{"cmd": "ls'},
{"type": "function_call_output", "call_id": "a", "output": "invalid JSON"},
{"type": "function_call", "call_id": "b", "name": "y", "arguments": '{"ok": true}'},
ResponseFunctionToolCall(call_id="c", name="z", arguments="", type="function_call"),
]
rebuilt, changed = repair_history_arguments(items)
assert changed
assert rebuilt[0] is items[0]
assert json.loads(rebuilt[1]["arguments"]) == {MALFORMED_ARGUMENTS_KEY: '{"cmd": "ls'}
assert rebuilt[1]["call_id"] == "a"
assert rebuilt[2] is items[2]
assert rebuilt[3] is items[3]
assert isinstance(rebuilt[4], ResponseFunctionToolCall)
assert rebuilt[4].arguments == "{}"
assert isinstance(items[1], dict)
assert items[1]["arguments"] == '{"cmd": "ls'
def test_repair_input_returns_same_object_when_nothing_changes() -> None:
items = [{"type": "function_call", "call_id": "a", "name": "x", "arguments": "{}"}]
assert repair_input(items) is items
assert repair_input("plain prompt") == "plain prompt"
def test_repair_input_wraps_over_nested_function_call_arguments() -> None:
items = [
{
"type": "function_call",
"call_id": "a",
"name": "x",
"arguments": OVER_NESTED,
}
]
repaired = repair_input(items)
assert isinstance(repaired, list)
assert _loads_strict(repaired[0]["arguments"]) == {MALFORMED_ARGUMENTS_KEY: OVER_NESTED}
@pytest.mark.parametrize(
"arguments",
['{"cmd": "ls -la', "[1, 2]", "null", "not json", '{"n": NaN}', OVER_NESTED],
)
def test_describe_malformed_arguments_tells_the_model_to_reissue(arguments: str) -> None:
message = describe_malformed_arguments("exec_command", arguments)
assert message is not None
assert message.startswith("exec_command: the tool call was not executed")
assert "Re-issue the call" in message
@pytest.mark.parametrize("arguments", ["", " ", "{}", '{"cmd": "ls"}'])
def test_describe_malformed_arguments_accepts_objects_and_empty_input(arguments: str) -> None:
assert describe_malformed_arguments("exec_command", arguments) is None