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
Ian 442b3c3df5 fix(tui): give the targets box only rows the agents panel would get, and always count hidden targets 2026-10-12 20:17:29 +03:00
Ian 710fe7b3b6 feat(tui): move the run's targets into their own sidebar box under the viewer 2026-10-12 20:17:29 +03:00
Ian b68cdb8edc feat(tui): list the run's targets in the sidebar stats panel 2026-10-12 20:17:29 +03:00
yoni-at-strix 429f6218f5 feat(reporting): accept code_locations on dependency reports (#1366) 2026-10-12 08:54:36 -07:00
yoni-at-strix f7a32ae62d docs(readme): add update instructions (#1501) 2026-10-10 20:31:50 -04:00
ian-at-strix 62b496430d feat(agents): let callers choose the top-level agent's finish tool (#1496) 2026-10-08 19:11:50 -04:00
alex s f5a900416b fix(mcp): load catalogs before agents discover connections (#1492)
* fix(mcp): load catalogs before agents discover connections

* fix(mcp): bound discovery waits without cancelling catalog loads

* fix(mcp): keep discovery waiters alive across deadlines
2026-10-08 12:42:39 -04:00
ian-at-strix 4695290682 feat(llm): bound the first stream event and the whole stream (#1487) 2026-10-07 18:35:45 -07:00
ian-at-strix 928f053610 feat(llm): retry 4xx model errors (#1486)
* feat(llm): log the upstream provider behind OpenRouter

LiteLLM drops OpenRouter's top-level `provider` from stream chunks (including error chunks) and the `metadata.provider_name` of non-2xx replies, so the request log could not say which provider served or failed an attempt. Record both, plus OpenRouter's `metadata.error_type`, per attempt and emit them as `upstream=` / `upstream_error=` on the llm_request line.

* feat(llm): retry 4xx model errors

* test: let the blocked-provider test run out of 4xx retries

* refactor: retry any 4xx or 5xx status

* fix: don't retry 401, 402 or 403

* fix: don't retry 404
2026-10-07 18:35:33 -07:00
ian-at-strix 39792c9d61 feat(llm): log the upstream provider behind OpenRouter (#1459)
LiteLLM drops OpenRouter's top-level `provider` from stream chunks (including error chunks) and the `metadata.provider_name` of non-2xx replies, so the request log could not say which provider served or failed an attempt. Record both, plus OpenRouter's `metadata.error_type`, per attempt and emit them as `upstream=` / `upstream_error=` on the llm_request line.
2026-10-07 18:35:12 -07:00
ian-at-strix 278b6a280b fix(execution): only strip images on image-rejection errors (#1461)
* fix(execution): only strip images on image-rejection errors

The image-strip recovery fired on any 400/404/422, wiping a session's screenshots and replaying the turn on unrelated errors (e.g. an upstream incomplete-tool-call 400, a context overflow). Require an image-related message too.

* fix(execution): match only replies that mean the model takes no images

* fix(execution): match DeepSeek's image rejection

* fix(execution): guess at Vercel AI Gateway's image rejection
2026-10-07 23:59:43 +03:00
ian-at-strix 5a30961852 feat(llm): scrub images from every request to a text-only model (#1466)
* feat(llm): scrub images from every request to a text-only model

Replace images at the MCP call before any result_transform, give load_skill the
same agent_browser screenshot swap as the system prompt, and add a
call_model_input_filter that turns any image left in the input into a tool error
while chaining any filter already set.

* chore: fix the ruff and bandit failures on main
2026-10-07 23:59:19 +03:00
ian-at-strix 7f1d148778 feat(llm): keep images away from text-only models (#1465)
* feat(llm): keep images away from text-only models

Leave view_image out of the tools, replace MCP image blocks with a text note,
and swap agent_browser's screenshot section for text-only guidance when the
run's model does not accept images.

* refactor(llm): disable view_image via is_enabled and gate skill image guidance by modality

* refactor(skills): render opt-in Jinja skills instead of comment markers

* Tweaked skill
2026-10-07 23:11:58 +03:00
ian-at-strix acd64e929c feat(llm): detect whether a model accepts image input (#1460)
* feat(llm): detect whether a model accepts image input

Add model_supports_images(): OpenRouter models are checked against architecture.input_modalities from OpenRouter's public model list (fetched once per process), others against LiteLLM's supports_vision. Unknown models and lookup failures count as image-capable. Not wired in yet.

* refactor(llm): read image support from litellm's catalog, assuming it for unknown models

* style: drop a trailing space in the image-support docstring
2026-10-07 22:58:43 +03:00
ian-at-strix 11976ca119 fix(logging): suppress Agents SDK PTY warning that interferes with TUI (#1392) 2026-10-07 11:32:42 -07:00
Ian 6f2803295f feat(tui): set the terminal title to the scan target 2026-10-07 20:41:34 +03:00
48 changed files with 1598 additions and 409 deletions

No files matched your search

+12
View File
@@ -94,6 +94,18 @@ strix --target ./app-directory
> [!NOTE]
> First run automatically pulls the sandbox Docker image. Results are saved to `strix_runs/<run-name>`
### Updating
```bash
# Self-update the standalone binary to the latest release
strix --update
# ...or re-run the installer
curl -sSL https://strix.ai/install | bash
```
Installed with pipx, uv, or pip? `strix --update` prints the matching upgrade command instead (`pipx upgrade strix-agent`, `uv tool upgrade strix-agent`, `pip install --upgrade strix-agent`). Strix also shows a notice with the right command when a new version is available.
---
## Ways to Run Strix
+13 -2
View File
@@ -375,11 +375,15 @@ def _configure_filesystem_tools(
)
def _make_filesystem_configurator(*, chat_completions: bool, strict_schemas: bool) -> Any:
def _make_filesystem_configurator(
*, chat_completions: bool, strict_schemas: bool, supports_images: bool = True
) -> Any:
def configure(toolset: Any) -> None:
_configure_filesystem_tools(
toolset, chat_completions=chat_completions, strict_schemas=strict_schemas
)
if not supports_images:
toolset.view_image.is_enabled = False
return configure
@@ -670,6 +674,8 @@ def build_strix_agent(
system_prompt_context: dict[str, Any] | None = None,
extra_tools: Sequence[Tool] | None = None,
instructions_override: str | None = None,
supports_images: bool = True,
finish_tool: Tool = finish_scan,
) -> SandboxAgent[Any]:
"""Build a SandboxAgent for either root or child use.
@@ -682,6 +688,7 @@ def build_strix_agent(
registered via ``register_agent_tools``.
instructions_override: Use this verbatim as the system prompt instead
of rendering the built-in scan prompt.
finish_tool: The tool that ends the run, given to the root agent.
"""
if instructions_override is not None:
instructions = instructions_override
@@ -694,6 +701,7 @@ def build_strix_agent(
is_diff_scoped=is_diff_scoped,
interactive=interactive,
system_prompt_context=system_prompt_context,
supports_images=supports_images,
)
agent_tools = [*_EXTRA_TOOLS, *(extra_tools or [])]
@@ -701,7 +709,7 @@ def build_strix_agent(
# Yielding to the user is only meaningful when one is attached.
agent_tools.append(wait_for_user)
if is_root:
tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_scan]
tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_tool]
else:
tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
_ensure_unique_tool_names(tools)
@@ -733,6 +741,7 @@ def build_strix_agent(
configure_tools=_make_filesystem_configurator(
chat_completions=chat_completions_tools,
strict_schemas=strict_tool_schemas,
supports_images=supports_images,
),
),
Shell(
@@ -754,6 +763,7 @@ def make_child_factory(
chat_completions_tools: bool = False,
strict_tool_schemas: bool = True,
system_prompt_context: dict[str, Any] | None = None,
supports_images: bool = True,
) -> Any:
"""Return the runner-owned builder used by ``spawn_child_agent``.
@@ -774,6 +784,7 @@ def make_child_factory(
chat_completions_tools=chat_completions_tools,
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=system_prompt_context,
supports_images=supports_images,
)
return _factory
+2 -1
View File
@@ -85,6 +85,7 @@ def render_system_prompt(
interactive: bool = False,
system_prompt_context: dict[str, Any] | None = None,
include_scope: bool = True,
supports_images: bool = True,
) -> str:
"""Render the system prompt. Returns empty string on template failure.
@@ -119,7 +120,7 @@ def render_system_prompt(
is_root=is_root,
is_diff_scoped=is_diff_scoped,
)
skill_content = load_skills(skills_to_load)
skill_content = load_skills(skills_to_load, supports_images=supports_images)
env.globals["get_skill"] = lambda name: skill_content.get(name, "")
# Skills every agent of this kind loads come first, so siblings share them
+1 -1
View File
@@ -21,7 +21,7 @@ MCP CONNECTIONS (available this run):
{% if system_prompt_context.mcp_connections %}
- Connected this run (search one to find relevant tools):
{% for connection in system_prompt_context.mcp_connections %}
- {{ connection.name }} ({{ connection.tool_count }} tools){% if connection.purpose %}: {{ connection.purpose }}{% endif %}
- {{ connection.name }}{% if connection.state == "unavailable" %} (unavailable right now){% elif connection.tool_count is not none %} ({{ connection.tool_count }} tools){% endif %}{% if connection.purpose %}: {{ connection.purpose }}{% endif %}
{% endfor %}
{% endif %}
- Reach for a connection whenever the target itself cannot give you information a connection could: its database schema and access policies, real deployment or infrastructure configuration, known issues or prior findings, or server logs. In those cases call list_mcps early to see what is available, and prefer a connection's authoritative data over inferring from the target's responses. Do not wait to be told a connection exists.
+102 -16
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import asyncio
import contextlib
import inspect
import json
import logging
import os
import time
@@ -262,6 +263,8 @@ class _TurnGuardModel(Model):
not covered by the request timeout, which resets on any byte (keepalives
included). ``LLM_STREAM_IDLE_TIMEOUT`` bounds the gap between events so the
turn fails instead of hanging, and the existing retry path replays it.
``LLM_STREAM_FIRST_EVENT_TIMEOUT`` and ``LLM_STREAM_TOTAL_TIMEOUT`` bound
the first event and the whole stream.
"""
def __init__(
@@ -269,11 +272,15 @@ class _TurnGuardModel(Model):
inner: Model,
*,
max_tool_calls_per_turn: int = 0,
stream_idle_timeout: float = 0.0,
stream_idle_timeout: float | None = None,
first_event_timeout: float | None = None,
total_timeout: float | None = None,
) -> None:
self._inner = inner
self._max_tool_calls_per_turn = max_tool_calls_per_turn
self._stream_idle_timeout = stream_idle_timeout
self._first_event_timeout = first_event_timeout
self._total_timeout = total_timeout
def _limiter(self) -> TurnToolCallLimiter:
return TurnToolCallLimiter(self._max_tool_calls_per_turn)
@@ -360,7 +367,12 @@ class _TurnGuardModel(Model):
conversation_id=conversation_id,
prompt=prompt,
)
async for event in _with_idle_timeout(stream, self._stream_idle_timeout):
async for event in _with_timeouts(
stream,
idle=self._stream_idle_timeout,
first_event=self._first_event_timeout,
total=self._total_timeout,
):
guarded = _guard_event(event, rewriter, limiter)
if guarded is not None:
yield guarded
@@ -394,28 +406,68 @@ async def _aclose(stream: AsyncIterator[TResponseStreamEvent]) -> None:
await stream.aclose()
async def _with_idle_timeout(
stream: AsyncIterator[TResponseStreamEvent], timeout: float
async def _with_timeouts(
stream: AsyncIterator[TResponseStreamEvent],
*,
idle: float | None = None,
first_event: float | None = None,
total: float | None = None,
) -> AsyncIterator[TResponseStreamEvent]:
if timeout <= 0:
async for event in stream:
yield event
return
"""Bound the first event, the gap between events and the whole stream; None is no bound."""
iterator = stream.__aiter__()
started = time.monotonic()
while True:
limits: list[tuple[float, float, str]] = []
# Wait until the soonest limit expires: first-event (else idle) until the
# first event arrives, then idle; plus whatever is left of the total.
if first_event is not None:
limits.append((first_event, first_event, "stream_first_event_timeout"))
elif idle is not None:
limits.append((idle, idle, "stream_idle_timeout"))
if total is not None:
left = max(0.0, started + total - time.monotonic())
limits.append((left, total, "stream_total_timeout"))
try:
event = await asyncio.wait_for(iterator.__anext__(), timeout)
if limits:
event = await _next_event(iterator, *min(limits))
else:
event = await iterator.__anext__()
except StopAsyncIteration:
return
except TimeoutError:
except TimeoutError as exc:
await _aclose(stream)
message = f"model stream produced no event for {timeout:.0f}s"
logger.warning("%s; abandoning the turn", message)
raise TimeoutError(message) from None
logger.warning("%s; abandoning the turn", exc)
raise
first_event = None
yield event
async def _next_event(
iterator: AsyncIterator[TResponseStreamEvent], wait: float, limit: float, reason: str
) -> TResponseStreamEvent:
# asyncio.timeout() cancels without a message, so the request log could not
# tell this from a shutdown; this is asyncio.timeout() with a message.
task = asyncio.current_task()
assert task is not None
cancelling = task.cancelling()
expired = False
def expire() -> None:
nonlocal expired
expired = True
task.cancel(msg=f"{request_log.CANCEL_REASON_PREFIX}{reason}")
handle = asyncio.get_running_loop().call_later(wait, expire)
try:
return await iterator.__anext__()
except BaseException as exc:
if expired and task.uncancel() <= cancelling and isinstance(exc, asyncio.CancelledError):
raise TimeoutError(f"model stream hit {reason} ({limit:.0f}s)") from None
raise
finally:
handle.cancel()
def _guard_event(
event: TResponseStreamEvent, rewriter: TurnCallIdRewriter, limiter: TurnToolCallLimiter
) -> TResponseStreamEvent | None:
@@ -550,7 +602,10 @@ class StrixProvider(MultiProvider):
def get_model(self, model_name: str | None) -> Model:
llm = load_settings().llm
slug = codex.subscription_model(model_name)
idle_timeout = float(llm.stream_idle_timeout)
# The settings use 0 for no bound.
idle_timeout = float(llm.stream_idle_timeout) or None
first_event_timeout = float(llm.stream_first_event_timeout) or None
total_timeout = float(llm.stream_total_timeout) or None
if slug:
# The ChatGPT subscription backend is always streamed; it has no
# non-streaming mode to fall back to, so LLM_DISABLE_STREAMING
@@ -591,11 +646,13 @@ class StrixProvider(MultiProvider):
# The wrapper emits its single event only once the whole request
# is done, so an idle gap is meaningless here; the request
# timeout bounds it instead.
idle_timeout = 0.0
idle_timeout = first_event_timeout = total_timeout = None
return _TurnGuardModel(
model,
max_tool_calls_per_turn=llm.max_tool_calls_per_turn,
stream_idle_timeout=idle_timeout,
first_event_timeout=first_event_timeout,
total_timeout=total_timeout,
)
@@ -745,6 +802,11 @@ def _install_openrouter_stream_cost_capture() -> None:
class _StrixOpenRouterStreamingHandler(OpenRouterChatCompletionStreamingHandler):
def chunk_parser(self, chunk: dict[str, Any]) -> Any:
# Before parsing: LiteLLM raises on an error chunk and drops its
# top-level ``provider``.
request_log.record_upstream_provider(
chunk.get("provider"), _openrouter_error_type(chunk.get("error"))
)
stream = super().chunk_parser(chunk)
usage = chunk.get("usage")
response_id = chunk.get("id") or getattr(stream, "id", None)
@@ -763,12 +825,22 @@ def _install_openrouter_stream_cost_capture() -> None:
json_mode=json_mode,
)
def get_error_class(self, error_message: str, status_code: int, headers: Any) -> Any:
# A non-2xx reply names the provider in ``error.metadata``.
with contextlib.suppress(Exception):
error = json.loads(error_message)["error"]
request_log.record_upstream_provider(
error["metadata"].get("provider_name"), _openrouter_error_type(error)
)
return super().get_error_class(error_message, status_code, headers)
def transform_response(self, *args: Any, **kwargs: Any) -> Any:
# Non-streamed replies (LLM_DISABLE_STREAMING) skip the chunk parser.
response = super().transform_response(*args, **kwargs)
raw_response = kwargs.get("raw_response", args[1] if len(args) > 1 else None)
with contextlib.suppress(Exception):
body = raw_response.json() # type: ignore[union-attr]
request_log.record_upstream_provider(body.get("provider"))
if body.get("usage"):
record_openrouter_provider(body.get("provider"), body["usage"])
return response
@@ -789,6 +861,14 @@ def _install_openrouter_stream_cost_capture() -> None:
litellm.OpenrouterConfig = _StrixOpenrouterConfig # type: ignore[misc]
def _openrouter_error_type(error: object) -> object:
if isinstance(error, dict):
metadata = error.get("metadata")
if isinstance(metadata, dict):
return metadata.get("error_type")
return None
OPENROUTER_ATTRIBUTION_HEADERS = {
"HTTP-Referer": "https://strix.ai",
"X-Title": "Strix",
@@ -905,6 +985,12 @@ def model_supports_reasoning(model_name: str) -> bool:
return bool(entry and entry.get("supports_reasoning"))
def model_supports_images(model_name: str) -> bool:
"""Return whether the model accepts image input. Assume yes until proven otherwise."""
entry = _catalog_entry(model_name)
return entry is None or bool(entry.get("supports_vision"))
def _bare_openai_name(model_name: str) -> str:
name = model_name.strip().lower()
for prefix in ("litellm/", "any-llm/", "openai/"):
+4
View File
@@ -79,6 +79,10 @@ class LlmSettings(BaseSettings):
timeout: int = Field(default=300, alias="LLM_TIMEOUT")
preflight_timeout: int = Field(default=30, ge=1, alias="LLM_PREFLIGHT_TIMEOUT")
stream_idle_timeout: int = Field(default=300, ge=0, alias="LLM_STREAM_IDLE_TIMEOUT")
stream_first_event_timeout: int = Field(
default=120, ge=0, alias="LLM_STREAM_FIRST_EVENT_TIMEOUT"
)
stream_total_timeout: int = Field(default=600, ge=0, alias="LLM_STREAM_TOTAL_TIMEOUT")
max_tool_calls_per_turn: int = Field(
default=32,
ge=0,
+1
View File
@@ -71,6 +71,7 @@ class AgentCoordinator:
self._lock = asyncio.Lock()
self._snapshot_path: Path | None = None
self.is_shutting_down = False
self.root_finish_tool = "finish_scan"
self._budget_stopped = False
self._reserve_stopped = False
self._budget_paused = False
+45 -78
View File
@@ -6,6 +6,7 @@ import asyncio
import contextlib
import inspect
import logging
import re
import uuid
from collections.abc import Callable
from dataclasses import replace
@@ -32,16 +33,12 @@ from strix.core.sessions import (
enforce_image_budget,
open_agent_session,
replace_session_items,
scrub_images_from_items,
seed_initial_input,
strip_all_images_from_session,
)
from strix.llm import request_log
from strix.llm.compaction import (
CompactionNeededError,
is_context_overflow,
maybe_compact,
needs_compaction,
)
from strix.llm.compaction import is_context_overflow, maybe_compact
if TYPE_CHECKING:
@@ -61,6 +58,18 @@ logger = logging.getLogger(__name__)
StreamEventSink = Callable[[str, Any], None]
_INPUT_REJECTION_CODES = frozenset({400, 404, 422})
# Replies meaning "this model takes no images", not image errors in general: a
# context overflow that counts "image/vision expansion" must not match.
_IMAGE_REJECTION = re.compile(
r"no endpoints found that support image input" # OpenRouter
r"|image_url is only supported by certain models" # OpenAI
r"|is not a multimodal model|at most 0 image\(s\)" # vLLM
r"|does not support image input" # LiteLLM's Fireworks check
r"|doesn't support the image field" # Bedrock Converse
r"|unknown variant `image_url`" # DeepSeek, e.g. via Vercel AI Gateway
r"|'[^']*image[^']*' functionality not supported", # Vercel AI Gateway (AI SDK), unverified
re.IGNORECASE,
)
_MAX_COMPACTIONS_PER_CYCLE = 2
@@ -128,49 +137,26 @@ async def _compact_session(
)
async def _session_needs_compaction(agent: Any, session: Session, run_config: RunConfig) -> bool:
model = _run_config_model(run_config)
if model is None:
return False
return await asyncio.to_thread(
needs_compaction,
model,
_agent_instructions(agent),
_agent_tools_text(agent),
list(await session.get_items()),
)
_TEXT_ONLY_IMAGE_TEXT = "[error: this model cannot view images; use `snapshot -i` instead]"
def _with_compaction_check(run_config: RunConfig) -> RunConfig:
"""Stop the run before a model request that would overrun the context window so
the cycle can compact and resume: the SDK builds each request from its in-memory
history, so compacting the session mid-run would not shrink it."""
model = _run_config_model(run_config)
if model is None:
def _with_image_scrub(run_config: RunConfig, context: dict[str, Any]) -> RunConfig:
if context.get("supports_images", True):
return run_config
# Chain any filter already set; it sees the scrubbed input.
inner = run_config.call_model_input_filter
first_call = True
async def _check(data: CallModelData[Any]) -> ModelInputData:
nonlocal first_call
model_data = data.model_data
if inner is not None:
result = inner(data)
model_data = await result if inspect.isawaitable(result) else result
# Skip the first call: the cycle compacts right before the run, and the run's
# new input is only persisted after this filter, so stopping here would lose it.
if not first_call and await asyncio.to_thread(
needs_compaction,
model,
model_data.instructions or "",
_agent_tools_text(data.agent),
model_data.input,
):
raise CompactionNeededError(model)
first_call = False
return model_data
async def _scrub(data: CallModelData[Any]) -> ModelInputData:
model_data = replace(
data.model_data,
input=scrub_images_from_items(data.model_data.input, text=_TEXT_ONLY_IMAGE_TEXT),
)
if inner is None:
return model_data
result = inner(replace(data, model_data=model_data))
return await result if inspect.isawaitable(result) else result
return replace(run_config, call_model_input_filter=_check)
return replace(run_config, call_model_input_filter=_scrub)
_MAX_TRANSIENT_MODEL_RETRIES = 5
@@ -183,6 +169,12 @@ def _model_error_status_code(exc: BaseException) -> int | None:
return code if isinstance(code, int) else None
def _is_image_rejection(exc: BaseException) -> bool:
return _model_error_status_code(exc) in _INPUT_REJECTION_CODES and bool(
_IMAGE_REJECTION.search(str(exc))
)
def _is_transient_model_error(exc: BaseException) -> bool:
if codex.is_content_guardrail_error(exc):
return False
@@ -192,9 +184,7 @@ def _is_transient_model_error(exc: BaseException) -> bool:
return True
code = _model_error_status_code(exc)
if code is not None:
import litellm
return bool(litellm._should_retry(code))
return code >= 400 and code not in (401, 402, 403, 404)
return isinstance(exc, APIError)
@@ -312,7 +302,7 @@ async def _run_agent_loop(
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())
await coordinator.send(agent_id, _reserve_notice(coordinator.root_finish_tool))
if not (start_parked and interactive):
with contextlib.suppress(BudgetPausedError):
@@ -652,6 +642,7 @@ async def _run_until_lifecycle(
input_data = await _append_tool_required_message(
session=session,
context=context,
root_finish_tool=coordinator.root_finish_tool,
attempt=recoveries,
limit=recovery_limit,
interactive=interactive,
@@ -778,7 +769,6 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
image_strips = 0
compactions = 0
model_retries = 0
check_context = session is not None
request_log.set_retry_attempt(0)
while True:
stream: Any = None
@@ -801,7 +791,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
stream = Runner.run_streamed(
agent,
input=input_data,
run_config=_with_compaction_check(run_config) if check_context else run_config,
run_config=_with_image_scrub(run_config, context),
context=context,
max_turns=max_turns,
session=session,
@@ -859,11 +849,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
await coordinator.trigger_budget_stop()
raise
except Exception as exc:
if (
image_strips < 3
and session is not None
and getattr(exc, "status_code", None) in _INPUT_REJECTION_CODES
):
if image_strips < 3 and session is not None and _is_image_rejection(exc):
try:
stripped = await strip_all_images_from_session(session)
except Exception:
@@ -878,26 +864,6 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
)
input_data = []
continue
if session is not None and isinstance(exc, CompactionNeededError):
# Rerun without the check if compaction can't bring the session under
# the budget; a real overflow then falls to the capped recovery below.
try:
compacted = await _compact_session(agent, session, run_config, force=True)
check_context = compacted and not await _session_needs_compaction(
agent, session, run_config
)
except Exception:
logger.exception("context budget compaction failed for %s", agent_id)
check_context = False
logger.info(
"%s hit the context budget; %s",
agent_id,
"compacted and resuming"
if check_context
else "rerunning without the budget check",
)
input_data = []
continue
if (
compactions < _MAX_COMPACTIONS_PER_CYCLE
and session is not None
@@ -1023,8 +989,9 @@ async def _append_tool_required_message(
limit: int,
interactive: bool,
silent_yield: bool = False,
root_finish_tool: str = "finish_scan",
) -> list[dict[str, str]]:
finish_tool = "finish_scan" if context.get("parent_id") is None else "agent_finish"
finish_tool = root_finish_tool if context.get("parent_id") is None else "agent_finish"
if silent_yield:
message = (
"You called wait_for_user without having written anything to the user since "
@@ -1142,7 +1109,7 @@ async def notify_parent_on_terminal(
)
def _reserve_notice() -> dict[str, Any]:
def _reserve_notice(finish_tool: str) -> dict[str, Any]:
return {
"from": "system",
"type": "budget_reserve_stop",
@@ -1152,7 +1119,7 @@ def _reserve_notice() -> dict[str, Any]:
"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."
f"spawn new ones — wrap up now and call {finish_tool}."
),
}
@@ -1161,7 +1128,7 @@ 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())
await coordinator.send(root, _reserve_notice(coordinator.root_finish_tool))
async def _notify_parent_on_exit(
+8 -5
View File
@@ -104,16 +104,16 @@ _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."
"track so you can call {finish_tool} 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."
"{finish_tool} 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 "
"secure your findings and call {finish_tool} now — anything left unfinished when the "
"limit is hit is discarded."
),
)
@@ -138,8 +138,11 @@ _SUBAGENT_DIRECTIVES: tuple[str, ...] = (
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]
if not is_root:
return _SUBAGENT_DIRECTIVES[stage]
coordinator = coordinator_from_context(context.context)
finish_tool = coordinator.root_finish_tool if coordinator is not None else "finish_scan"
return _ROOT_DIRECTIVES[stage].format(finish_tool=finish_tool)
def _urgency(stage: int) -> str:
+40 -11
View File
@@ -23,6 +23,7 @@ from strix.config.models import (
StrixProvider,
configure_sdk_api_route,
configure_sdk_model_defaults,
model_supports_images,
supports_strict_tool_schemas,
uses_chat_completions_tool_schema,
)
@@ -48,6 +49,7 @@ from strix.report.state import get_global_report_state
from strix.runtime import session_manager
from strix.telemetry import set_scan_phase
from strix.telemetry.logging import set_scan_id, setup_scan_logging
from strix.tools.finish.tool import finish_scan
from strix.tools.output_store import (
WORKSPACE_SPILL_DIR,
configure_spill_writer,
@@ -57,6 +59,7 @@ from strix.tools.output_store import (
if TYPE_CHECKING:
from agents.memory import SQLiteSession
from agents.result import RunResultBase
from agents.tool import Tool
from strix.runtime.status import StatusSink
from strix.tools.mcp import (
@@ -67,6 +70,8 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
_MCP_PROMPT_WARMUP_TIMEOUT_SECONDS = 30.0
StreamEventSink = Callable[[str, Any], None]
# Receives the run's MCP connection roster as a list of non-secret status dicts
@@ -92,6 +97,19 @@ def _mcp_roster_payload(registry: McpRegistry) -> list[dict[str, Any]]:
]
def _mcp_prompt_roster(registry: McpRegistry) -> list[dict[str, Any]]:
"""The MCP roster rendered in agent prompts, with only verified counts."""
return [
{
"name": summary.name,
"purpose": summary.purpose,
"tool_count": summary.tool_count if summary.state == "catalog_ready" else None,
"state": summary.state,
}
for summary in registry.summaries()
]
def _record_mcp_connections(connection_names: list[str]) -> None:
"""Record which MCP servers this run configured, for the interfaces.
@@ -153,6 +171,7 @@ def _compose_root_instructions_override(
is_diff_scoped: bool,
interactive: bool,
system_prompt_context: dict[str, Any],
supports_images: bool,
) -> str | None:
if root_instructions_override is None:
return None
@@ -166,6 +185,7 @@ def _compose_root_instructions_override(
interactive=interactive,
system_prompt_context=system_prompt_context,
include_scope=False,
supports_images=supports_images,
)
return (
f"{base_instructions}\n\n"
@@ -197,6 +217,7 @@ async def run_strix_scan(
status_sink: StatusSink | None = None,
mcp_connection_requests: list[McpConnectionRequest] | None = None,
mcp_status_sink: McpStatusSink | None = None,
root_finish_tool: Tool = finish_scan,
) -> RunResultBase | None:
"""Run or resume one Strix scan against a sandbox.
@@ -219,6 +240,7 @@ async def run_strix_scan(
command-line default) it reads ``~/.strix/mcp-servers.json`` itself. Either
way the engine does the connecting, so the caller passes inert configs plus
metadata and never live sessions.
``root_finish_tool`` is the tool the root agent ends the run with.
"""
def report(phase: str) -> None:
@@ -265,6 +287,9 @@ async def run_strix_scan(
strict_tool_schemas = supports_strict_tool_schemas(resolved_model)
if not strict_tool_schemas:
logger.info("Sending non-strict tool schemas: %s caps strict tools", resolved_model)
supports_images = model_supports_images(resolved_model)
if not supports_images:
logger.info("Leaving out image tools: %s does not accept images", resolved_model)
if budget_policy not in ("stop", "pause"):
raise ValueError(f"unknown budget_policy: {budget_policy!r}")
@@ -272,6 +297,7 @@ async def run_strix_scan(
coordinator = AgentCoordinator()
coordinator.set_snapshot_path(agents_path)
coordinator.set_budget_policy(budget_policy)
coordinator.root_finish_tool = root_finish_tool.name
from strix.tools.coverage.tools import hydrate_coverage_from_disk
from strix.tools.notes.tools import hydrate_notes_from_disk
@@ -432,19 +458,13 @@ async def run_strix_scan(
_record_mcp_connections(mcp_registry.names())
report(
f"MCP: configured {len(mcp_registry)} connection(s); "
"warming them in the background"
"connecting and listing tools"
)
scope_context["mcp_available"] = True
scope_context["mcp_connections"] = [
{
"name": summary.name,
"purpose": summary.purpose,
"tool_count": summary.tool_count,
}
for summary in mcp_registry.summaries()
]
scope_context["mcp_connections"] = _mcp_prompt_roster(mcp_registry)
def _emit_mcp_status() -> None:
scope_context["mcp_connections"] = _mcp_prompt_roster(mcp_registry)
roster = _mcp_roster_payload(mcp_registry)
_persist_mcp_status(roster)
if mcp_status_sink is not None:
@@ -455,7 +475,11 @@ async def run_strix_scan(
mcp_registry.set_status_sink(_emit_mcp_status)
_emit_mcp_status()
mcp_registry.start_warmup(max_concurrency=6)
warmup_task = mcp_registry.start_warmup(max_concurrency=6)
await asyncio.wait(
{warmup_task},
timeout=_MCP_PROMPT_WARMUP_TIMEOUT_SECONDS,
)
except Exception:
logger.exception("Failed to configure user MCP servers; continuing without them")
@@ -468,6 +492,7 @@ async def run_strix_scan(
is_diff_scoped=is_diff_scoped,
interactive=interactive,
system_prompt_context=root_context,
supports_images=supports_images,
)
root_agent = build_strix_agent(
@@ -482,6 +507,8 @@ async def run_strix_scan(
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=root_context,
instructions_override=root_instructions,
supports_images=supports_images,
finish_tool=root_finish_tool,
)
if not is_resume:
@@ -501,6 +528,7 @@ async def run_strix_scan(
chat_completions_tools=chat_completions_tools,
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=scope_context,
supports_images=supports_images,
)
async def spawn_child_agent(**kwargs: Any) -> dict[str, Any]:
@@ -528,6 +556,7 @@ async def run_strix_scan(
"spawn_child_agent": spawn_child_agent,
"scan_targets": build_scan_targets(scan_config),
"max_context_images": settings.runtime.max_context_images,
"supports_images": supports_images,
}
root_session = open_agent_session(root_id, agents_db)
@@ -592,7 +621,7 @@ async def run_strix_scan(
)
if not interactive and result is not None:
final = getattr(result, "final_output", None)
# Lifecycle tools mark the root completed.
# Lifecycle tools mark the root completed.
async with coordinator._lock:
root_completed = coordinator.statuses.get(root_id) == "completed"
if not root_completed:
+2 -2
View File
@@ -199,13 +199,13 @@ async def enforce_image_budget(session: Session, max_images: int) -> bool:
return await _rewrite_session(session, _transform)
def scrub_images_from_items(items: list[Any]) -> list[Any]:
def scrub_images_from_items(items: list[Any], *, text: str = _INHERITED_IMAGE_TEXT) -> list[Any]:
"""Return a copy of ``items`` with every image block replaced by text."""
def _scrub(obj: Any) -> Any:
if isinstance(obj, dict):
if obj.get("type") == "input_image":
return {"type": "input_text", "text": _INHERITED_IMAGE_TEXT}
return {"type": "input_text", "text": text}
return {k: _scrub(v) for k, v in obj.items()}
if isinstance(obj, list):
return [_scrub(v) for v in obj]
@@ -90,6 +90,7 @@ type Model struct {
client *Client
width, height int
snapshot protocol.Snapshot
windowTitle string
input textarea.Model
viewport viewport.Model
viewportContent string
+2 -2
View File
@@ -66,7 +66,7 @@ func panelFocus(panel sidebarPanel) (focusMode, bool) {
func (m Model) sidebarPanels() []panelRect {
statsHeight, vulnHeight, mcpHeight, agentHeight := m.sidebarHeights()
top := m.viewerHeight()
top := m.sidebarTopHeight()
rects := []panelRect{{panelAgents, top, agentHeight}}
top += agentHeight
if vulnHeight > 0 {
@@ -83,7 +83,7 @@ func (m Model) sidebarPanels() []panelRect {
func (m Model) sidebarGap() int {
statsHeight, vulnHeight, mcpHeight, agentHeight := m.sidebarHeights()
return max(0, m.height-m.viewerHeight()-agentHeight-vulnHeight-mcpHeight-statsHeight)
return max(0, m.height-m.sidebarTopHeight()-agentHeight-vulnHeight-mcpHeight-statsHeight)
}
func (m Model) panelAt(y int) (panelRect, bool) {
@@ -355,3 +355,77 @@ func TestLonePanelHasNoControls(t *testing.T) {
t.Fatalf("controls should return once a second panel exists")
}
}
func TestTargetsBoxSitsUnderViewer(t *testing.T) {
m := panelsModel(t)
m.snapshot.Targets = []string{"https://example.com", "./app"}
m.snapshot.TargetCount = 2
m.resizeViewport()
_, sidebarWidth, _, _ := m.layout()
rows := strings.Split(ansi.Strip(m.sidebarView(sidebarWidth, m.height)), "\n")
if len(rows) != m.height {
t.Fatalf("sidebar renders %d rows, want %d", len(rows), m.height)
}
top := m.viewerHeight()
for i, want := range []string{"Targets 2", "▸ https://example.com", "▸ ./app"} {
if !strings.Contains(rows[top+1+i], want) {
t.Fatalf("row %d missing %q:\n%s", top+1+i, want, strings.Join(rows, "\n"))
}
}
if !strings.Contains(rows[panelRectOf(t, m, panelAgents).top+1], "Agents") {
t.Fatalf("agents panel not where its rect says:\n%s", strings.Join(rows, "\n"))
}
}
func TestTargetsBoxDropsOnShortTerminal(t *testing.T) {
m := panelsModel(t)
m.height = 19
m.snapshot.Targets = []string{"a", "b", "c", "d", "e"}
m.snapshot.TargetCount = 5
m.resizeViewport()
_, sidebarWidth, _, _ := m.layout()
if got := lipgloss.Height(m.sidebarView(sidebarWidth, m.height)); got > m.height {
t.Fatalf("sidebar renders %d rows on a %d-row terminal", got, m.height)
}
}
// The targets box only takes rows the agent tree would otherwise grow into, so
// the stats panel keeps the Caido URL and the other panels keep their heights.
func TestTargetsBoxNeverSqueezesPanels(t *testing.T) {
for height := 19; height <= 40; height++ {
m := panelsModel(t)
m.height = height
m.snapshot.Usage = map[string]any{"total_tokens": 1234, "cost": 0.5}
m.snapshot.CaidoURL = "http://127.0.0.1:8080"
m.resizeViewport()
stats, vuln, mcp, _ := m.sidebarHeights()
m.snapshot.Targets = []string{"a", "b", "c", "d", "e"}
m.snapshot.TargetCount = 5
m.resizeViewport()
gotStats, gotVuln, gotMcp, _ := m.sidebarHeights()
if gotStats != stats || gotVuln != vuln || gotMcp != mcp {
t.Fatalf("height %d: targets squeezed panels: stats %d->%d findings %d->%d mcp %d->%d",
height, stats, gotStats, vuln, gotVuln, mcp, gotMcp)
}
_, sidebarWidth, _, _ := m.layout()
if got := lipgloss.Height(m.sidebarView(sidebarWidth, m.height)); got != m.height {
t.Fatalf("height %d: sidebar renders %d rows", height, got)
}
}
}
func TestTargetsViewCountsTheTargetsItCannotShow(t *testing.T) {
m := panelsModel(t)
m.snapshot.Targets = []string{"a", "b", "c", "d", "e"}
m.snapshot.TargetCount = 5
for rows, want := range map[int]string{
2: "Targets 5\n+5 more",
3: "Targets 5\n▸ a\n+4 more",
6: "Targets 5\n▸ a\n▸ b\n▸ c\n▸ d\n+1 more",
9: "Targets 5\n▸ a\n▸ b\n▸ c\n▸ d\n+1 more",
} {
if got := ansi.Strip(m.targetsView(20, rows)); got != want {
t.Fatalf("%d rows:\n%s\nwant:\n%s", rows, got, want)
}
}
}
+13 -6
View File
@@ -421,7 +421,7 @@ func (m Model) setupComposer(width int) string {
inner := max(1, width-4)
body := m.highlightInputSelection(m.input.View())
body += "\n\n" + m.setupSummaryView(inner)
if targets := m.setupTargetsView(inner); targets != "" {
if targets := m.targetsView(inner, maxTargetRows); targets != "" {
body += "\n" + targets
}
return lipgloss.NewStyle().Width(max(1, width-2)).Padding(0, 1).
@@ -454,15 +454,22 @@ func (m Model) setupSummaryView(width int) string {
return truncate(strings.Join(chips, render.Dim().Render(" · ")), max(1, width))
}
// setupTargetsView lists what the scan is pointed at, once anything is queued.
func (m Model) setupTargetsView(width int) string {
if len(m.snapshot.Targets) == 0 {
// maxTargetRows fits the header, four targets and the count of the rest.
const maxTargetRows = 6
// targetsView lists what the scan is pointed at, once anything is queued, in at
// most maxRows rows. Targets that do not fit are counted on the last row.
func (m Model) targetsView(width, maxRows int) string {
if len(m.snapshot.Targets) == 0 || maxRows < 2 {
return ""
}
const visible = 4
total := max(m.snapshot.TargetCount, len(m.snapshot.Targets))
visible := min(len(m.snapshot.Targets), maxTargetRows-2, maxRows-1)
if visible < total {
visible = min(visible, maxRows-2)
}
rows := []string{render.Bold(green).Render("Targets") + render.Dim().Render(fmt.Sprintf(" %d", total))}
for _, target := range m.snapshot.Targets[:min(visible, len(m.snapshot.Targets))] {
for _, target := range m.snapshot.Targets[:visible] {
rows = append(rows, render.Col(dim).Render("▸ ")+render.Col(white).Render(truncate(target, max(1, width-2))))
}
if hidden := total - visible; hidden > 0 {
+35
View File
@@ -542,6 +542,15 @@ func (m Model) sidebarView(width, height int) string {
BorderForeground(dark).
Padding(0, 1).
Render(m.viewerBox(width - 4))}
if targetsHeight := m.targetsBoxHeight(); targetsHeight > 0 {
parts = append(parts, lipgloss.NewStyle().
Width(width-2).
Height(targetsHeight-2).
Border(lipgloss.RoundedBorder()).
BorderForeground(dark).
Padding(0, 1).
Render(m.targetsView(width-4, targetsHeight-2)))
}
agents := ""
if agentHeight > 1 {
agentRows := max(1, agentHeight-4)
@@ -592,6 +601,13 @@ func (m Model) sidebarView(width, height int) string {
}
func (m Model) sidebarHeights() (statsHeight, vulnHeight, mcpHeight, agentHeight int) {
_, statsHeight, vulnHeight, mcpHeight, agentHeight = m.sidebarLayout()
return
}
// sidebarLayout sizes the targets box and the panels. The targets box only
// takes rows no panel needs, so it never squeezes the panels below it.
func (m Model) sidebarLayout() (targetsHeight, statsHeight, vulnHeight, mcpHeight, agentHeight int) {
// Measure the stats panel the way its box will render it: a long model name
// wraps inside the sidebar, and counting only its newlines would size the
// box short and push the whole frame past the bottom of the terminal.
@@ -628,6 +644,14 @@ func (m Model) sidebarHeights() (statsHeight, vulnHeight, mcpHeight, agentHeight
}
}
spare := max(0, m.height-m.viewerHeight()-statsHeight-vulnHeight-mcpHeight-agentHeight)
// The box needs its border and two rows: the header and a target or the
// count of hidden ones. A zoomed panel keeps the whole sidebar.
if targets := m.targetsView(m.sidebarInnerWidth(), maxTargetRows); targets != "" && m.zoomedPanel == panelNone {
if fit := min(lipgloss.Height(targets)+2, spare); fit >= 4 {
targetsHeight = fit
spare -= fit
}
}
switch {
case m.zoomedPanel == panelFindings && vulnHeight > 1:
vulnHeight += spare
@@ -643,6 +667,17 @@ func (m Model) viewerHeight() int {
return strings.Count(m.viewerView(m.viewerContentWidth()), "\n") + 3
}
// sidebarTopHeight is the rows above the panels: the viewer box and, when the
// run has targets, the targets box under it.
func (m Model) sidebarTopHeight() int {
return m.viewerHeight() + m.targetsBoxHeight()
}
func (m Model) targetsBoxHeight() int {
targetsHeight, _, _, _, _ := m.sidebarLayout()
return targetsHeight
}
func (m Model) sidebarInnerWidth() int {
_, sidebarWidth, _, _ := m.layout()
if sidebarWidth == 0 {
+27
View File
@@ -2,10 +2,13 @@ package app
import (
"encoding/json"
"fmt"
"os"
"strings"
"time"
tea "github.com/charmbracelet/bubbletea"
"github.com/charmbracelet/x/ansi"
"github.com/usestrix/strix/tui/internal/protocol"
"github.com/usestrix/strix/tui/internal/render"
)
@@ -56,6 +59,10 @@ func (m *Model) handleEnvelope(envelope protocol.Envelope) tea.Cmd {
// resize (not just refresh): status-row visibility changes the chat height.
m.resizeViewport()
m.resizeVulnerabilityViewport()
if title := windowTitle(m.snapshot.Targets); title != m.windowTitle {
m.windowTitle = title
return setTerminalTitle(title)
}
case "collection_bootstrap":
return m.handleCollectionBootstrap(envelope.Payload)
case "collection_delta":
@@ -118,6 +125,26 @@ func (m *Model) handleEnvelope(envelope protocol.Envelope) tea.Cmd {
return nil
}
// tea.SetWindowTitle only sends OSC 2 (window title); iTerm tabs show the OSC 1
// icon name, which shells usually set, so send OSC 0 to set both.
func setTerminalTitle(title string) tea.Cmd {
return func() tea.Msg {
_, _ = os.Stdout.WriteString(ansi.SetIconNameWindowTitle(title))
return nil
}
}
func windowTitle(targets []string) string {
switch len(targets) {
case 0:
return "strix"
case 1:
return targets[0]
default:
return fmt.Sprintf("%s +%d", targets[0], len(targets)-1)
}
}
func (m *Model) consumeMessages(messages []protocol.Message, setupMode bool) {
if m.seenMessages == nil {
m.seenMessages = map[string]bool{}
+21 -51
View File
@@ -15,11 +15,7 @@ from typing import TYPE_CHECKING, Any
from agents.model_settings import ModelSettings
from agents.models.interface import ModelTracing
from openai.types.responses import (
ResponseOutputMessage,
ResponseOutputText,
ResponseReasoningItem,
)
from openai.types.responses import ResponseOutputMessage, ResponseOutputText
from strix.config import load_settings
from strix.config.models import StrixProvider
@@ -164,22 +160,22 @@ def _content_text(content: Any) -> str:
return ""
def _truncate(text: str, limit: int | None) -> str:
return text if limit is None or len(text) <= limit else f"{text[:limit]}\n[truncated]"
def _truncate(text: str, limit: int) -> str:
return text if len(text) <= limit else f"{text[:limit]}\n[truncated]"
def _serialize_item(item: Any, max_chars: int | None = _TOOL_OUTPUT_MAX_CHARS) -> str:
def _serialize_item(item: Any) -> str:
if not isinstance(item, dict):
return str(item)
item_type = item.get("type")
role = item.get("role")
if item_type == "function_call":
args = _truncate(str(item.get("arguments", "")), max_chars)
args = _truncate(str(item.get("arguments", "")), _TOOL_OUTPUT_MAX_CHARS)
return f"[tool_call {item.get('name', '?')}] {args}"
if item_type == "function_call_output":
output = item.get("output")
text = output if isinstance(output, str) else _content_text(output)
return f"[tool_result] {_truncate(text, max_chars)}"
return f"[tool_result] {_truncate(text, _TOOL_OUTPUT_MAX_CHARS)}"
if item_type == "reasoning":
return ""
if role or item_type == "message":
@@ -187,30 +183,8 @@ def _serialize_item(item: Any, max_chars: int | None = _TOOL_OUTPUT_MAX_CHARS) -
return ""
def _serialize_items(items: list[Any], max_chars: int | None = _TOOL_OUTPUT_MAX_CHARS) -> str:
return "\n".join(s for s in (_serialize_item(item, max_chars) for item in items) if s)
class CompactionNeededError(Exception):
"""The next model request would overrun the model's usable context window."""
def _context_budget(model: str) -> int:
context = load_settings().context
reserve = max(context.compact_buffer_tokens, output_limit(model))
return max(context.keep_tokens, context_window(model) - reserve)
def _request_tokens(model: str, instructions: str, tools_text: str, items: list[Any]) -> int:
text = "\n".join((instructions, tools_text, _serialize_items(items, max_chars=None)))
return count_tokens(model, text)
def needs_compaction(model: str, instructions: str, tools_text: str, items: list[Any]) -> bool:
"""Whether a request carrying ``items`` would overrun ``model``'s context budget."""
if not load_settings().context.auto_compact:
return False
return _request_tokens(model, instructions, tools_text, items) > _context_budget(model)
def _serialize_items(items: list[Any]) -> str:
return "\n".join(s for s in (_serialize_item(item) for item in items) if s)
def _is_tool_call(item: Any) -> bool:
@@ -314,22 +288,15 @@ def _checkpoint_item(summary: str) -> dict[str, Any]:
def _extract_text(response: ModelResponse) -> str:
parts: list[str] = []
reasoning: list[str] = []
for item in response.output:
if isinstance(item, ResponseOutputMessage):
parts.extend(
chunk.text
for chunk in item.content
if isinstance(chunk, ResponseOutputText) and chunk.text
)
elif isinstance(item, ResponseReasoningItem):
reasoning.extend(summary.text for summary in item.summary if summary.text)
if parts or not reasoning:
return "".join(parts)
# Reasoning models can spend the whole output budget thinking and return no
# message; their reasoning is a degraded but usable summary.
logger.warning("compaction summary had no output text; using reasoning text")
return "\n".join(reasoning)
if not isinstance(item, ResponseOutputMessage):
continue
parts.extend(
chunk.text
for chunk in item.content
if isinstance(chunk, ResponseOutputText) and chunk.text
)
return "".join(parts)
async def _summarize(model: str, prompt: str, max_tokens: int) -> str | None:
@@ -391,8 +358,11 @@ async def maybe_compact(
if len(items) < _MIN_ITEMS_TO_COMPACT:
return False
used = _request_tokens(model, instructions, tools_text, items)
if not force and used <= _context_budget(model):
window = context_window(model)
reserve = max(context.compact_buffer_tokens, output_limit(model))
budget = max(context.keep_tokens, window - reserve)
used = count_tokens(model, "\n".join((instructions, tools_text, _serialize_items(items))))
if not force and used <= budget:
return False
split = _select_split(model, items, context.keep_tokens)
+47 -4
View File
@@ -159,11 +159,33 @@ class HttpReply:
status_code: int | None = None
headers: dict[str, str] | None = None
upstream_provider: str | None = None
upstream_error_type: str | None = None
_http_reply: ContextVar[HttpReply | None] = ContextVar("strix_llm_http_reply", default=None)
def record_upstream_provider(provider: object, error_type: object = None) -> None:
"""Remember which provider behind a gateway served the attempt in flight."""
reply = _http_reply.get()
if reply is None:
return
if isinstance(provider, str) and provider:
reply.upstream_provider = provider
if isinstance(error_type, str) and error_type:
reply.upstream_error_type = error_type
def _upstream_details(reply: HttpReply | None) -> list[tuple[str, object]]:
if reply is None:
return []
return [
("upstream_provider", reply.upstream_provider),
("upstream_error_type", reply.upstream_error_type),
]
async def record_http_reply(response: Response) -> None:
"""httpx ``response`` event hook: remember the reply for the attempt in flight.
@@ -772,6 +794,7 @@ def _litellm_success(
("request", _litellm_request_details(kwargs)),
("response", _litellm_response_details(response)),
("litellm", _litellm_hidden_details(hidden, slo)),
*_upstream_details(_http_reply.get()),
),
)
@@ -821,6 +844,7 @@ def _litellm_failure(
("request", _litellm_request_details(kwargs)),
("error", error or None),
("litellm", _litellm_hidden_details(hidden, slo)),
*_upstream_details(_http_reply.get()),
),
)
@@ -894,15 +918,18 @@ def _observe_sdk_shared_http_client() -> None:
def _log_line_sink(event: LlmRequestEvent) -> None:
details = event.details or {}
level = logging.DEBUG if event.outcome == "success" else logging.WARNING
logger.log(
level,
"llm_request route=%s provider=%s model=%s host=%s outcome=%s status=%s "
"request_id=%s response_id=%s stream=%s duration_ms=%d ttft_ms=%s "
"llm_request route=%s provider=%s upstream=%s upstream_error=%s model=%s host=%s "
"outcome=%s status=%s request_id=%s response_id=%s stream=%s duration_ms=%d ttft_ms=%s "
"req_bytes=%s res_bytes=%s finish=%s "
"in=%s out=%s cached=%s cost=%s agent=%s attempt=%d%s",
event.route,
event.provider or "-",
details.get("upstream_provider") or "-",
details.get("upstream_error_type") or "-",
event.model,
event.api_host or "-",
event.outcome,
@@ -1070,6 +1097,7 @@ class RequestLoggingModel(Model):
details=merge_details(
("request", request.details),
("response", _openai_response_details(response, raw_response)),
*_upstream_details(reply),
),
)
status, request_id = _openai_error_fields(exc)
@@ -1082,13 +1110,14 @@ class RequestLoggingModel(Model):
outcome="error",
status_code=status,
provider_request_id=request_id or base.provider_request_id,
error_type=type(exc).__name__,
error_type=_cancel_reason(exc) or type(exc).__name__,
error_message=_abandonment_message(exc) or clean_error_message(exc),
response_bytes=_exception_body_size(exc, status),
response_headers=headers_from_response(error_headers) or base.response_headers,
details=merge_details(
("request", request.details),
("error", _exception_details(exc)),
*_upstream_details(reply),
),
)
@@ -1318,9 +1347,23 @@ def _is_abandonment(exc: BaseException | None) -> bool:
return isinstance(exc, asyncio.CancelledError | GeneratorExit)
# Prefixes the reason a stream timeout gives when it cancels an attempt.
CANCEL_REASON_PREFIX = "strix:"
def _cancel_reason(exc: BaseException) -> str | None:
"""The stream timeout that cancelled the attempt, e.g. ``stream_idle_timeout``."""
if isinstance(exc, asyncio.CancelledError) and exc.args:
message = str(exc.args[0])
if message.startswith(CANCEL_REASON_PREFIX):
return message.removeprefix(CANCEL_REASON_PREFIX)
return None
def _abandonment_message(exc: BaseException) -> str | None:
if isinstance(exc, asyncio.CancelledError):
return "attempt cancelled before the reply was consumed (stream idle timeout or shutdown)"
reason = f": {exc.args[0]}" if exc.args else ""
return f"attempt cancelled before the reply was consumed{reason}"
if isinstance(exc, GeneratorExit):
return "stream closed by the caller before it finished"
return None
+15 -2
View File
@@ -6,6 +6,7 @@ from pathlib import Path
from typing import TypeGuard
import yaml
from jinja2 import Environment, StrictUndefined
from strix.utils.resource_paths import get_strix_resource_path
@@ -13,6 +14,14 @@ from strix.utils.resource_paths import get_strix_resource_path
logger = logging.getLogger(__name__)
_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n(?P<body>.*?)\n---\s*\n", re.DOTALL)
# Only skills with `template: jinja` in their frontmatter are rendered; the rest hold
# literal `{{ }}` payloads.
_SKILL_TEMPLATES = Environment(
autoescape=False, # noqa: S701 # nosec B701 - prompts, not HTML
undefined=StrictUndefined,
trim_blocks=True,
lstrip_blocks=True,
)
_INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordination", "analysis"})
_ROOT_SKILL_CATEGORY = "root"
@@ -261,7 +270,7 @@ def _candidate_skill_files(skill_name: str) -> list[Path]:
return _bare_skill_files(skill_name)
def load_skills(skill_names: list[str]) -> dict[str, str]:
def load_skills(skill_names: list[str], *, supports_images: bool = True) -> dict[str, str]:
"""Load skill markdown bodies (frontmatter stripped) by name.
Skill files live at ``strix/skills/<category>/<name>.md`` (or any
@@ -291,7 +300,11 @@ def load_skills(skill_names: list[str]) -> dict[str, str]:
continue
var_name = skill_name.split("/")[-1]
_, skill_body = _parse_skill_content(content, file_path)
metadata, skill_body = _parse_skill_content(content, file_path)
if metadata.get("template") == "jinja":
skill_body = _SKILL_TEMPLATES.from_string(skill_body).render(
supports_images=supports_images
)
skill_content[var_name] = skill_body
logger.debug("Loaded skill: %s -> %s", skill_name, var_name)
_track_skill_loaded(var_name, file_path)
+14 -3
View File
@@ -115,7 +115,8 @@ not just "upgrade <vulnerable pkg> to <fixed>".
### Usage / reachability analysis (required for every dependency CVE)
For every CVE you are about to report, run a static usage analysis and record
the result in the structured `reachability` + `reachability_evidence` fields.
the result in the structured `reachability` + `reachability_evidence` fields,
and report every usage site you find as a `code_locations` entry (see Reporting).
The level is an **evidence ladder, never an exploitability verdict** — claim
only what you proved, and cite the proof. It never changes severity (that is
`advisory_cvss` alone); it exists so the reader can prioritize.
@@ -220,8 +221,18 @@ findings and rejects empty PoC fields):
GHSA, look up the mapped CVE; if there is genuinely no CVE, do not report it
with this tool.
- There are no PoC fields — `create_dependency_report` does not take
`poc_description` / `poc_script_code` / `code_locations`. The proof lives in
`description` and `technical_analysis` (scanner output + advisory).
`poc_description` / `poc_script_code`. The proof lives in `description` and
`technical_analysis` (scanner output + advisory).
- **You MUST report usage sites in `code_locations`.** Add one entry for each
import of the package and each call of an affected symbol that your usage
analysis found. Every `file:line` you cite in `reachability_evidence` must
also be a `code_locations` entry. Also add the line in `manifest_path` that
pins the vulnerable version. Each entry needs the repo-relative `file`, the
exact `start_line` / `end_line`, the verbatim `snippet` at those lines, and a
short `label` (e.g. "imports js-yaml", "calls yaml.load on CLI input"). Read
the real file to confirm the lines; never guess. When the package is
`not_imported`, the manifest line is the only entry. If an entry carries
`fix_before` / `fix_after`, also fill `fix_verification`.
- **Always fill the structured dependency fields** (they power the dedicated
dependency-report card; do not leave them only in free-text):
- `package_name` — `PkgName` (required).
+15 -6
View File
@@ -1,6 +1,7 @@
---
name: agent_browser
description: agent-browser CLI for headless Chrome via shell. Snapshot-and-ref workflow, click/fill/extract, screenshots, multi-tab, multi-session, network mocking. Pre-installed in the sandbox; invoke via exec_command.
template: jinja
---
@@ -260,8 +261,9 @@ only for simple expressions.
### Screenshot
`agent-browser screenshot` writes a PNG to disk in the sandbox. The
shell command alone does **not** put the image into your context —
`agent-browser screenshot` writes a PNG to disk in the sandbox.
{% if supports_images %}
The shell command alone does **not** put the image into your context —
chain it with the SDK ``view_image`` tool to actually see it:
```bash
@@ -275,6 +277,12 @@ the full path on stdout — pass that to ``view_image``). If you need a
specific filename, keep it inside that directory or a sibling hidden
dir under ``/workspace``. Never write screenshots to ``/tmp`` —
``view_image`` rejects anything outside the workspace root.
{% else %}
You are running on a text-only model and cannot view images. Screenshots
are only useful to you as a saved file (for example, login evidence).
Drive the page from `snapshot -i` refs, `eval` for any DOM/JS state you
need to read, and `text @ref` / `get text` for content extraction.
{% endif %}
```bash
agent-browser screenshot # path printed on stdout
@@ -283,6 +291,7 @@ agent-browser screenshot --full # full scroll height
agent-browser screenshot --annotate # numbered labels + legend keyed to snapshot refs
```
{% if supports_images %}
`--annotate` is designed for multimodal models: each label `[N]` maps
to ref `@eN`. Take the annotated screenshot, then ``view_image`` it,
and you can correlate visual layout with snapshot refs.
@@ -294,10 +303,10 @@ layout questions, captchas, custom widgets where the accessibility
tree is incomplete).
If ``view_image`` errors back at you (rejected image, "vision not
supported", or similar), you are running on a text-only model — stop
calling it and stop taking screenshots. Drive the page entirely from
`snapshot -i` refs, `eval` for any DOM/JS state you need to read, and
`text @ref` / `get text` for content extraction.
supported", or similar), you can drive the page from `snapshot -i`
refs, `eval` for any DOM/JS state you need to read, and `text @ref`
or `get text` for content extraction.
{% endif %}
### Handle multiple pages via tabs
+12
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import contextlib
import functools
import logging
import os
import sys
@@ -78,6 +79,16 @@ class _StdoutQuietFilter(logging.Filter):
)
class _PtyThresholdFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
return "PTY process count reached warning threshold" not in record.getMessage()
@functools.cache # install once; tests reset via ``cache_clear()``
def _silence_pty_threshold_warning() -> None:
logging.getLogger("agents.sandbox.sandboxes.docker").addFilter(_PtyThresholdFilter())
def configure_dependency_logging() -> None:
"""Quiet dependency logging/warnings that obscure Strix scan logs."""
litellm = sys.modules.get("litellm")
@@ -89,6 +100,7 @@ def configure_dependency_logging() -> None:
logging.getLogger("asyncio").propagate = False
warnings.filterwarnings("ignore", category=RuntimeWarning, module="asyncio")
_route_unraisable_to_log()
_silence_pty_threshold_warning()
def _route_unraisable_to_log() -> None:
+2 -1
View File
@@ -683,7 +683,8 @@ async def agent_finish(
{
"success": False,
"error": (
"agent_finish is for subagents. Root/main agents must call finish_scan instead"
"agent_finish is for subagents. Root/main agents must call "
f"{coordinator.root_finish_tool} instead"
),
},
ensure_ascii=False,
+2 -2
View File
@@ -24,12 +24,12 @@ async def load_skill(ctx: RunContextWrapper, skills: list[str]) -> str:
Max 5. Names match the bare files under
``strix/skills/<category>/<name>.md``.
"""
del ctx
requested = list(skills or [])
err = validate_requested_skills(requested)
if err:
return f"load_skill: {err}"
contents = load_skills(requested)
context = ctx.context if isinstance(ctx.context, dict) else {}
contents = load_skills(requested, supports_images=context.get("supports_images", True))
if not contents:
return "load_skill: no content loaded for requested skills."
sections = [f"## Skill: {name}\n\n{body}" for name, body in contents.items()]
+26 -3
View File
@@ -24,6 +24,7 @@ automatically.
from __future__ import annotations
import asyncio
import json
from typing import TYPE_CHECKING, Any
@@ -46,6 +47,7 @@ def _registry_from_ctx(ctx: RunContextWrapper) -> McpRegistry | None:
_NO_CONNECTIONS = "No MCP connections are configured for this run."
_MCP_DISCOVERY_WAIT_SECONDS = 30.0
def _unknown_connection(connection: str, registry: McpRegistry) -> str:
@@ -61,11 +63,13 @@ def _format_tool(tool: MCPTool) -> str:
@function_tool(timeout=60)
async def list_mcps(ctx: RunContextWrapper) -> dict[str, Any]:
"""List the MCP connections available this run, so you can discover them.
"""List available MCP connections, loading each connection's catalog on first use.
Read-only. Returns one entry per connection with its ``id`` (the exact name
you pass to the other MCP tools), ``name``, ``description``, and
``tool_count``. The response does not include tool schemas. Call
``tool_count`` (null while the catalog is unknown). Waits at most 30 seconds
for catalogs; unfinished catalogs keep loading in the background.
The response does not include tool schemas. Call
``search_mcp_tools`` next. Then call ``get_mcp_tool_schema`` for one selected
tool before you call ``call_mcp``. Returns an empty ``connections`` list when
the run has no MCP connections.
@@ -73,6 +77,23 @@ async def list_mcps(ctx: RunContextWrapper) -> dict[str, Any]:
registry = _registry_from_ctx(ctx)
if registry is None or not registry:
return {"connections": []}
pending = [
entry
for name in registry.names()
if (entry := registry.get(name)) is not None
and entry.state != "catalog_ready"
and (entry.session is None or not entry.session.is_dead)
]
if pending:
await asyncio.wait(
{
asyncio.gather(
*(entry.ensure_catalog() for entry in pending),
return_exceptions=True,
)
},
timeout=_MCP_DISCOVERY_WAIT_SECONDS,
)
dead_by_name = {status.name: status.dead for status in registry.statuses()}
return {
"connections": [
@@ -80,7 +101,7 @@ async def list_mcps(ctx: RunContextWrapper) -> dict[str, Any]:
"id": summary.name,
"name": summary.name,
"description": summary.purpose,
"tool_count": summary.tool_count,
"tool_count": summary.tool_count if summary.state == "catalog_ready" else None,
"dead": dead_by_name.get(summary.name, False),
"state": summary.state,
}
@@ -288,9 +309,11 @@ async def call_mcp(
"Call search_mcp_tools, then get_mcp_tool_schema."
)
session = await entry.ensure_connected()
context = ctx.context if isinstance(ctx.context, dict) else {}
return await session.dispatch(
tool,
arguments or {},
label=namespaced_tool_name(connection, tool),
result_transform=entry.result_transform,
supports_images=context.get("supports_images", True),
)
+16
View File
@@ -32,6 +32,7 @@ from agents.mcp import (
)
from mcp.client.stdio import stdio_client
from mcp.shared._httpx_utils import create_mcp_http_client
from mcp.types import TextContent
from strix.tools.mcp.failures import HttpStatusRecorder
from strix.tools.mcp.session import McpConnectionUnavailableError, SupervisedMcpSession
@@ -183,6 +184,18 @@ def _build_server(config: McpConnectionConfig) -> BuiltMcpServer:
)
def _without_images(result: Any) -> Any:
content = [
TextContent(
type="text", text=f"[{item.mimeType} image omitted: this model cannot view images]"
)
if item.type == "image"
else item
for item in result.content
]
return result.model_copy(update={"content": content})
def _mcp_result_to_tool_output(server: MCPServer, result: Any) -> Any:
"""Serialize a ``CallToolResult`` to a tool output, mirroring the agents SDK.
@@ -217,6 +230,7 @@ async def dispatch_mcp_call(
*,
label: str,
result_transform: ResultTransform | None = None,
supports_images: bool = True,
) -> Any:
"""Run one MCP tool call and convert its result to a tool output.
@@ -233,6 +247,8 @@ async def dispatch_mcp_call(
corrupt the content the agent receives).
"""
result = await server.call_tool(tool_name, arguments)
if not supports_images:
result = _without_images(result)
if result_transform is not None:
return result_transform(label, result.model_dump(mode="json"))
tool_output = _mcp_result_to_tool_output(server, result)
+5 -5
View File
@@ -432,19 +432,19 @@ class McpRegistry:
entry.set_status_sink(sink)
def start_warmup(self, *, max_concurrency: int = 6) -> asyncio.Task[None]:
"""Connect every configured entry in the background with a fixed bound."""
"""Connect and list every configured entry with a fixed concurrency bound."""
if self._warmup_task is not None:
return self._warmup_task
async def warm() -> None:
semaphore = asyncio.Semaphore(max(1, max_concurrency))
async def connect(entry: McpConnectionEntry) -> None:
async def load_catalog(entry: McpConnectionEntry) -> None:
async with semaphore:
with contextlib.suppress(McpConnectionUnavailableError):
await entry.ensure_connected()
with contextlib.suppress(Exception):
await entry.ensure_catalog()
await asyncio.gather(*(connect(entry) for entry in self._entries.values()))
await asyncio.gather(*(load_catalog(entry) for entry in self._entries.values()))
self._warmup_task = asyncio.create_task(warm(), name="mcp-warmup")
return self._warmup_task
+2
View File
@@ -365,6 +365,7 @@ class SupervisedMcpSession:
*,
label: str,
result_transform: ResultTransform | None = None,
supports_images: bool = True,
) -> Any:
"""Run one tool call with bounded retries for transient session failures.
@@ -382,6 +383,7 @@ class SupervisedMcpSession:
arguments,
label=label,
result_transform=result_transform,
supports_images=supports_images,
)
outcome = await self._run_job(job, phase="call")
+6 -1
View File
@@ -13,6 +13,7 @@ from datetime import datetime
from typing import TYPE_CHECKING, Any, Literal
from agents import RunContextWrapper, function_tool
from caido_sdk_client import BaseError
from strix.runtime.caido_handle import CaidoBootstrapHandle
from strix.tools.nullish import clean_optional
@@ -124,7 +125,11 @@ def _no_client() -> str:
def _err(name: str, exc: Exception) -> str:
logger.exception("%s failed", name)
# Caido-side failures (timeouts, bad filters, ...) are returned to the model, not our bugs.
if isinstance(exc, BaseError):
logger.warning("%s failed", name, exc_info=True)
else:
logger.exception("%s failed", name)
return json.dumps(
{"success": False, "error": f"{name} failed: {exc}"},
ensure_ascii=False,
+36 -1
View File
@@ -1883,7 +1883,7 @@ def _build_dependency_evidence(
return evidence
async def _do_create_dependency( # noqa: PLR0912
async def _do_create_dependency( # noqa: PLR0912, PLR0915
*,
title: str,
description: str,
@@ -1905,6 +1905,8 @@ async def _do_create_dependency( # noqa: PLR0912
manifest_path: str | None = None,
reachability: str = "unknown",
reachability_evidence: str | None = None,
code_locations: list[dict[str, Any]] | None = None,
fix_verification: str | None = None,
contextual_cvss_breakdown: dict[str, str] | None = None,
contextual_cvss_reasoning: str | None = None,
agent_id: str | None = None,
@@ -1947,6 +1949,16 @@ async def _do_create_dependency( # noqa: PLR0912
if manifest_err:
errors.append(manifest_err)
parsed_locations = _normalize_code_locations(code_locations)
if parsed_locations:
errors.extend(_validate_code_locations(parsed_locations))
elif code_locations:
errors.append(
"code_locations were dropped as unusable - every location needs a relative "
"'file' and an integer 'start_line'"
)
errors.extend(_validate_fix_verification(parsed_locations, fix_verification))
reachability = (reachability or "unknown").strip().lower()
if reachability not in _VALID_REACHABILITY:
errors.append(
@@ -2053,6 +2065,8 @@ async def _do_create_dependency( # noqa: PLR0912
cvss=cvss_score if advisory_cvss is not None else None,
cve=parsed_cve,
cwe=cwe,
code_locations=parsed_locations,
fix_verification=fix_verification,
finding_class="dependency_cve",
dependency_metadata=dependency_metadata,
agent_id=agent_id if isinstance(agent_id, str) else None,
@@ -2107,6 +2121,8 @@ async def create_dependency_report(
dependency_path: str | None = None,
reachability: str = "unknown",
reachability_evidence: str | None = None,
code_locations: list[dict[str, Any]] | None = None,
fix_verification: str | None = None,
contextual_cvss_breakdown: dict[str, str] | None = None,
contextual_cvss_reasoning: str | None = None,
) -> str:
@@ -2217,6 +2233,23 @@ async def create_dependency_report(
is off in production), and say who controls the input. State
it plainly when no entry point reaches the sink — that is the
most useful result a reader can get.
code_locations: Where application code imports or calls the package —
the evidence behind your ``reachability`` claim. List of dicts,
same shape as ``create_vulnerability_report``::
{"file": "src/api/client.ts", "start_line": 14, "end_line": 14,
"snippet": "const x = require('pkg')", "label": "imports it"}
Cite the repo-relative ``file`` and the exact 1-based
``start_line`` where ``snippet`` actually sits — verify against
the real file, never guess. A location may also propose
the fix itself: ``fix_before`` (verbatim current code, e.g. the
manifest pin line) + ``fix_after`` (same lines with the fixed
version) — that combination requires ``fix_verification``.
fix_verification: Required whenever any ``code_locations`` entry
carries ``fix_after``. Same bar as a normal finding: security
closure, bypass review, preserved behavior, and how each was
checked.
contextual_cvss_breakdown: **Required.** Full CVSS v3.1 rating of this
CVE **in this codebase** — the same 8-metric object as
``create_vulnerability_report``'s ``cvss_breakdown``:
@@ -2277,6 +2310,8 @@ async def create_dependency_report(
manifest_path=manifest_path,
reachability=reachability,
reachability_evidence=reachability_evidence,
code_locations=code_locations,
fix_verification=fix_verification,
contextual_cvss_breakdown=contextual_cvss_breakdown,
contextual_cvss_reasoning=contextual_cvss_reasoning,
agent_id=agent_id,
+12
View File
@@ -115,3 +115,15 @@ def test_function_tools_are_result_bounded() -> None:
by_name = {t.name: t for t in agent.tools}
assert getattr(by_name["think"], "_strix_bounded", False) is True
@pytest.mark.parametrize(("supports_images", "has_view_image"), [(True, True), (False, False)])
def test_view_image_is_left_out_for_text_only_models(
supports_images: bool, has_view_image: bool
) -> None:
agent = factory.build_strix_agent(is_root=True, supports_images=supports_images)
filesystem = agent.capabilities[0]
filesystem.bind(cast("Any", object()))
view_image = next(tool for tool in filesystem.tools() if tool.name == "view_image")
assert cast("FunctionTool", view_image).is_enabled is has_view_image
+9
View File
@@ -59,6 +59,15 @@ def test_registered_tools_appear_before_lifecycle_tool() -> None:
assert child_names[-2:] == ["extra", "agent_finish"]
def test_root_ends_with_the_given_finish_tool() -> None:
root = factory.build_strix_agent(is_root=True, finish_tool=_tool("finish_pr_review"))
root_names = [t.name for t in root.tools]
assert root_names[-1] == "finish_pr_review"
assert "finish_scan" not in root_names
def test_per_call_extra_tools_stack_with_registry() -> None:
factory.register_agent_tools(_tool("registered"))
+1 -43
View File
@@ -6,8 +6,6 @@ from types import SimpleNamespace
from typing import TYPE_CHECKING, Any
import pytest
from agents.extensions.models.litellm_model import InternalChatCompletionMessage
from agents.models.chatcmpl_converter import Converter
from litellm.exceptions import BadRequestError, ContextWindowExceededError, RateLimitError
from openai.types.responses import ResponseOutputMessage, ResponseOutputText
@@ -138,9 +136,7 @@ def test_select_split_handles_parallel_calls(monkeypatch: pytest.MonkeyPatch) ->
assert not _has_orphan_tool_output(items[split:])
def _patch_budget(
monkeypatch: pytest.MonkeyPatch, *, keep_tokens: int, window: int
) -> ContextSettings:
def _patch_budget(monkeypatch: pytest.MonkeyPatch, *, keep_tokens: int, window: int) -> None:
monkeypatch.setattr(compaction, "count_tokens", lambda _m, t: len(t))
monkeypatch.setattr(compaction, "context_window", lambda _m: window)
monkeypatch.setattr(compaction, "output_limit", lambda _m: 0)
@@ -154,7 +150,6 @@ def _patch_budget(
llm=SimpleNamespace(api_key=None, api_base=None, timeout=1, extra_headers=None),
)
monkeypatch.setattr(compaction, "load_settings", lambda: settings)
return context
def _model_response(text: str) -> Any:
@@ -183,20 +178,6 @@ def _patch_summary(
monkeypatch.setattr(compaction, "StrixProvider", FakeProvider)
def test_needs_compaction_counts_full_tool_outputs(monkeypatch: pytest.MonkeyPatch) -> None:
_patch_budget(monkeypatch, keep_tokens=50, window=10_000)
items = [_call("c1"), _output("c1", "x" * 20_000)]
assert compaction.needs_compaction("m", "", "", items) is True
assert compaction.needs_compaction("m", "", "", _turns(2)) is False
def test_needs_compaction_respects_auto_compact_off(monkeypatch: pytest.MonkeyPatch) -> None:
_patch_budget(monkeypatch, keep_tokens=50, window=10_000).auto_compact = False
assert compaction.needs_compaction("m", "", "", [_output("c1", "x" * 20_000)]) is False
@pytest.mark.asyncio
async def test_maybe_compact_noop_when_within_budget(monkeypatch: pytest.MonkeyPatch) -> None:
_patch_budget(monkeypatch, keep_tokens=50, window=1_000_000)
@@ -261,29 +242,6 @@ async def test_summarize_routes_through_provider_with_settings(
assert settings.max_tokens == 64
@pytest.mark.asyncio
async def test_summarize_falls_back_to_reasoning_when_no_output_text(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_patch_budget(monkeypatch, keep_tokens=30, window=4_000)
message = InternalChatCompletionMessage(
role="assistant", content=None, reasoning_content="REASONED SUMMARY"
)
response = SimpleNamespace(output=Converter.message_to_output_items(message))
class FakeModel:
async def get_response(self, **_kwargs: Any) -> Any:
return response
class FakeProvider:
def get_model(self, _model_name: str | None) -> Any:
return FakeModel()
monkeypatch.setattr(compaction, "StrixProvider", FakeProvider)
assert await compaction._summarize("m", "p", 64) == "REASONED SUMMARY"
def test_fit_to_tokens_truncates_oversized_text(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(compaction, "count_tokens", lambda _m, t: len(t))
text = "x" * 10_000
+65
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import json
import uuid
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any
@@ -389,3 +390,67 @@ def test_openrouter_request_carries_agent_session_id() -> None:
assert body()["session_id"] == session_id
finally:
request_log.reset_call_context(token)
def _openrouter_config() -> Any:
_install_openrouter_stream_cost_capture()
config = ProviderConfigManager.get_provider_chat_config(
model="z-ai/glm-5.3", provider=LlmProviders.OPENROUTER
)
assert config is not None
return config
def test_openrouter_records_upstream_of_stream_and_mid_stream_error() -> None:
handler = _openrouter_config().get_model_response_iterator(
streaming_response=iter([]), sync_stream=True
)
reply = request_log.HttpReply()
token = request_log._http_reply.set(reply)
try:
handler.chunk_parser(
{
"id": "gen-a",
"created": 1,
"model": "z-ai/glm-5.3",
"provider": "Relace",
"choices": [{"index": 0, "delta": {"content": "x"}}],
}
)
assert reply.upstream_provider == "Relace"
assert reply.upstream_error_type is None
with pytest.raises(Exception, match="incomplete tool call"):
handler.chunk_parser(
{
"id": "gen-a",
"provider": "Relace",
"error": {
"code": 400,
"message": "Generation stopped with an incomplete tool call.",
"metadata": {"error_type": "invalid_request"},
},
}
)
finally:
request_log._http_reply.reset(token)
assert reply.upstream_provider == "Relace"
assert reply.upstream_error_type == "invalid_request"
def test_openrouter_records_upstream_of_pre_stream_error() -> None:
body = {
"error": {
"message": "Provider returned error",
"code": 400,
"metadata": {"raw": "...", "provider_name": "InferenceNet"},
}
}
reply = request_log.HttpReply()
token = request_log._http_reply.set(reply)
try:
_openrouter_config().get_error_class(json.dumps(body), 400, {})
_openrouter_config().get_error_class("<html>502 Bad Gateway</html>", 502, {})
finally:
request_log._http_reply.reset(token)
assert reply.upstream_provider == "InferenceNet"
assert reply.upstream_error_type is None
+58 -137
View File
@@ -24,7 +24,6 @@ from strix.core.execution import (
notify_parent_on_terminal,
)
from strix.core.sessions import seed_initial_input
from strix.llm.compaction import CompactionNeededError
from strix.tools.agents_graph.tools import agent_finish, stop_agent
from strix.tools.finish.tool import finish_scan
@@ -133,6 +132,27 @@ async def test_reserve_stop_notifies_root_once(monkeypatch: pytest.MonkeyPatch)
assert "finish_scan" in str(message["content"])
@pytest.mark.asyncio
async def test_reserve_notice_names_the_root_finish_tool(monkeypatch: pytest.MonkeyPatch) -> None:
coordinator = AgentCoordinator()
coordinator.root_finish_tool = "finish_pr_review"
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child-a", "recon", parent_id="root")
sent: list[dict[str, Any]] = []
async def _record(_target_agent_id: str, message: dict[str, Any]) -> bool:
sent.append(message)
return True
monkeypatch.setattr(coordinator, "send", _record)
await _notify_root_on_budget_reserve(coordinator)
assert "call finish_pr_review" in str(sent[0]["content"])
assert "finish_scan" not in str(sent[0]["content"])
@pytest.mark.asyncio
async def test_concurrent_reserve_claims_yield_single_root() -> None:
coordinator = AgentCoordinator()
@@ -1296,6 +1316,21 @@ async def test_interactive_nudge_offers_waiting_without_repeating() -> None:
assert "do not repeat it" in items[0]["content"]
@pytest.mark.asyncio
async def test_nudge_names_the_root_finish_tool() -> None:
items = await execution._append_tool_required_message(
session=None,
context={"parent_id": None},
attempt=1,
limit=3,
interactive=False,
root_finish_tool="finish_pr_review",
)
assert "call finish_pr_review" in items[0]["content"]
assert "finish_scan" not in items[0]["content"]
def _cycle_with_items(
coordinator: AgentCoordinator,
agent_id: str,
@@ -1537,143 +1572,29 @@ async def test_autonomous_nudge_does_not_offer_the_user() -> None:
@pytest.mark.asyncio
async def test_compaction_check_stops_run_after_first_call_when_over_budget(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(execution, "needs_compaction", lambda *_args: True)
guard: Any = execution._with_compaction_check(RunConfig(model="m")).call_model_input_filter
async def test_text_only_filter_scrubs_images_then_chains_existing_filter() -> None:
run_config = RunConfig(model="m")
assert execution._with_image_scrub(run_config, {"supports_images": True}) is run_config
seen: list[ModelInputData] = []
def _inner(data: CallModelData[Any]) -> ModelInputData:
seen.append(data.model_data)
return data.model_data
run_config = RunConfig(model="m", call_model_input_filter=_inner)
scrub: Any = execution._with_image_scrub(run_config, {"supports_images": False})
image = {"type": "input_image", "image_url": "data:image/png;base64,aGk="}
output = {"type": "function_call_output", "call_id": "c1", "output": [image]}
data = CallModelData(
model_data=ModelInputData(input=[{"role": "user", "content": "hi"}], instructions="sys"),
agent=MagicMock(tools=[]),
model_data=ModelInputData(input=[cast("Any", output)], instructions="sys"),
agent=MagicMock(),
context=None,
)
assert await guard(data) is data.model_data
with pytest.raises(CompactionNeededError):
await guard(data)
@pytest.mark.asyncio
async def test_compaction_check_chains_an_existing_input_filter(
monkeypatch: pytest.MonkeyPatch,
) -> None:
seen: list[Any] = []
monkeypatch.setattr(execution, "needs_compaction", lambda *args: seen.append(args[3]))
filtered = ModelInputData(input=[{"role": "user", "content": "scrubbed"}], instructions="sys")
run_config = RunConfig(model="m", call_model_input_filter=lambda _data: filtered)
check: Any = execution._with_compaction_check(run_config).call_model_input_filter
data = CallModelData(
model_data=ModelInputData(input=[], instructions="sys"),
agent=MagicMock(tools=[]),
context=None,
)
assert await check(data) is filtered
assert await check(data) is filtered
assert seen == [filtered.input]
class _BudgetStream:
def __init__(self, exc: BaseException | None) -> None:
self.run_loop_exception = exc
self.new_items: list[Any] = []
async def stream_events(self) -> Any:
for event in _NO_STREAM_EVENTS:
yield event
@pytest.mark.asyncio
async def test_run_cycle_compacts_and_resumes_on_every_budget_stop(
tmp_path: Any, monkeypatch: pytest.MonkeyPatch
) -> None:
stops = execution._MAX_COMPACTIONS_PER_CYCLE + 1
streams = [_BudgetStream(CompactionNeededError("m")) for _ in range(stops)]
streams.append(_BudgetStream(None))
runs: list[dict[str, Any]] = []
def _run_streamed(*_args: Any, **kwargs: Any) -> _BudgetStream:
runs.append(kwargs)
return streams[len(runs) - 1]
forced: list[bool] = []
async def _compact(*_args: Any, force: bool) -> bool:
forced.append(force)
return True
monkeypatch.setattr("strix.core.execution.Runner.run_streamed", _run_streamed)
monkeypatch.setattr(execution, "_compact_session", _compact)
monkeypatch.setattr(execution, "needs_compaction", lambda *_args: False)
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
session = SQLiteSession("root", tmp_path / "agents.db")
result = await execution._run_cycle(
MagicMock(),
coordinator,
"root",
input_data="task",
run_config=RunConfig(model="m"),
context={},
max_turns=5,
session=session,
interactive=True,
event_sink=None,
hooks=None,
)
assert cast("Any", result) is streams[-1]
assert len(runs) == stops + 1
assert forced.count(True) == stops
assert all(run["input"] == [] for run in runs[1:])
assert all(run["run_config"].call_model_input_filter is not None for run in runs)
session.close()
@pytest.mark.parametrize(
("compacted", "still_over"),
[(False, False), (True, True)],
ids=["compaction_failed", "still_over_budget"],
)
@pytest.mark.asyncio
async def test_run_cycle_reruns_unchecked_when_compaction_cannot_fit(
tmp_path: Any, monkeypatch: pytest.MonkeyPatch, compacted: bool, still_over: bool
) -> None:
streams = [_BudgetStream(CompactionNeededError("m")), _BudgetStream(None)]
runs: list[dict[str, Any]] = []
def _run_streamed(*_args: Any, **kwargs: Any) -> _BudgetStream:
runs.append(kwargs)
return streams[len(runs) - 1]
async def _compact(*_args: Any, **_kwargs: Any) -> bool:
return compacted
monkeypatch.setattr("strix.core.execution.Runner.run_streamed", _run_streamed)
monkeypatch.setattr(execution, "_compact_session", _compact)
monkeypatch.setattr(execution, "needs_compaction", lambda *_args: still_over)
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
session = SQLiteSession("root", tmp_path / "agents.db")
result = await execution._run_cycle(
MagicMock(),
coordinator,
"root",
input_data="task",
run_config=RunConfig(model="m"),
context={},
max_turns=5,
session=session,
interactive=True,
event_sink=None,
hooks=None,
)
assert cast("Any", result) is streams[-1]
assert len(runs) == 2
assert runs[0]["run_config"].call_model_input_filter is not None
assert runs[1]["run_config"].call_model_input_filter is None
assert runs[1]["input"] == []
session.close()
result = await scrub.call_model_input_filter(data)
assert result is seen[0]
assert result.input[0]["output"] == [
{"type": "input_text", "text": execution._TEXT_ONLY_IMAGE_TEXT}
]
assert data.model_data.input[0]["output"] == [image]
+80 -9
View File
@@ -79,12 +79,13 @@ def test_content_guardrail_is_not_retried() -> None:
assert execution._is_transient_model_error(guardrail) is False
def test_client_errors_are_not_transient() -> None:
def test_client_errors_are_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(bad_request) is True
for status in (401, 402, 403, 404):
assert execution._is_transient_model_error(_status_error(status)) is False
assert execution._is_transient_model_error(ValueError("nope")) is False
@@ -109,6 +110,7 @@ def _patch_fast_backoff(monkeypatch: pytest.MonkeyPatch) -> None:
async def _run_once(
monkeypatch: pytest.MonkeyPatch,
streams: list[_FakeStream],
session: Any = None,
) -> Any:
_patch_fast_backoff(monkeypatch)
calls = {"n": 0}
@@ -131,7 +133,7 @@ async def _run_once(
run_config=cast("RunConfig", object()),
context={},
max_turns=5,
session=None,
session=session,
interactive=False,
event_sink=None,
hooks=None,
@@ -166,9 +168,78 @@ async def test_run_cycle_gives_up_after_max_retries(
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):
streams = [_FakeStream(exc=ValueError("nope")), _FakeStream()]
with pytest.raises(ValueError, match="nope"):
await _run_once(monkeypatch, streams)
@pytest.mark.parametrize(
("status", "message", "expected"),
[
(
404,
"litellm.NotFoundError: NotFoundError: OpenrouterException - "
'{"error":{"message":"No endpoints found that support image input","code":404,'
'"metadata":{"failed_routing_step":"Filter by Image Support"}}}',
True,
),
(400, "Invalid content type. image_url is only supported by certain models.", True),
(400, "my-model is not a multimodal model", True),
(400, "This model doesn't support the image field for user messages.", True),
(422, "messages[3]: unknown variant `image_url`, expected `text`", True),
(400, "'Image URLs in user messages' functionality not supported.", True),
(
400,
"OpenRouterException: Message: This model's maximum context length is 1048576 "
"tokens, but the request requires 1073484 tokens (942412 input including "
"image/vision expansion + 131072 for the completion). Reduce the input length "
"or max_tokens.",
False,
),
(400, "At most 5 image(s) may be provided in one prompt.", False),
(
400,
"Upstream error from Relace: Generation stopped with an incomplete tool call",
False,
),
(500, "No endpoints found that support image input", False),
],
)
def test_is_image_rejection(status: int, message: str, expected: bool) -> None:
exc = APIStatusError(message, response=httpx.Response(status, request=_request()), body=None)
assert execution._is_image_rejection(exc) is expected
async def _run_with_image_session(
monkeypatch: pytest.MonkeyPatch, streams: list[_FakeStream]
) -> tuple[Any, int, int]:
strips = {"n": 0}
async def _fake_strip(_session: Any) -> bool:
strips["n"] += 1
return True
async def _no_compact(*_args: Any, **_kwargs: Any) -> bool:
return False
monkeypatch.setattr(execution, "strip_all_images_from_session", _fake_strip)
monkeypatch.setattr(execution, "_compact_session", _no_compact)
result, attempts, _coordinator = await _run_once(monkeypatch, streams, session=object())
return result, attempts, strips["n"]
@pytest.mark.asyncio
async def test_run_cycle_strips_images_on_image_rejection(
monkeypatch: pytest.MonkeyPatch,
) -> None:
rejection = APIStatusError(
"No endpoints found that support image input",
response=httpx.Response(status_code=404, request=_request()),
body=None,
)
streams = [_FakeStream(exc=rejection), _FakeStream()]
result, attempts, strips = await _run_with_image_session(monkeypatch, streams)
assert result is streams[1]
assert attempts == 2
assert strips == 1
+16
View File
@@ -7,6 +7,7 @@ from unittest.mock import MagicMock, patch
import pytest
from strix.core.agents import AgentCoordinator
from strix.core.hooks import (
BudgetExceededError,
BudgetPausedError,
@@ -335,6 +336,21 @@ async def test_budget_warning_root_directive_distinct_from_subagent() -> None:
assert "confirmed" in sub
@pytest.mark.asyncio
async def test_root_directive_names_the_coordinators_finish_tool() -> None:
hooks = ReportUsageHooks(model="test-model", max_turns=100)
coordinator = AgentCoordinator()
coordinator.root_finish_tool = "finish_pr_review"
ctx = _make_warn_context(requests=85, parent_id=None)
ctx.context["coordinator"] = coordinator
items: list[Any] = []
await hooks.on_llm_start(ctx, MagicMock(), None, items)
assert "finish_pr_review" in items[0]["content"]
assert "finish_scan" not in items[0]["content"]
@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:
+21
View File
@@ -394,9 +394,30 @@ def test_log_line_sink_formats_without_content(caplog: pytest.LogCaptureFixture)
assert "request_id=req_line01" in line
assert "status=400" in line
assert "provider=anthropic" in line
assert "upstream=- upstream_error=-" in line
assert "SECRET PROMPT" not in line
def test_log_line_names_upstream_recorded_during_attempt(
caplog: pytest.LogCaptureFixture,
) -> None:
exc = _anthropic_error(400, ANTHROPIC_BLOCK_BODY, {})
token = request_log._http_reply.set(request_log.HttpReply())
try:
request_log.record_upstream_provider("Together", "provider_unavailable")
event = request_log.event_from_litellm(
_anthropic_kwargs(exc), None, None, None, outcome="error"
)
finally:
request_log._http_reply.reset(token)
assert event.details is not None
assert event.details["upstream_provider"] == "Together"
with caplog.at_level(logging.DEBUG, logger="strix.llm.request_log"):
request_log._log_line_sink(event)
line = caplog.records[-1].getMessage()
assert "upstream=Together upstream_error=provider_unavailable" in line
# --------------------------------------------------------------------------- #
# sizes, timing, finish reason, free-form headers and details #
# --------------------------------------------------------------------------- #
+272 -6
View File
@@ -18,7 +18,7 @@ from typing import TYPE_CHECKING, Any
import pytest
from agents.mcp import MCPServer, MCPServerStdio, MCPServerStreamableHttp
from agents.tool_context import ToolContext
from mcp.types import CallToolResult, TextContent
from mcp.types import CallToolResult, ImageContent, TextContent
from mcp.types import Tool as MCPTool
from pydantic import ValidationError
@@ -43,6 +43,7 @@ from strix.tools.mcp import (
resolve_mcp_call,
search_mcp_tools,
)
from strix.tools.mcp import agent_tools as mcp_agent_tools
from strix.tools.mcp import client as mcp_client
from strix.tools.mcp import registry as mcp_registry_mod
from strix.tools.mcp import session as mcp_session_mod
@@ -421,8 +422,18 @@ def test_registry_summaries() -> None:
@pytest.mark.asyncio
async def test_list_mcps_returns_connections_with_ids_and_descriptions() -> None:
registry = McpRegistry()
registry.add(name="fs", server=FakeMCPServer("fs", []), purpose="local files", tool_count=2)
registry.add(name="db", server=FakeMCPServer("db", []), purpose=None, tool_count=1)
registry.add(
name="fs",
server=FakeMCPServer("fs", [_mcp_tool("read_file"), _mcp_tool("write_file")]),
purpose="local files",
tool_count=2,
)
registry.add(
name="db",
server=FakeMCPServer("db", [_mcp_tool("query")]),
purpose=None,
tool_count=1,
)
out = await list_mcps.on_invoke_tool(_ctx(registry), "{}")
@@ -437,7 +448,7 @@ async def test_list_mcps_returns_connections_with_ids_and_descriptions() -> None
"description": "local files",
"tool_count": 2,
"dead": False,
"state": "connected",
"state": "catalog_ready",
},
{
"id": "db",
@@ -445,10 +456,143 @@ async def test_list_mcps_returns_connections_with_ids_and_descriptions() -> None
"description": None,
"tool_count": 1,
"dead": False,
"state": "connected",
"state": "catalog_ready",
},
]
}
await registry.close()
@pytest.mark.asyncio
async def test_list_mcps_loads_catalog_for_unwarmed_connection(
monkeypatch: pytest.MonkeyPatch,
) -> None:
server = FakeMCPServer("fs", [_mcp_tool("read_file"), _mcp_tool("write_file")])
monkeypatch.setattr(
mcp_client,
"_build_server",
lambda _config: _built_server(server),
)
registry = McpRegistry()
entry = registry.register(McpConnectionRequest(config=_config("fs", None)))
out = await list_mcps.on_invoke_tool(_ctx(registry), "{}")
assert entry.state == "catalog_ready"
assert entry.tool_count == 2
assert out["connections"][0]["tool_count"] == 2
assert out["connections"][0]["state"] == "catalog_ready"
await registry.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("slow_stage", ["connect", "catalog"])
@pytest.mark.parametrize("warmup", [False, True])
async def test_list_mcps_returns_healthy_connections_while_slow_catalog_keeps_loading(
monkeypatch: pytest.MonkeyPatch,
slow_stage: str,
warmup: bool,
) -> None:
release = asyncio.Event()
started = asyncio.Event()
catalog_calls = 0
class _SlowServer(FakeMCPServer):
async def connect(self) -> None:
if slow_stage == "connect":
started.set()
await release.wait()
async def list_tools(self, run_context: Any = None, agent: Any = None) -> list[MCPTool]:
nonlocal catalog_calls
catalog_calls += 1
if slow_stage == "catalog":
started.set()
await release.wait()
return await super().list_tools(run_context, agent)
servers = {
"healthy": FakeMCPServer("healthy", [_mcp_tool("read_file")]),
"slow": _SlowServer("slow", [_mcp_tool("query")]),
}
monkeypatch.setattr(
mcp_client, "_build_server", lambda config: _built_server(servers[config.name])
)
monkeypatch.setattr(mcp_agent_tools, "_MCP_DISCOVERY_WAIT_SECONDS", 0.01)
registry = McpRegistry()
for name in servers:
registry.register(McpConnectionRequest(config=_config(name, None)))
if warmup:
registry.start_warmup()
entry = registry.get("slow")
assert entry is not None
try:
out = await asyncio.wait_for(list_mcps.on_invoke_tool(_ctx(registry), "{}"), timeout=1)
assert started.is_set()
assert out["connections"][0]["tool_count"] == 1
assert out["connections"][0]["state"] == "catalog_ready"
assert out["connections"][1]["tool_count"] is None
assert out["connections"][1]["state"] in {"connecting", "catalog_loading"}
assert entry._catalog_task is not None
assert not entry._catalog_task.done()
repeated = await asyncio.wait_for(list_mcps.on_invoke_tool(_ctx(registry), "{}"), timeout=1)
assert repeated == out
release.set()
await asyncio.wait_for(entry.ensure_catalog(), timeout=1)
finished = await list_mcps.on_invoke_tool(_ctx(registry), "{}")
assert finished["connections"][1]["tool_count"] == 1
assert finished["connections"][1]["state"] == "catalog_ready"
assert catalog_calls == 1
finally:
release.set()
await registry.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("fail_after_deadline", [False, True])
async def test_list_mcps_retains_and_cleans_up_background_catalog_tasks(
monkeypatch: pytest.MonkeyPatch,
fail_after_deadline: bool,
) -> None:
release = asyncio.Event()
finished = asyncio.Event()
loop_errors: list[dict[str, Any]] = []
async def load_catalog() -> list[MCPTool]:
await release.wait()
raise RuntimeError("catalog unavailable")
registry = McpRegistry()
entry = registry.add(
name="blocked", server=FakeMCPServer("blocked", []), purpose=None, tool_count=0
)
assert entry.session is not None
monkeypatch.setattr(entry.session, "list_tools", load_catalog)
monkeypatch.setattr(mcp_agent_tools, "_MCP_DISCOVERY_WAIT_SECONDS", 0.01)
loop = asyncio.get_running_loop()
previous_handler = loop.get_exception_handler()
loop.set_exception_handler(lambda _loop, context: loop_errors.append(context))
try:
out = await asyncio.wait_for(list_mcps.on_invoke_tool(_ctx(registry), "{}"), timeout=1)
assert out["connections"][0]["tool_count"] is None
assert entry._catalog_task is not None
assert not entry._catalog_task.done()
if fail_after_deadline:
entry._catalog_task.add_done_callback(lambda _task: finished.set())
release.set()
await asyncio.wait_for(finished.wait(), timeout=1)
assert entry.state == "connected"
# Drop the finished task without awaiting it; its error must already be consumed.
entry._catalog_task = None
else:
task = entry._catalog_task
await registry.close()
assert task.cancelled()
assert loop_errors == []
finally:
await registry.close()
loop.set_exception_handler(previous_handler)
@pytest.mark.asyncio
@@ -733,7 +877,37 @@ async def test_registry_warmup_bounds_parallel_connections(
release.set()
await warmup
assert [summary.state for summary in registry.summaries()] == ["connected"] * 4
assert [summary.state for summary in registry.summaries()] == ["catalog_ready"] * 4
assert [summary.tool_count for summary in registry.summaries()] == [1] * 4
await registry.close()
@pytest.mark.asyncio
async def test_registry_warmup_swallows_catalog_listing_errors(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class _FailingListServer(FakeMCPServer):
async def list_tools(
self,
run_context: Any = None,
agent: Any = None,
) -> list[MCPTool]:
raise RuntimeError("catalog unavailable")
monkeypatch.setattr(
mcp_client,
"_build_server",
lambda _config: _built_server(_FailingListServer("db", [])),
)
registry = McpRegistry()
entry = registry.register(McpConnectionRequest(config=_config("db", None)))
warmup = registry.start_warmup()
await warmup
assert warmup.done()
assert entry.state == "connected"
assert entry.tool_count == 0
await registry.close()
@@ -884,6 +1058,51 @@ async def test_call_mcp_flags_an_errored_result_failed_for_the_tui() -> None:
assert out == {"type": "text", "text": "boom:read_file", "success": False}
@pytest.mark.asyncio
async def test_call_mcp_replaces_images_for_a_text_only_model() -> None:
class _ImageServer(FakeMCPServer):
async def call_tool(
self,
tool_name: str,
arguments: dict[str, Any] | None,
meta: dict[str, Any] | None = None,
) -> CallToolResult:
return CallToolResult(
content=[
TextContent(type="text", text="shot"),
ImageContent(type="image", data="aGk=", mimeType="image/png"),
]
)
registry = McpRegistry()
registry.add(
name="fs", server=_ImageServer("fs", [_mcp_tool("shot")]), purpose=None, tool_count=1
)
ctx = _ctx(registry)
args = json.dumps({"connection": "fs", "tool": "shot"})
out = await call_mcp.on_invoke_tool(ctx, args)
assert out[1] == {"type": "image", "image_url": "data:image/png;base64,aGk="}
ctx.context["supports_images"] = False
out = await call_mcp.on_invoke_tool(ctx, args)
assert out == [
{"type": "text", "text": "shot"},
{"type": "text", "text": "[image/png image omitted: this model cannot view images]"},
]
transformed: list[Any] = []
registry.add(
name="pro",
server=_ImageServer("pro", [_mcp_tool("shot")]),
purpose=None,
tool_count=1,
result_transform=lambda _label, result: transformed.append(result),
)
await call_mcp.on_invoke_tool(ctx, json.dumps({"connection": "pro", "tool": "shot"}))
assert [block["type"] for block in transformed[0]["content"]] == ["text", "text"]
# --- generic MCP tools are the only MCP surface every agent gets -------------
@@ -967,6 +1186,53 @@ def test_prompt_renders_named_connection_inventory() -> None:
assert "read the app's schema" in prompt
def test_prompt_omits_unloaded_tool_count() -> None:
prompt = render_system_prompt(
system_prompt_context={
"mcp_available": True,
"mcp_connections": [
{"name": "pending_conn", "purpose": None, "tool_count": None, "state": "connected"}
],
}
)
connection_line = next(line for line in prompt.splitlines() if "pending_conn" in line)
assert "(0 tools)" not in connection_line
assert "tools)" not in connection_line
def test_prompt_renders_loaded_tool_count() -> None:
prompt = render_system_prompt(
system_prompt_context={
"mcp_available": True,
"mcp_connections": [
{"name": "ready_conn", "purpose": None, "tool_count": 8, "state": "catalog_ready"}
],
}
)
assert "ready_conn (8 tools)" in prompt
def test_prompt_renders_unavailable_connection_without_tool_count() -> None:
prompt = render_system_prompt(
system_prompt_context={
"mcp_available": True,
"mcp_connections": [
{
"name": "unavailable_conn",
"purpose": None,
"tool_count": None,
"state": "unavailable",
}
],
}
)
assert "unavailable_conn (unavailable right now)" in prompt
assert "unavailable_conn (0 tools)" not in prompt
def test_prompt_inventory_is_gated_on_availability() -> None:
"""The block is gated on ``mcp_available``; an ``mcp_connections`` payload
without it renders nothing, so a stale or spoofed list cannot leak names."""
+9
View File
@@ -16,6 +16,7 @@ from strix.config.models import (
_NonStreamingModel,
_TurnGuardModel,
configure_sdk_model_defaults,
model_supports_images,
request_timeout_extra_args,
resolve_api_type,
routes_through_litellm,
@@ -202,3 +203,11 @@ def test_configure_sdk_api_route_follows_the_given_model(
models.configure_sdk_api_route("my-private-model", settings)
assert routes == ["responses", "chat_completions"]
def test_image_support_follows_the_catalog(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setitem(litellm.model_cost, "acme-text", {})
monkeypatch.setitem(litellm.model_cost, "acme-vision", {"supports_vision": True})
assert model_supports_images("acme-text") is False
assert model_supports_images("litellm/openai/acme-vision") is True
assert model_supports_images("acme-unknown-model") is True
+9
View File
@@ -13,6 +13,7 @@ from types import SimpleNamespace
from typing import TYPE_CHECKING, Any, cast
import pytest
from caido_sdk_client.errors import OperationUserError
from strix.runtime.caido_handle import CaidoBootstrapHandle
from strix.tools.proxy import caido_api, tools
@@ -252,3 +253,11 @@ async def test_existing_request_ids_queries_current_project(
assert existing == {"1042"}
assert looked_up == ["1042", "1088"]
def test_err_logs_caido_errors_as_warnings(caplog: pytest.LogCaptureFixture) -> None:
with caplog.at_level("WARNING", logger="strix.tools.proxy.tools"):
tools._err("list_requests", OperationUserError(""))
[record] = caplog.records
assert record.levelname == "WARNING"
+41
View File
@@ -0,0 +1,41 @@
import logging
from collections.abc import Iterator
import pytest
from strix.telemetry import logging as tlog
_SDK_LOGGER = "agents.sandbox.sandboxes.docker"
@pytest.fixture
def sdk_logger() -> Iterator[logging.Logger]:
logger = logging.getLogger(_SDK_LOGGER)
original = list(logger.filters)
logger.filters.clear()
tlog._silence_pty_threshold_warning.cache_clear()
yield logger
logger.filters[:] = original
tlog._silence_pty_threshold_warning.cache_clear()
def _record(msg: str, *args: object) -> logging.LogRecord:
return logging.LogRecord(_SDK_LOGGER, logging.WARNING, __file__, 0, msg, args, None)
def test_installed_filter_drops_pty_threshold_warning(sdk_logger: logging.Logger) -> None:
tlog._silence_pty_threshold_warning()
record = _record("PTY process count reached warning threshold: %s active sessions", 62)
assert not sdk_logger.filter(record)
def test_installed_filter_keeps_other_warnings(sdk_logger: logging.Logger) -> None:
tlog._silence_pty_threshold_warning()
assert sdk_logger.filter(_record("something else went wrong"))
def test_install_is_idempotent(sdk_logger: logging.Logger) -> None:
tlog._silence_pty_threshold_warning()
tlog._silence_pty_threshold_warning()
assert len(sdk_logger.filters) == 1
+159
View File
@@ -408,6 +408,165 @@ async def test_dependency_report_records_transitive_chain(report_state: ReportSt
)
async def test_dependency_report_persists_code_locations(report_state: ReportState) -> None:
"""Usage-site locations file onto a dependency finding like any other."""
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="4.17.21",
cwe="CWE-94",
advisory_cvss=7.2,
technical_analysis=None,
fix_effort="trivial",
reachability="vulnerable_symbol_used",
reachability_evidence=_DEP_EVIDENCE,
code_locations=[
{
"file": "src/render.ts",
"start_line": 14,
"end_line": 14,
"snippet": "const merge = require('lodash').merge",
"label": "imports the vulnerable package",
}
],
contextual_cvss_breakdown=_DEP_CONTEXT,
contextual_cvss_reasoning=_DEP_REASONING,
)
assert result["success"] is True
locs = report_state.vulnerability_reports[0]["code_locations"]
assert locs[0]["file"] == "src/render.ts"
assert locs[0]["label"] == "imports the vulnerable package"
async def test_dependency_report_fix_locations_require_verification(
report_state: ReportState,
) -> None:
"""A location proposing a fix still requires fix_verification."""
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="4.17.21",
cwe="CWE-94",
advisory_cvss=7.2,
technical_analysis=None,
fix_effort="trivial",
reachability="imported",
reachability_evidence=_DEP_EVIDENCE,
code_locations=[
{
"file": "package-lock.json",
"start_line": 10,
"end_line": 10,
"fix_before": '"lodash": "4.17.20"',
"fix_after": '"lodash": "4.17.21"',
}
],
contextual_cvss_breakdown=_DEP_CONTEXT,
contextual_cvss_reasoning=_DEP_REASONING,
)
assert result["success"] is False
assert any("fix_verification" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_dependency_report_fix_locations_persist_with_verification(
report_state: ReportState,
) -> None:
"""A fix-pair location files when fix_verification is supplied, and both
fields persist on the stored report."""
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="4.17.21",
cwe="CWE-94",
advisory_cvss=7.2,
technical_analysis=None,
fix_effort="trivial",
reachability="imported",
reachability_evidence=_DEP_EVIDENCE,
code_locations=[
{
"file": "package-lock.json",
"start_line": 10,
"end_line": 10,
"fix_before": '"lodash": "4.17.20"',
"fix_after": '"lodash": "4.17.21"',
}
],
fix_verification="Bump verified: lockfile parses and the sink is unreachable.",
contextual_cvss_breakdown=_DEP_CONTEXT,
contextual_cvss_reasoning=_DEP_REASONING,
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
loc = report["code_locations"][0]
assert loc["fix_before"] == '"lodash": "4.17.20"'
assert loc["fix_after"] == '"lodash": "4.17.21"'
assert report["fix_verification"] == (
"Bump verified: lockfile parses and the sink is unreachable."
)
async def test_dependency_report_rejects_unusable_locations(
report_state: ReportState,
) -> None:
"""A supplied list that normalizes to nothing errors instead of filing
silently without the requested usage evidence."""
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="4.17.21",
cwe="CWE-94",
advisory_cvss=7.2,
technical_analysis=None,
fix_effort="trivial",
reachability="imported",
reachability_evidence=_DEP_EVIDENCE,
code_locations=[{"file": "/abs/path.ts", "snippet": "require('lodash')"}],
contextual_cvss_breakdown=_DEP_CONTEXT,
contextual_cvss_reasoning=_DEP_REASONING,
)
assert result["success"] is False
assert any("dropped as unusable" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_dependency_report_omits_blank_chain_fields(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
+125
View File
@@ -8,6 +8,7 @@ routine does the connecting.
from __future__ import annotations
import importlib
import types
from typing import Any
@@ -17,10 +18,21 @@ from agents import ModelSettings
import strix.tools.mcp as mcp_pkg
import strix.tools.notes.tools as notes_tools
import strix.tools.todo.tools as todo_tools
from strix.agents import factory
from strix.core import runner
from strix.core.agents import AgentCoordinator
from strix.runtime import session_manager
from strix.tools.mcp import McpConnectionConfig, McpConnectionRequest
from strix.tools.mcp import client as mcp_client
_test_mcp_client = importlib.import_module("tests.test_mcp_client")
FakeMCPServer: Any = _test_mcp_client.FakeMCPServer
_mcp_tool: Any = _test_mcp_client._mcp_tool
def _built_server(server: Any) -> Any:
return mcp_client.BuiltMcpServer(server, None)
def _settings() -> Any:
@@ -62,6 +74,11 @@ def _wire_runner(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
monkeypatch.setattr(runner, "build_strix_agent", lambda **_k: object())
monkeypatch.setattr(runner, "make_child_factory", lambda **_k: lambda **_kk: object())
monkeypatch.setattr(runner, "open_agent_session", lambda _root_id, _db: object())
monkeypatch.setattr(
mcp_client,
"_build_server",
lambda config: _built_server(FakeMCPServer(config.name, [])),
)
async def _run_agent_loop(**_kwargs: Any) -> None:
return None
@@ -156,6 +173,11 @@ async def test_roster_is_persisted_even_without_a_status_sink(
"load_user_mcp_configs",
lambda: [McpConnectionConfig(name="local_fs", transport="stdio", command="npx")],
)
monkeypatch.setattr(
mcp_client,
"_build_server",
lambda _config: _built_server(FakeMCPServer("local_fs", [])),
)
persisted: list[list[dict[str, Any]]] = []
@@ -182,3 +204,106 @@ async def test_roster_is_persisted_even_without_a_status_sink(
"state": "configured",
}
]
@pytest.mark.asyncio
async def test_warmup_listing_failure_does_not_interrupt_scan(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Any,
) -> None:
_wire_runner(monkeypatch, tmp_path)
server = FakeMCPServer("db", [])
async def _failing_list_tools(
run_context: Any = None,
agent: Any = None,
) -> list[Any]:
del run_context, agent
raise RuntimeError("catalog unavailable")
server.list_tools = _failing_list_tools
monkeypatch.setattr(
mcp_client,
"_build_server",
lambda _config: _built_server(server),
)
root_builds: list[dict[str, Any]] = []
monkeypatch.setattr(runner, "build_strix_agent", lambda **kwargs: root_builds.append(kwargs))
await runner.run_strix_scan(
scan_config={"targets": [], "scan_mode": "deep"},
scan_id="scan-mcp-list-failure",
image="img",
coordinator=AgentCoordinator(),
mcp_connection_requests=[
McpConnectionRequest(
config=McpConnectionConfig(name="db", url="https://mcp.example.com")
)
],
)
assert len(root_builds) == 1
@pytest.mark.asyncio
async def test_warmed_tool_counts_reach_root_and_child_agent_builds(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Any,
) -> None:
_wire_runner(monkeypatch, tmp_path)
server = FakeMCPServer("db", [_mcp_tool(f"tool_{index}") for index in range(3)])
monkeypatch.setattr(mcp_client, "_build_server", lambda _config: _built_server(server))
root_builds: list[dict[str, Any]] = []
def _build_root(**kwargs: Any) -> object:
root_builds.append(kwargs)
return object()
monkeypatch.setattr(
runner,
"build_strix_agent",
_build_root,
)
child_builds: list[dict[str, Any]] = []
def _build_child(**kwargs: Any) -> object:
child_builds.append(kwargs)
return object()
monkeypatch.setattr(
factory,
"build_strix_agent",
_build_child,
)
real_make_child_factory = factory.make_child_factory
child_factory_capture: dict[str, Any] = {}
def _make_child_factory(**kwargs: Any) -> Any:
child_factory_capture["context"] = kwargs["system_prompt_context"]
child_factory_capture["factory"] = real_make_child_factory(**kwargs)
return child_factory_capture["factory"]
monkeypatch.setattr(runner, "make_child_factory", _make_child_factory)
request = McpConnectionRequest(
config=McpConnectionConfig(name="db", url="https://mcp.example.com", notes="database")
)
await runner.run_strix_scan(
scan_config={"targets": [], "scan_mode": "deep"},
scan_id="scan-mcp-prompt-counts",
image="img",
coordinator=AgentCoordinator(),
mcp_connection_requests=[request],
)
root_context = root_builds[0]["system_prompt_context"]
assert root_context["mcp_connections"][0]["tool_count"] == 3
assert child_factory_capture["context"] is root_context
assert child_factory_capture["context"]["mcp_connections"][0]["tool_count"] == 3
child_factory_capture["factory"](name="child", skills=[])
assert child_builds[0]["system_prompt_context"] is child_factory_capture["context"]
assert child_builds[0]["system_prompt_context"]["mcp_connections"][0]["tool_count"] == 3
+42 -1
View File
@@ -6,6 +6,7 @@ flow through to the root agent's ``build_strix_agent`` call.
from __future__ import annotations
import importlib
import os
import types
from typing import Any
@@ -13,6 +14,7 @@ from typing import Any
import httpx
import pytest
from agents import ModelSettings
from agents.tool_context import ToolContext
from openai import RateLimitError
import strix.tools.mcp as mcp_pkg
@@ -24,7 +26,13 @@ from strix.core import runner
from strix.core.agents import AgentCoordinator
from strix.core.inputs import make_model_settings
from strix.runtime import session_manager
from strix.tools.load_skill.tool import load_skill
from strix.tools.mcp import BearerAuth, McpConnectionConfig, McpConnectionRequest
from strix.tools.mcp import client as mcp_client
_test_mcp_client = importlib.import_module("tests.test_mcp_client")
FakeMCPServer: Any = _test_mcp_client.FakeMCPServer
def _make_rate_limit_error() -> RateLimitError:
@@ -82,6 +90,11 @@ def _patch_engine_scaffold(
monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: scope_context)
monkeypatch.setattr(runner, "make_model_settings", lambda *_args, **_kwargs: ModelSettings())
monkeypatch.setattr(
mcp_client,
"_build_server",
lambda config: mcp_client.BuiltMcpServer(FakeMCPServer(config.name, []), None),
)
captured: dict[str, Any] = {}
@@ -220,7 +233,12 @@ async def test_mcp_available_flag_set_when_a_connection_attaches(
assert kwargs["system_prompt_context"]["mcp_available"] is True
# The named inventory names each connected server for the prompt.
assert kwargs["system_prompt_context"]["mcp_connections"] == [
{"name": "fs", "purpose": "local files", "tool_count": 0}
{
"name": "fs",
"purpose": "local files",
"tool_count": 0,
"state": "catalog_ready",
}
]
@@ -287,6 +305,29 @@ def test_requested_skills_follow_the_shared_prefix() -> None:
assert "<xss>" in xss.split("<cache_point>")[1]
def test_text_only_prompt_drops_screenshot_guidance() -> None:
assert "view_image" in render_system_prompt(include_scope=False)
prompt = render_system_prompt(include_scope=False, supports_images=False)
assert "view_image" not in prompt
assert "text-only model and cannot view images" in prompt
assert "<!--" not in prompt
@pytest.mark.asyncio
async def test_text_only_load_skill_drops_screenshot_guidance() -> None:
ctx = ToolContext(
context={"supports_images": False},
tool_name="load_skill",
tool_call_id="call-1",
tool_arguments="{}",
)
out = await load_skill.on_invoke_tool(ctx, '{"skills": ["agent_browser"]}')
assert "view_image" not in out
assert "text-only model" in out
def test_scope_is_sent_as_its_own_system_message_on_cache_point_routes() -> None:
settings = make_model_settings(None, model_name="anthropic/claude-sonnet-5-5")
prompt = render_system_prompt(
+11
View File
@@ -70,6 +70,17 @@ def test_registered_dir_adds_new_skill(tmp_path: Path) -> None:
assert load_skills(["widget"]) == {"widget": "widget body"}
def test_only_jinja_skills_are_rendered(tmp_path: Path) -> None:
body = "{% if supports_images %}\nlook\n{% else %}\nread\n{% endif %}\n"
_write_skill(tmp_path, "extra", "plain", body)
_write_skill(tmp_path, "extra", "templated", f"---\ntemplate: jinja\n---\n{body}")
register_skill_dir(tmp_path)
assert load_skills(["plain"]) == {"plain": body}
assert load_skills(["templated"]) == {"templated": "look\n"}
assert load_skills(["templated"], supports_images=False) == {"templated": "read\n"}
def test_available_skill_includes_frontmatter_description(tmp_path: Path) -> None:
_write_skill(
tmp_path,
+67 -10
View File
@@ -23,7 +23,8 @@ from openai import AsyncOpenAI
from strix.config import loader
from strix.config.loader import load_settings
from strix.config.models import StrixProvider, _TurnGuardModel, _with_idle_timeout
from strix.config.models import StrixProvider, _TurnGuardModel, _with_timeouts
from strix.llm import request_log
if TYPE_CHECKING:
@@ -78,9 +79,21 @@ def stalling_gateway() -> Iterator[str]:
server.server_close()
def _stream(base_url: str, *, idle_timeout: float) -> AsyncIterator[Any]:
class _ServedByRelace(OpenAIChatCompletionsModel):
async def stream_response(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any]:
request_log.record_upstream_provider("Relace")
async for event in super().stream_response(*args, **kwargs):
yield event
def _stream(base_url: str, *, idle_timeout: float | None) -> AsyncIterator[Any]:
client = AsyncOpenAI(api_key="tok", base_url=base_url, max_retries=0, timeout=_STALL_SECONDS)
inner: Model = OpenAIChatCompletionsModel(model="gw-model", openai_client=client)
inner: Model = request_log.RequestLoggingModel(
_ServedByRelace(model="gw-model", openai_client=client),
model_name="gw-model",
provider="openai",
base_url=base_url,
)
guarded = _TurnGuardModel(inner, stream_idle_timeout=idle_timeout)
return guarded.stream_response(
None,
@@ -96,7 +109,7 @@ def _stream(base_url: str, *, idle_timeout: float) -> AsyncIterator[Any]:
)
async def _drain(base_url: str, *, idle_timeout: float) -> list[Any]:
async def _drain(base_url: str, *, idle_timeout: float | None) -> list[Any]:
return [event async for event in _stream(base_url, idle_timeout=idle_timeout)]
@@ -105,16 +118,31 @@ async def test_stalled_stream_hangs_without_the_watchdog(stalling_gateway: str)
# Repro: tokens arrive, then nothing. Un-watched, the turn just sits there;
# the request timeout is far away and would reset on any keepalive byte.
with pytest.raises(TimeoutError):
await asyncio.wait_for(_drain(stalling_gateway, idle_timeout=0), timeout=2)
await asyncio.wait_for(_drain(stalling_gateway, idle_timeout=None), timeout=2)
@pytest.mark.asyncio
async def test_stalled_stream_is_abandoned_by_the_watchdog(stalling_gateway: str) -> None:
logged: list[request_log.LlmRequestEvent] = []
request_log.register_sink(logged.append)
started = time.monotonic()
with pytest.raises(TimeoutError, match="produced no event"):
await _drain(stalling_gateway, idle_timeout=1)
try:
with pytest.raises(TimeoutError, match="stream_idle_timeout"):
await _drain(stalling_gateway, idle_timeout=1)
finally:
request_log.unregister_sink(logged.append)
assert time.monotonic() - started < _STALL_SECONDS
# The request log can tell our timeout from a shutdown.
assert [
(e.error_type, (e.details or {}).get("upstream_provider"), e.error_message) for e in logged
] == [
(
"stream_idle_timeout",
"Relace",
"attempt cancelled before the reply was consumed: strix:stream_idle_timeout",
)
]
@pytest.mark.asyncio
@@ -124,14 +152,39 @@ async def test_events_keep_flowing_while_the_stream_is_alive() -> None:
await asyncio.sleep(0.05)
yield f"event-{i}"
seen: list[Any] = [event async for event in _with_idle_timeout(_live(), 1.0)]
seen: list[Any] = [event async for event in _with_timeouts(_live(), idle=1.0)]
assert seen == [f"event-{i}" for i in range(5)]
@pytest.mark.asyncio
async def test_first_event_and_whole_stream_are_bounded() -> None:
async def _slow_start() -> AsyncIterator[Any]:
await asyncio.sleep(_STALL_SECONDS)
yield "late"
async def _endless() -> AsyncIterator[Any]:
while True:
await asyncio.sleep(0.05)
yield "event"
with pytest.raises(TimeoutError, match="stream_first_event_timeout"):
[event async for event in _with_timeouts(_slow_start(), idle=5, first_event=0.2)]
with pytest.raises(TimeoutError, match="stream_idle_timeout"):
[event async for event in _with_timeouts(_slow_start(), idle=0.2)]
with pytest.raises(TimeoutError, match="stream_total_timeout"):
[event async for event in _with_timeouts(_endless(), idle=5, total=0.3)]
@pytest.fixture
def _reset_settings(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
for key in ("STRIX_LLM", "LLM_DISABLE_STREAMING", "LLM_STREAM_IDLE_TIMEOUT"):
for key in (
"STRIX_LLM",
"LLM_DISABLE_STREAMING",
"LLM_STREAM_IDLE_TIMEOUT",
"LLM_STREAM_FIRST_EVENT_TIMEOUT",
"LLM_STREAM_TOTAL_TIMEOUT",
):
monkeypatch.delenv(key, raising=False)
monkeypatch.setattr(loader, "_cached", None)
monkeypatch.setattr(loader, "_override", None)
@@ -151,11 +204,15 @@ def test_idle_timeout_is_configurable(
) -> None:
monkeypatch.setattr("strix.config.models.MultiProvider.get_model", lambda *_: _DummyModel())
monkeypatch.setenv("LLM_STREAM_IDLE_TIMEOUT", "45")
monkeypatch.setenv("LLM_STREAM_FIRST_EVENT_TIMEOUT", "30")
monkeypatch.setenv("LLM_STREAM_TOTAL_TIMEOUT", "400")
load_settings()
model = StrixProvider().get_model("openai/gpt-4o-mini")
assert isinstance(model, _TurnGuardModel)
assert model._stream_idle_timeout == 45
assert model._first_event_timeout == 30
assert model._total_timeout == 400
def test_idle_timeout_is_off_without_streaming(
@@ -170,4 +227,4 @@ def test_idle_timeout_is_off_without_streaming(
model = StrixProvider().get_model("openai/gpt-4o-mini")
assert isinstance(model, _TurnGuardModel)
assert model._stream_idle_timeout == 0
assert model._stream_idle_timeout is model._first_event_timeout is model._total_timeout is None