Compare commits

..
Author SHA1 Message Date
Ian 1f30c47e0e Revert the Vercel AI Gateway image-rejection guess
It also matched errors about one image form, e.g. Groq's
'Non-image file content parts' functionality not supported.
2026-10-07 17:15:07 -04:00
Ian 887ec1727f fix(execution): guess at Vercel AI Gateway's image rejection 2026-10-07 16:54:31 -04:00
Ian 8b0ab9ddb2 fix(execution): match DeepSeek's image rejection 2026-10-07 16:48:45 -04:00
Ian 9c44bba118 fix(execution): match only replies that mean the model takes no images 2026-10-07 16:48:45 -04:00
Ian b88bfd27b6 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.
2026-10-07 16:48:45 -04:00
Ian 0a6818f243 chore: fix the ruff and bandit failures on main 2026-10-07 16:48:35 -04:00
Ian 52ba4ab266 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.
2026-10-07 16:48:27 -04: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
25 changed files with 420 additions and 318 deletions

No files matched your search

+10 -1
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,7 @@ 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,
) -> SandboxAgent[Any]:
"""Build a SandboxAgent for either root or child use.
@@ -694,6 +699,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 [])]
@@ -733,6 +739,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 +761,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 +782,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
+6
View File
@@ -905,6 +905,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/"):
+36 -70
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,17 @@ 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
re.IGNORECASE,
)
_MAX_COMPACTIONS_PER_CYCLE = 2
@@ -128,49 +136,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 +168,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
@@ -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
+11 -1
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,
)
@@ -153,6 +154,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 +168,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"
@@ -265,6 +268,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}")
@@ -468,6 +474,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 +489,7 @@ async def run_strix_scan(
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=root_context,
instructions_override=root_instructions,
supports_images=supports_images,
)
if not is_resume:
@@ -501,6 +509,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 +537,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 +602,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
+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)
+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)
+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 -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()]
+2
View File
@@ -288,9 +288,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)
+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")
+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
+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
+22 -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
@@ -1537,143 +1536,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]
+73 -1
View File
@@ -109,6 +109,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 +132,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,
@@ -172,3 +173,74 @@ async def test_run_cycle_does_not_retry_permanent_error(
streams = [_FakeStream(exc=bad_request), _FakeStream()]
with pytest.raises(BadRequestError):
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,
"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
+46 -1
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
@@ -884,6 +884,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 -------------
+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
+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
+25
View File
@@ -13,6 +13,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,6 +25,7 @@ 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
@@ -287,6 +289,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,