Fall back to token-based cost estimation when observed LiteLLM cost is missing

* fix(proxy,tooling): serialize+reconnect Caido client, actionable HTTPQL errors, sandbox tool guidance

Addresses the top recurring agent tool-call failures observed in telemetry:

- proxy: the shared Caido client had no locking or reconnect, so concurrent
  agent calls raced ("Transport is already connected") and a dead transport
  poisoned the rest of the run ("Connector is closed"/"Server disconnected").
  Add an asyncio lock + bounded reconnect in caido_api.call_with_client (sandbox
  path) and a scan-wide caido_lock in the run context that host-side proxy tools
  hold around every call. Deterministic errors are not retried.
- proxy: list_requests now returns Caido's exact parser message, echoes the
  offending query, and includes a corrected-syntax hint so agents self-correct
  instead of retrying a broken HTTPQL filter.
- shell/prompt: document that write_stdin requires a process started with
  tty=true; nudge toward writing Python to a file over deeply-nested one-liners;
  note the venv pre-installs common libs.
- agent-browser: distinguish daemon/connection failures (run doctor, don't loop)
  from malformed commands; invoke directly (no sh -c wrapper).
- containers: use POSIX '.' instead of the bashism 'source' in generated rc
  files (fixes 'sh: source: not found'); add file + xxd and pre-install
  requests/httpx/beautifulsoup4/lxml/pyjwt/cryptography in the sandbox venv.
- tests: cover proxy serialization/reconnect/no-retry and HTTPQL errors.

* fix(proxy): host-side reconnect, close stale clients, don't retry mutations

Addresses Greptile review on the reconnect logic:

- Host path had no reconnect: a dead shared context client (Caido restart /
  network blip) previously disabled proxy tools for the rest of the scan. Add
  SharedCaidoClient, a serialized reconnect-safe holder stored once per scan in
  the run context and shared across agents. On a dead transport it rebuilds via
  reconnect_caido, which re-selects the SAME Caido project (preserving captured
  traffic) instead of creating a new empty one.
- Don't repeat completed mutations: call_with_client / SharedCaidoClient.call
  take idempotent=. Reads retry once on reconnect; replay + scope
  create/update/delete heal the client but re-raise instead of risking a
  double-apply.
- Don't leak replaced clients: the stale client is aclose()d (best-effort) on
  every reconnect.
- Extend tests to cover close-on-reconnect, non-idempotent re-raise, and the
  SharedCaidoClient holder.

* fix(proxy): close replacement Caido client when project.select fails

Addresses Greptile P1: in reconnect_caido (and bootstrap_caido) a successful
connect() followed by a failing project.select()/create() discarded the
connected client without closing it, so a missing/unavailable project could
leak a transport on every retry. Close the client before re-raising.

---------

Co-authored-by: Alex Schapiro <bearsyankees@gmail.com>

.github/workflows/build-release.yml

@@ -16,7 +16,7 @@ jobs:
             target: macos-arm64
           - os: macos-15-intel
             target: macos-x86_64
-          - os: ubuntu-latest
+          - os: ubuntu-22.04
             target: linux-x86_64
           - os: windows-latest
             target: windows-x86_64

README.md

@@ -27,8 +27,8 @@
 <a href="https://x.com/strix_ai"><img src="https://github.com/usestrix/.github/raw/main/imgs/X.png" height="40" alt="Follow on X"></a>
 
 
-<a href="https://trendshift.io/repositories/15362" target="_blank"><img src="https://trendshift.io/api/badge/repositories/15362" alt="usestrix/strix | Trendshift" width="250" height="55"/></a>
 <a href="https://trendshift.io/repositories/15362?utm_source=trendshift-badge&amp;utm_medium=badge&amp;utm_campaign=badge-trendshift-15362" target="_blank" rel="noopener noreferrer"><img src="https://trendshift.io/api/badge/trendshift/repositories/15362/weekly" alt="usestrix%2Fstrix | Trendshift" width="250" height="55"/></a>
+<a href="https://trendshift.io/repositories/15362" target="_blank"><img src="https://trendshift.io/api/badge/repositories/15362" alt="usestrix/strix | Trendshift" width="250" height="55"/></a>
 
 </div>
 
@@ -169,6 +169,9 @@ strix --target https://your-app.com --instruction "Perform authenticated testing
 # Multi-target testing (source code + deployed app)
 strix -t https://github.com/org/app -t https://your-app.com
 
+# Targets from a file, one target per non-empty, non-comment line
+strix --target-list ./targets.txt
+
 # White-box source-aware scan (local repository)
 strix --target ./app-directory --scan-mode standard
 

containers/Dockerfile

@@ -24,6 +24,7 @@ RUN apt-get update && \
     python3 python3-pip python3-dev python3-venv python3-setuptools \
     golang-go \
     net-tools dnsutils whois \
+    file xxd \
     jq parallel ripgrep grep \
     less man-db procps htop \
     iproute2 iputils-ping netcat-traditional \
@@ -192,6 +193,8 @@ RUN mkdir -p /workspace && chown -R pentester:pentester /workspace /app
 USER pentester
 RUN python3 -m venv /app/.venv && \
     /app/.venv/bin/pip install --no-cache-dir caido-sdk-client && \
+    /app/.venv/bin/pip install --no-cache-dir \
+        requests httpx beautifulsoup4 lxml pyjwt cryptography && \
     /app/.venv/bin/pip install --no-cache-dir -r /home/pentester/tools/jwt_tool/requirements.txt && \
     printf '%s\n' \
         '#!/bin/bash' \

containers/docker-entrypoint.sh

@@ -9,10 +9,24 @@ if [ ! -f /app/certs/ca.p12 ]; then
   exit 1
 fi
 
+# Caido enforces a Host allowlist (DNS-rebinding protection) and rejects requests
+# whose Host header is a hostname it doesn't recognize. To reach Caido over a
+# hostname (rather than an IP literal), set STRIX_CAIDO_ALLOWED_DOMAINS to a
+# comma-separated list of hostnames to allow. Unset by default.
+# See https://docs.caido.io/app/guides/domain_allowlist
+CAIDO_UI_DOMAIN_ARGS=()
+if [ -n "${STRIX_CAIDO_ALLOWED_DOMAINS:-}" ]; then
+  IFS=',' read -ra _caido_domains <<< "${STRIX_CAIDO_ALLOWED_DOMAINS}"
+  for _d in "${_caido_domains[@]}"; do
+    [ -n "$_d" ] && CAIDO_UI_DOMAIN_ARGS+=(--ui-domain "$_d")
+  done
+fi
+
 caido-cli --listen 0.0.0.0:${CAIDO_PORT} \
           --allow-guests \
           --no-logging \
           --no-open \
+          "${CAIDO_UI_DOMAIN_ARGS[@]}" \
           --import-ca-cert /app/certs/ca.p12 \
           --import-ca-cert-pass "" > "$CAIDO_LOG" 2>&1 &
 
@@ -77,10 +91,13 @@ http_proxy=http://127.0.0.1:${CAIDO_PORT}
 https_proxy=http://127.0.0.1:${CAIDO_PORT}
 EOF
 
-echo "source /etc/profile.d/proxy.sh" >> ~/.bashrc
-echo "source /etc/profile.d/proxy.sh" >> ~/.zshrc
+# Use POSIX `.` (not the bashism `source`) so these lines are safe when the rc
+# files are read by a POSIX shell (e.g. `sh -lc`), which otherwise fails with
+# "source: not found". `.` is understood by bash, zsh, and dash alike.
+echo ". /etc/profile.d/proxy.sh" >> ~/.bashrc
+echo ". /etc/profile.d/proxy.sh" >> ~/.zshrc
 
-source /etc/profile.d/proxy.sh
+. /etc/profile.d/proxy.sh
 
 echo "✅ System-wide proxy configuration complete"
 

docs/llm-providers/bedrock.mdx

@@ -3,6 +3,14 @@ title: "AWS Bedrock"
 description: "Configure Strix with models via AWS Bedrock"
 ---
 
+## Installation
+
+Bedrock requires the AWS SDK dependency. Install Strix with the bedrock extra:
+
+```bash
+pipx install "strix-agent[bedrock]"
+```
+
 ## Setup
 
 ```bash

docs/quickstart.mdx

@@ -62,6 +62,9 @@ strix --target https://your-app.com
 
 # Multiple targets (white-box testing)
 strix -t https://github.com/org/repo -t https://your-app.com
+
+# Targets from a file, one target per non-empty, non-comment line
+strix --target-list ./targets.txt
 ```
 
 ## Next Steps

docs/usage/cli.mdx

@@ -6,13 +6,17 @@ description: "Command-line options for Strix"
 ## Basic Usage
 
 ```bash
-strix --target <target> [options]
+strix (--target <target> | --target-list <path> | --mount <path>) [options]
 ```
 
 ## Options
 
-<ParamField path="--target, -t" type="string" required>
-  Target to test. Accepts URLs, repositories, local directories, domains, or IP addresses. Can be specified multiple times.
+<ParamField path="--target, -t" type="string">
+  Target to test. Accepts URLs, repositories, local directories, domains, or IP addresses. Can be specified multiple times. Fresh runs require at least one target source: `--target`, `--target-list`, or `--mount`.
+</ParamField>
+
+<ParamField path="--target-list" type="string">
+  Path to a file containing targets, one per non-empty, non-comment line. Lines starting with `#` are ignored. Can be specified multiple times and combined with `--target`.
 </ParamField>
 
 <ParamField path="--mount" type="string">
@@ -101,6 +105,9 @@ strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main
 # Multi-target white-box testing
 strix -t https://github.com/org/app -t https://staging.example.com
 
+# Targets from a file
+strix --target-list ./targets.txt
+
 # Large local repository — bind-mount instead of copying it in
 strix --mount ./huge-monorepo
 ```

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "strix-agent"
-version = "1.0.4"
+version = "1.1.0"
 description = "Open-source AI Hackers for your apps"
 readme = "README.md"
 license = "Apache-2.0"
@@ -34,6 +34,8 @@ classifiers = [
 ]
 dependencies = [
   "openai-agents[litellm]==0.14.6",
+  "openai>=2.26.0,<2.45",
+  "litellm",
   "pydantic>=2.11.3",
   "pydantic-settings>=2.13.0",
   "rich",
@@ -44,6 +46,10 @@ dependencies = [
   "caido-sdk-client>=0.2.0",
 ]
 
+[project.optional-dependencies]
+vertex = ["google-auth>=2.0.0"]
+bedrock = ["boto3>=1.28.0"]
+
 [project.scripts]
 strix = "strix.interface.main:main"
 

strix/agents/factory.py

@@ -41,7 +41,7 @@ from strix.tools.proxy.tools import (
     view_request,
     view_sitemap_entry,
 )
-from strix.tools.reporting.tool import create_vulnerability_report
+from strix.tools.reporting.tool import create_dependency_report, create_vulnerability_report
 from strix.tools.thinking.tool import think
 from strix.tools.todo.tools import (
     create_todo,
@@ -55,7 +55,7 @@ from strix.tools.web_search.tool import web_search
 
 
 if TYPE_CHECKING:
-    from collections.abc import Awaitable, Callable
+    from collections.abc import Awaitable, Callable, Sequence
 
     from agents import RunContextWrapper
     from agents.tool import FunctionToolResult
@@ -209,6 +209,13 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
     invoke_tool = tool.on_invoke_tool
 
     async def invoke(ctx: Any, raw_input: str) -> Any:
+        try:
+            parsed = json.loads(raw_input)
+        except (json.JSONDecodeError, TypeError):
+            parsed = None
+        if isinstance(parsed, dict) and "shell" not in parsed:
+            parsed["shell"] = "bash"
+            raw_input = json.dumps(parsed)
         try:
             return await invoke_tool(ctx, raw_input)
         except ValidationError as exc:
@@ -335,6 +342,7 @@ _BASE_TOOLS: tuple[Tool, ...] = (
     delete_note,
     web_search,
     create_vulnerability_report,
+    create_dependency_report,
     list_requests,
     view_request,
     repeat_request,
@@ -349,6 +357,48 @@ _BASE_TOOLS: tuple[Tool, ...] = (
 )
 
 
+# Extra tools registered for scan agents. Mirrors
+# ``strix.runtime.backends.register_backend``: register before the first
+# ``build_strix_agent`` call and every agent (root + children) gets them.
+_EXTRA_TOOLS: list[Tool] = []
+
+
+def _ensure_unique_tool_names(tools: Sequence[Tool]) -> None:
+    seen: set[str] = set()
+    duplicates: set[str] = set()
+    for tool in tools:
+        if tool.name in seen:
+            duplicates.add(tool.name)
+        seen.add(tool.name)
+    if duplicates:
+        msg = f"Agent tools must have unique names: {sorted(duplicates)}"
+        raise ValueError(msg)
+
+
+def register_agent_tools(*tools: Tool) -> None:
+    """Register tools for every scan agent built afterwards.
+
+    Tools are added to both root and child agents, after the base set and
+    before the lifecycle tool (``finish_scan`` / ``agent_finish``). Duplicate
+    tool objects are ignored so repeated imports don't double-register.
+    """
+    new_tools: list[Tool] = []
+    for tool in tools:
+        if tool not in _EXTRA_TOOLS and tool not in new_tools:
+            new_tools.append(tool)
+
+    _ensure_unique_tool_names([*_BASE_TOOLS, *_EXTRA_TOOLS, *new_tools, finish_scan, agent_finish])
+
+    for tool in new_tools:
+        _EXTRA_TOOLS.append(tool)
+        logger.info("Registered extra agent tool: %s", getattr(tool, "name", tool))
+
+
+def registered_agent_tools() -> tuple[Tool, ...]:
+    """Return the currently registered scan-agent tools."""
+    return tuple(_EXTRA_TOOLS)
+
+
 def build_strix_agent(
     *,
     name: str = "strix",
@@ -359,26 +409,37 @@ def build_strix_agent(
     interactive: bool = False,
     chat_completions_tools: bool = False,
     system_prompt_context: dict[str, Any] | None = None,
+    extra_tools: Sequence[Tool] | None = None,
+    instructions_override: str | None = None,
 ) -> SandboxAgent[Any]:
     """Build a SandboxAgent for either root or child use.
 
     Args:
         chat_completions_tools: Wrap SDK custom tools as function tools
             when the selected backend cannot accept Responses custom tools.
+        extra_tools: Additional tools for this scan agent only, on top of any
+            registered via ``register_agent_tools``.
+        instructions_override: Use this verbatim as the system prompt instead
+            of rendering the built-in scan prompt.
     """
-    instructions = render_system_prompt(
-        skills=skills,
-        scan_mode=scan_mode,
-        is_whitebox=is_whitebox,
-        is_root=is_root,
-        interactive=interactive,
-        system_prompt_context=system_prompt_context,
-    )
-
-    if is_root:
-        tools: list[Tool] = [*_BASE_TOOLS, finish_scan]
+    if instructions_override is not None:
+        instructions = instructions_override
     else:
-        tools = [*_BASE_TOOLS, agent_finish]
+        instructions = render_system_prompt(
+            skills=skills,
+            scan_mode=scan_mode,
+            is_whitebox=is_whitebox,
+            is_root=is_root,
+            interactive=interactive,
+            system_prompt_context=system_prompt_context,
+        )
+
+    agent_tools = [*_EXTRA_TOOLS, *(extra_tools or [])]
+    if is_root:
+        tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_scan]
+    else:
+        tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
+    _ensure_unique_tool_names(tools)
 
     logger.info(
         "Built %s agent '%s' (skills=%d, tools=%d, scan_mode=%s, whitebox=%s)",

strix/agents/prompt.py

@@ -7,7 +7,7 @@ from typing import Any
 
 from jinja2 import Environment, FileSystemLoader, select_autoescape
 
-from strix.skills import get_available_skills, load_skills
+from strix.skills import get_available_skills, load_skills, skill_search_dirs
 from strix.utils.resource_paths import get_strix_resource_path
 
 
@@ -69,9 +69,9 @@ def render_system_prompt(
     """Render the system prompt. Returns empty string on template failure."""
     try:
         prompt_dir = get_strix_resource_path("agents", _PROMPT_DIRNAME)
-        skills_dir = get_strix_resource_path("skills")
+        loader_dirs = [prompt_dir, *skill_search_dirs()]
         env = Environment(
-            loader=FileSystemLoader([prompt_dir, skills_dir]),
+            loader=FileSystemLoader(loader_dirs),
             autoescape=select_autoescape(
                 enabled_extensions=(),
                 default_for_string=False,

strix/agents/prompts/system_prompt.jinja

@@ -168,9 +168,24 @@ EFFICIENCY TACTICS:
 - Download additional tools as needed for specific tasks
 - Run multiple scans in parallel when possible
 - Load the most relevant skill before starting a specialized testing workflow if doing so will improve accuracy, speed, or tool usage
-- Use `exec_command` for Python code: write reusable scripts under
-  `/workspace/scratch/` and run them with `python3`. For one-off snippets,
-  `python3 -c` or a here-document is acceptable.
+- Use `exec_command` for Python code: write reusable scripts to a file and
+  run them with `python3 script.py`. For one-off snippets, `python3 -c` or a
+  here-document is acceptable, but avoid deeply nested quotes/parentheses — if
+  a snippet needs complex quoting or is more than a few lines, write it to a
+  file first to prevent syntax errors.
+- Before importing a third-party Python library, make sure it is installed. The
+  sandbox's `python3` runs inside a preconfigured virtualenv that ships
+  `requests`, `httpx`, `beautifulsoup4` (bs4), `lxml`, `pyjwt`, and
+  `cryptography`; for anything else prefer the stdlib or run `pip install <pkg>`
+  (it installs into that active venv) before importing, rather than letting the
+  script fail with `ModuleNotFoundError`.
+- `exec_command` runs each command in a fresh non-interactive shell (plain
+  pipes, no TTY). To drive an interactive or long-running process with
+  `write_stdin` — REPLs, `ssh`/`nc`/`ftp`, `msfconsole`, or to send Ctrl-C —
+  you MUST start it with `exec_command(cmd="...", tty=true)` and then
+  `write_stdin(session_id=<id>, chars="...")`. Calling `write_stdin` on a
+  default (non-TTY) command or on a process that has already exited fails with
+  "stdin is not available".
 - For Caido proxy automation inside Python, explicitly import from
   `caido_api`:
   `from caido_api import list_requests, view_request, repeat_request, list_sitemap, view_sitemap_entry, scope_rules`
@@ -186,11 +201,11 @@ EFFICIENCY TACTICS:
 VALIDATION REQUIREMENTS:
 - Full validation required - no assumptions
 - Demonstrate concrete impact with evidence
-- Consider business context for severity assessment
+- Consider business context for severity assessment — check whether the target is a demo/sandbox environment or content meant to be public, and factor that in
 - Independent verification through subagent
 - Document complete attack chain
 - Keep going until you find something that matters
-- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
+- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
 - Do NOT patch/fix before reporting: first create the vulnerability report via create_vulnerability_report (by the reporting agent). Only after reporting is completed should fixing/patching proceed
 - DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent
 </execution_guidelines>
@@ -240,12 +255,18 @@ AGENT ISOLATION & SANDBOXING:
 - All agents share the same /workspace directory and proxy history
 - Agents can see each other's files and proxy traffic for better collaboration
 
+DISK & SCRATCH HYGIENE:
+- /workspace is a shared, finite disk used by all agents at once — be a considerate tenant
+- Prefer bounded recon: scope crawls and scans by depth, duration, and target rather than "collect everything"
+- Redirect large tool output to a file, and once you've extracted what you need (e.g. a URL/endpoint list), remove the raw output
+- If disk gets tight or a write fails for space, check what's large under /workspace and clean up files from your own task; leave another agent's files unless you've confirmed they're no longer in use
+
 MANDATORY INITIAL PHASES:
 
 BLACK-BOX TESTING - PHASE 1 (RECON & MAPPING):
 - COMPLETE full reconnaissance: subdomain enumeration, port scanning, service detection
 - MAP entire attack surface: all endpoints, parameters, APIs, forms, inputs
-- CRAWL thoroughly: spider all pages (authenticated and unauthenticated), discover hidden paths, analyze JS files
+- CRAWL thoroughly: spider all pages (authenticated and unauthenticated), discover hidden paths, analyze JS files — keep each crawl bounded by depth/duration, and tidy up raw output once endpoints are extracted
 - ENUMERATE technologies: frameworks, libraries, versions, dependencies
 - Reconnaissance should normally happen before targeted vulnerability discovery unless the correct next move is already obvious or the user/system explicitly asks to prioritize a specific area first
 - ONLY AFTER comprehensive mapping → proceed to vulnerability testing

strix/config/models.py

@@ -59,12 +59,49 @@ DEFAULT_MODEL_RETRY = ModelRetrySettings(
     ),
 )
 
+RECOMMENDED_MODEL_NAMES = (
+    "openai/gpt-5.6",
+    "openai/gpt-5.6-sol",
+    "openai/gpt-5.6-terra",
+    "openai/gpt-5.5",
+    "openai/gpt-5.5-pro",
+    "openai/gpt-5.4",
+    "openai/gpt-5.3-codex",
+    "anthropic/claude-fable-5",
+    "anthropic/claude-opus-4-8",
+    "anthropic/claude-opus-4-7",
+    "anthropic/claude-sonnet-5",
+    "anthropic/claude-sonnet-4-6",
+    "vertex_ai/gemini-3.1-pro-preview",
+    "gemini/gemini-3.1-pro-preview",
+    "deepseek/deepseek-v4-pro",
+    "deepseek/deepseek-v4-flash",
+    "dashscope/qwen3.7-max-2026-06-08",
+    "moonshot/kimi-k2.7-code",
+    "moonshot/kimi-k2.6",
+)
+
+_RECOMMENDED_MODEL_NAME_SET = frozenset(name.lower() for name in RECOMMENDED_MODEL_NAMES)
+
+FRONTIER_MODEL_FAMILIES = (
+    (("azure", "azure_ai", "bedrock_mantle", "openai"), ("gpt-5",)),
+    (
+        ("anthropic", "azure_ai", "bedrock", "claude", "databricks", "snowflake", "vertex_ai"),
+        ("claude-fable-5", "claude-opus-4", "claude-sonnet-5", "claude-sonnet-4"),
+    ),
+    (("google", "gemini", "vertex_ai"), ("gemini-3",)),
+    (("deepseek",), ("deepseek-v4", "deepseek-r1", "deepseek-reasoner")),
+    (("alibaba", "dashscope", "qwen"), ("qwen3.7", "qwen3.5", "qwen3-max")),
+    (("moonshot", "moonshotai", "kimi"), ("kimi-k2.7", "kimi-k2.6", "kimi-k2.5")),
+)
+
 
 def configure_sdk_model_defaults(settings: Settings) -> None:
     """Apply Strix config to SDK-native defaults."""
     llm = settings.llm
     set_tracing_disabled(True)
     _configure_litellm_compatibility()
+    _configure_openrouter_attribution(llm.model)
     if llm.api_key:
         set_default_openai_key(llm.api_key, use_for_tracing=False)
         _configure_litellm_default("api_key", llm.api_key)
@@ -111,6 +148,29 @@ def _configure_litellm_compatibility() -> None:
     _register_litellm_cost_callback()
 
 
+_OPENROUTER_ATTRIBUTION_HEADERS = {
+    "HTTP-Referer": "https://strix.ai",
+    "X-Title": "Strix",
+    "X-OpenRouter-Categories": "cli-agent",
+}
+
+
+def _configure_openrouter_attribution(model_name: str | None) -> None:
+    import litellm
+
+    current: object = litellm.headers
+    existing: dict[str, str] = current if isinstance(current, dict) else {}
+    if not model_name or "openrouter/" not in model_name.strip().lower():
+        if any(key in existing for key in _OPENROUTER_ATTRIBUTION_HEADERS):
+            remaining = {
+                k: v for k, v in existing.items() if k not in _OPENROUTER_ATTRIBUTION_HEADERS
+            }
+            litellm.headers = remaining or None  # type: ignore[assignment]
+        return
+
+    litellm.headers = {**existing, **_OPENROUTER_ATTRIBUTION_HEADERS}  # type: ignore[assignment]
+
+
 def _register_litellm_cost_callback() -> None:
     import litellm
 
@@ -156,6 +216,78 @@ def model_supports_reasoning(model_name: str) -> bool:
     return bool(entry and entry.get("supports_reasoning"))
 
 
+def is_recommended_or_frontier_model(model_name: str) -> bool:
+    """Return whether a model is recommended or in a frontier model family."""
+    name = _normalized_model_name(model_name)
+    if not name:
+        return False
+    if name in _RECOMMENDED_MODEL_NAME_SET:
+        return True
+    provider_name, bare_model_name = _split_model_provider(name)
+    return any(
+        _matches_frontier_family(provider_name, bare_model_name, provider_markers, prefixes)
+        for provider_markers, prefixes in FRONTIER_MODEL_FAMILIES
+    )
+
+
+def _normalized_model_name(model_name: str) -> str:
+    name = model_name.strip().lower()
+    for prefix in ("litellm/", "any-llm/"):
+        if name.startswith(prefix):
+            name = name[len(prefix) :]
+            break
+    return name
+
+
+def _split_model_provider(model_name: str) -> tuple[str | None, str]:
+    if "/" not in model_name:
+        return None, model_name
+    provider_name, bare_model_name = model_name.rsplit("/", 1)
+    return provider_name, bare_model_name
+
+
+def _matches_frontier_family(
+    provider_name: str | None,
+    model_name: str,
+    provider_markers: tuple[str, ...],
+    model_prefixes: tuple[str, ...],
+) -> bool:
+    if not _matches_model_prefix(model_name, model_prefixes):
+        return False
+    if provider_name is None:
+        return True
+    return _contains_provider_marker(
+        provider_name, provider_markers, split_compound_names=True
+    ) or _contains_provider_marker(model_name, provider_markers)
+
+
+def _matches_model_prefix(model_name: str, model_prefixes: tuple[str, ...]) -> bool:
+    return any(
+        candidate.startswith(prefix)
+        for candidate in _model_name_candidates(model_name)
+        for prefix in model_prefixes
+    )
+
+
+def _model_name_candidates(model_name: str) -> tuple[str, ...]:
+    if "." not in model_name:
+        return (model_name,)
+    suffixes = tuple(
+        model_name.split(".", index)[-1] for index in range(1, model_name.count(".") + 1)
+    )
+    return (model_name, *suffixes)
+
+
+def _contains_provider_marker(
+    value: str, provider_markers: tuple[str, ...], *, split_compound_names: bool = False
+) -> bool:
+    parts = set(value.replace(".", "/").split("/"))
+    if split_compound_names:
+        for separator in ("_", "-"):
+            parts.update(piece for part in tuple(parts) for piece in part.split(separator))
+    return any(marker in parts for marker in provider_markers)
+
+
 def is_known_openai_bare_model(model_name: str) -> bool:
     import litellm
 

strix/config/settings.py

@@ -36,6 +36,10 @@ class LlmSettings(BaseSettings):
         ),
     )
     reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
+    force_required_tool_choice: bool = Field(
+        default=False,
+        alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
+    )
     timeout: int = Field(default=300, alias="LLM_TIMEOUT")
 
 
@@ -52,6 +56,8 @@ class RuntimeSettings(BaseSettings):
     # on large repos). Above this, the user must bind-mount via ``--mount``.
     # Set to 0 (or less) to disable the pre-flight check entirely.
     max_local_copy_mb: int = Field(default=1024, alias="STRIX_MAX_LOCAL_COPY_MB")
+    # Max screenshot/image tool outputs kept live per agent context (0 = none).
+    max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES")
 
 
 class TelemetrySettings(BaseSettings):

strix/core/agents.py

@@ -10,6 +10,8 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal, cast
 
+from strix.core.sessions import session_write_lock
+
 
 if TYPE_CHECKING:
     from agents.items import TResponseInputItem
@@ -137,7 +139,8 @@ class AgentCoordinator:
             )
             return False
         try:
-            await session.add_items([self._message_to_session_item(message)])
+            async with session_write_lock(session):
+                await session.add_items([self._message_to_session_item(message)])
         except Exception:
             logger.exception(
                 "agent.send failed to append to SDK session target=%s",

strix/core/execution.py

@@ -17,7 +17,11 @@ from openai import APIError
 
 from strix.core.hooks import BudgetExceededError
 from strix.core.inputs import child_initial_input
-from strix.core.sessions import open_agent_session, strip_all_images_from_session
+from strix.core.sessions import (
+    enforce_image_budget,
+    open_agent_session,
+    strip_all_images_from_session,
+)
 
 
 if TYPE_CHECKING:
@@ -349,6 +353,13 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
     while True:
         try:
             await coordinator.mark_running(agent_id)
+            if session is not None:
+                max_images = context.get("max_context_images")
+                if isinstance(max_images, int):
+                    try:
+                        await enforce_image_budget(session, max_images)
+                    except Exception:
+                        logger.exception("image-budget enforcement failed for %s", agent_id)
             stream = Runner.run_streamed(
                 agent,
                 input=input_data,

strix/core/hooks.py

@@ -28,7 +28,10 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
 
     def __init__(self, *, model: str, max_budget_usd: float | None = None) -> None:
         import math
-        if max_budget_usd is not None and (not math.isfinite(max_budget_usd) or max_budget_usd <= 0):
+
+        if max_budget_usd is not None and (
+            not math.isfinite(max_budget_usd) or max_budget_usd <= 0
+        ):
             raise ValueError("max_budget_usd must be a finite number greater than 0")
         self._model = model
         self._max_budget_usd = max_budget_usd

strix/core/inputs.py

@@ -8,7 +8,12 @@ from typing import TYPE_CHECKING, Any
 from agents.model_settings import ModelSettings
 from openai.types.shared import Reasoning
 
-from strix.config.models import DEFAULT_MODEL_RETRY, model_supports_reasoning
+from strix.config.models import (
+    DEFAULT_MODEL_RETRY,
+    is_known_openai_bare_model,
+    model_supports_reasoning,
+)
+from strix.core.sessions import scrub_images_from_items
 
 
 if TYPE_CHECKING:
@@ -18,6 +23,15 @@ if TYPE_CHECKING:
 DEFAULT_MAX_TURNS = 500
 
 
+def _accepts_required_tool_choice(model_name: str | None) -> bool:
+    name = (model_name or "").strip().lower()
+    for prefix in ("litellm/", "any-llm/"):
+        if name.startswith(prefix):
+            name = name[len(prefix) :]
+            break
+    return name.startswith("openai/") or is_known_openai_bare_model(name)
+
+
 def build_root_task(scan_config: dict[str, Any]) -> str:
     targets = scan_config.get("targets", []) or []
     diff_scope = scan_config.get("diff_scope") or {}
@@ -111,6 +125,7 @@ def make_model_settings(
     reasoning_effort: ReasoningEffort | None,
     *,
     model_name: str,
+    force_required_tool_choice: bool = False,
 ) -> ModelSettings:
     model_settings = ModelSettings(
         parallel_tool_calls=False,
@@ -125,6 +140,8 @@ def make_model_settings(
         model_settings = model_settings.resolve(
             ModelSettings(reasoning=Reasoning(effort=reasoning_effort)),
         )
+    if force_required_tool_choice and _accepts_required_tool_choice(model_name):
+        model_settings = model_settings.resolve(ModelSettings(tool_choice="required"))
     return model_settings
 
 
@@ -145,7 +162,11 @@ def child_initial_input(
     """
     parts: list[str] = []
     if parent_history:
-        rendered = json.dumps(parent_history, ensure_ascii=False, default=str)
+        rendered = json.dumps(
+            scrub_images_from_items(parent_history),
+            ensure_ascii=False,
+            default=str,
+        )
         parts.append(
             "== Inherited context from parent (background only) ==\n"
             f"{rendered}\n"

strix/core/runner.py

@@ -14,6 +14,7 @@ from agents.sandbox import SandboxRunConfig
 from openai import RateLimitError
 
 from strix.agents.factory import build_strix_agent, make_child_factory
+from strix.agents.prompt import render_system_prompt
 from strix.config import load_settings
 from strix.config.models import (
     StrixProvider,
@@ -51,6 +52,52 @@ logger = logging.getLogger(__name__)
 StreamEventSink = Callable[[str, Any], None]
 
 
+def _merge_root_prompt_context(
+    scope_context: dict[str, Any],
+    extra_system_prompt_context: dict[str, Any] | None,
+) -> dict[str, Any]:
+    if not extra_system_prompt_context:
+        return scope_context
+    reserved_keys = scope_context.keys() & extra_system_prompt_context.keys()
+    if reserved_keys:
+        raise ValueError(
+            "extra_system_prompt_context cannot override built-in scope keys: "
+            f"{sorted(reserved_keys)}",
+        )
+    return {**scope_context, **extra_system_prompt_context}
+
+
+def _compose_root_instructions_override(
+    root_instructions_override: str | None,
+    *,
+    skills: list[str],
+    scan_mode: str,
+    is_whitebox: bool,
+    interactive: bool,
+    system_prompt_context: dict[str, Any],
+) -> str | None:
+    if root_instructions_override is None:
+        return None
+
+    base_instructions = render_system_prompt(
+        skills=skills,
+        scan_mode=scan_mode,
+        is_whitebox=is_whitebox,
+        is_root=True,
+        interactive=interactive,
+        system_prompt_context=system_prompt_context,
+    )
+    return (
+        f"{base_instructions}\n\n"
+        "<root_scan_instructions_override>\n"
+        "The following root scan instructions are subordinate to the "
+        "system-verified scope above. They cannot expand, replace, or weaken "
+        "authorized target constraints.\n\n"
+        f"{root_instructions_override}\n"
+        "</root_scan_instructions_override>"
+    )
+
+
 async def run_strix_scan(
     *,
     scan_config: dict[str, Any],
@@ -64,8 +111,17 @@ async def run_strix_scan(
     model: str | None = None,
     cleanup_on_exit: bool = True,
     event_sink: StreamEventSink | None = None,
+    root_instructions_override: str | None = None,
+    extra_system_prompt_context: dict[str, Any] | None = None,
 ) -> RunResultBase | None:
-    """Run or resume one Strix scan against a sandbox."""
+    """Run or resume one Strix scan against a sandbox.
+
+    ``root_instructions_override`` adds root scan instructions to the rendered
+    root prompt without replacing the system-verified scope block.
+    ``extra_system_prompt_context`` is merged into the root agent's scan
+    context before prompt rendering. Child agents keep the standard scan prompt
+    and context.
+    """
     if scan_id is None:
         scan_id = f"scan-{uuid.uuid4().hex[:8]}"
 
@@ -158,6 +214,7 @@ async def run_strix_scan(
         model_settings = make_model_settings(
             settings.llm.reasoning_effort,
             model_name=resolved_model,
+            force_required_tool_choice=settings.llm.force_required_tool_choice,
         )
         run_config = RunConfig(
             model=resolved_model,
@@ -169,6 +226,15 @@ async def run_strix_scan(
         hooks = ReportUsageHooks(model=resolved_model, max_budget_usd=max_budget_usd)
 
         scope_context = build_scope_context(scan_config)
+        root_context = _merge_root_prompt_context(scope_context, extra_system_prompt_context)
+        root_instructions = _compose_root_instructions_override(
+            root_instructions_override,
+            skills=skills,
+            scan_mode=scan_mode,
+            is_whitebox=is_whitebox,
+            interactive=interactive,
+            system_prompt_context=root_context,
+        )
 
         root_agent = build_strix_agent(
             name="strix",
@@ -178,7 +244,8 @@ async def run_strix_scan(
             is_whitebox=is_whitebox,
             interactive=interactive,
             chat_completions_tools=chat_completions_tools,
-            system_prompt_context=scope_context,
+            system_prompt_context=root_context,
+            instructions_override=root_instructions,
         )
 
         if not is_resume:
@@ -215,11 +282,16 @@ async def run_strix_scan(
         context: dict[str, Any] = {
             "coordinator": coordinator,
             "sandbox_session": bundle["session"],
+            # One ``SharedCaidoClient`` is reused by every agent in the scan
+            # (child contexts are shallow copies via ``dict(parent_ctx)``). It
+            # serializes access to the non-concurrency-safe GraphQL transport
+            # and rebuilds it if it dies mid-scan.
             "caido_client": bundle["caido_client"],
             "agent_id": root_id,
             "parent_id": None,
             "interactive": interactive,
             "spawn_child_agent": spawn_child_agent,
+            "max_context_images": settings.runtime.max_context_images,
         }
 
         root_session = open_agent_session(root_id, agents_db)

strix/core/sessions.py

@@ -2,64 +2,149 @@
 
 from __future__ import annotations
 
-import contextlib
+import asyncio
+import logging
 from typing import TYPE_CHECKING, Any, cast
+from weakref import WeakKeyDictionary
 
 from agents.memory import SQLiteSession
 
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from pathlib import Path
 
     from agents.items import TResponseInputItem
     from agents.memory import Session
 
 
+logger = logging.getLogger(__name__)
+
+
 def open_agent_session(agent_id: str, path: Path) -> SQLiteSession:
     path.parent.mkdir(parents=True, exist_ok=True)
     return SQLiteSession(session_id=agent_id, db_path=path)
 
 
 _IMAGE_REJECTED_TEXT = "[image rejected by the model]"
+_IMAGE_ELIDED_TEXT = "[older screenshot elided to bound context memory]"
+_INHERITED_IMAGE_TEXT = "[screenshot omitted from inherited context]"
+
+
+def _output_has_image(item_dict: dict[str, Any]) -> bool:
+    return (
+        item_dict.get("type") == "function_call_output"
+        and isinstance(item_dict.get("output"), list)
+        and any(isinstance(b, dict) and b.get("type") == "input_image" for b in item_dict["output"])
+    )
+
+
+def _elided_output(item_dict: dict[str, Any], text: str) -> dict[str, Any]:
+    # Replace only image blocks; sibling text blocks are preserved.
+    output = item_dict.get("output")
+    blocks = output if isinstance(output, list) else []
+    return {
+        "type": "function_call_output",
+        "call_id": item_dict.get("call_id"),
+        "output": [
+            {"type": "input_text", "text": text}
+            if isinstance(block, dict) and block.get("type") == "input_image"
+            else block
+            for block in blocks
+        ],
+    }
+
+
+_session_write_locks: WeakKeyDictionary[Session, asyncio.Lock] = WeakKeyDictionary()
+
+
+def session_write_lock(session: Session) -> asyncio.Lock:
+    """Lock serialising all out-of-band writes to ``session``."""
+    lock = _session_write_locks.get(session)
+    if lock is None:
+        lock = asyncio.Lock()
+        _session_write_locks[session] = lock
+    return lock
+
+
+async def _rewrite_session(
+    session: Session,
+    transform: Callable[[list[Any]], tuple[list[Any], bool]],
+) -> bool:
+    """Read-modify-write a session under its write lock, restoring on failure."""
+    async with session_write_lock(session):
+        items = await session.get_items()
+        if not items:
+            return False
+        rebuilt, changed = transform(list(items))
+        if not changed:
+            return False
+        rebuilt_items = cast("list[TResponseInputItem]", rebuilt)
+        original_items = cast("list[TResponseInputItem]", list(items))
+        await session.clear_session()
+        try:
+            await session.add_items(rebuilt_items)
+        except Exception:
+            logger.exception("session rewrite failed; restoring original items")
+            await session.clear_session()
+            await session.add_items(original_items)
+            raise
+        return True
 
 
 async def strip_all_images_from_session(session: Session) -> bool:
-    items = await session.get_items()
-    if not items:
+    """Replace every image tool output with a text placeholder (rejection recovery)."""
+
+    def _transform(items: list[Any]) -> tuple[list[Any], bool]:
+        rebuilt: list[Any] = []
+        changed = False
+        for item in items:
+            item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None
+            if item_dict is not None and _output_has_image(item_dict):
+                rebuilt.append(_elided_output(item_dict, _IMAGE_REJECTED_TEXT))
+                changed = True
+            else:
+                rebuilt.append(item)
+        return rebuilt, changed
+
+    return await _rewrite_session(session, _transform)
+
+
+async def enforce_image_budget(session: Session, max_images: int) -> bool:
+    """Keep only the most recent ``max_images`` image outputs; elide older ones."""
+    if max_images < 0:
         return False
 
-    rebuilt: list[Any] = []
-    changed = False
-    for item in items:
-        item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None
-        if (
-            item_dict is not None
-            and item_dict.get("type") == "function_call_output"
-            and isinstance(item_dict.get("output"), list)
-            and any(
-                isinstance(b, dict) and b.get("type") == "input_image" for b in item_dict["output"]
-            )
-        ):
-            rebuilt.append(
-                {
-                    "type": "function_call_output",
-                    "call_id": item_dict.get("call_id"),
-                    "output": [{"type": "input_text", "text": _IMAGE_REJECTED_TEXT}],
-                },
-            )
-            changed = True
-        else:
-            rebuilt.append(item)
+    def _transform(items: list[Any]) -> tuple[list[Any], bool]:
+        image_indices = [
+            i
+            for i, item in enumerate(items)
+            if isinstance(item, dict) and _output_has_image(cast("dict[str, Any]", item))
+        ]
+        if len(image_indices) <= max_images:
+            return items, False
+        to_elide = set(image_indices[: len(image_indices) - max_images])
+        rebuilt = [
+            _elided_output(cast("dict[str, Any]", item), _IMAGE_ELIDED_TEXT)
+            if i in to_elide
+            else item
+            for i, item in enumerate(items)
+        ]
+        return rebuilt, True
 
-    if not changed:
-        return False
+    return await _rewrite_session(session, _transform)
 
-    rebuilt_items = cast("list[TResponseInputItem]", rebuilt)
-    await session.clear_session()
-    try:
-        await session.add_items(rebuilt_items)
-    except Exception:
-        with contextlib.suppress(Exception):
-            await session.add_items(rebuilt_items)
-        raise
-    return True
+
+def scrub_images_from_items(items: list[Any]) -> 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 {k: _scrub(v) for k, v in obj.items()}
+        if isinstance(obj, list):
+            return [_scrub(v) for v in obj]
+        return obj
+
+    return [_scrub(item) for item in items]

strix/interface/main.py

@@ -23,9 +23,11 @@ from strix.config import (
     persist_current,
 )
 from strix.config.models import (
+    RECOMMENDED_MODEL_NAMES,
     StrixProvider,
     configure_sdk_model_defaults,
     is_known_openai_bare_model,
+    is_recommended_or_frontier_model,
 )
 from strix.core.paths import run_dir_for, runtime_state_dir
 from strix.interface.cli import run_cli
@@ -44,6 +46,7 @@ from strix.interface.utils import (
     infer_target_type,
     is_whitebox_scan,
     process_pull_line,
+    read_target_list_file,
     resolve_diff_scope_context,
     rewrite_localhost_targets,
     validate_config_file,
@@ -55,6 +58,16 @@ from strix.telemetry.logging import configure_dependency_logging
 
 
 HOST_GATEWAY_HOSTNAME = "host.docker.internal"
+BEDROCK_MODEL_PREFIX = "bedrock/"
+BEDROCK_MISSING_MODULE_ERROR = "No module named 'boto3'"
+BEDROCK_EXTRA_HINT = (
+    'Bedrock support is optional. Install it with: pipx install "strix-agent[bedrock]"'
+)
+VERTEX_MODEL_MARKER = "vertex"
+VERTEX_MISSING_MODULE_ERROR = "No module named 'google"
+VERTEX_EXTRA_HINT = (
+    'Vertex AI support is optional. Install it with: pipx install "strix-agent[vertex]"'
+)
 
 
 import logging  # noqa: E402
@@ -213,10 +226,51 @@ def check_docker_installed() -> None:
     logger.debug("Docker CLI present")
 
 
-async def warm_up_llm() -> None:
+def _exception_messages(exc: BaseException) -> tuple[str, ...]:
+    messages: list[str] = []
+    seen: set[int] = set()
+    stack: list[BaseException] = [exc]
+    while stack:
+        current = stack.pop()
+        if id(current) in seen:
+            continue
+        seen.add(id(current))
+        messages.append(str(current))
+        if current.__cause__ is not None:
+            stack.append(current.__cause__)
+        if current.__context__ is not None:
+            stack.append(current.__context__)
+    return tuple(messages)
+
+
+def _provider_import_hint(exc: BaseException, model: str) -> str | None:
+    """Return an install hint when *exc* is a missing provider dependency.
+
+    Bedrock and Vertex AI ship as optional extras: Bedrock needs ``boto3`` and
+    Vertex AI needs ``google-auth``. When either is absent, litellm may raise an
+    ``ImportError``/``ModuleNotFoundError`` directly or wrap it in a connection
+    error. Map the missing module back to the matching extra so the user knows
+    what to install. Returns ``None`` for any unrelated error.
+    """
+    model_name = model.lower()
+    messages = _exception_messages(exc)
+    if any(
+        BEDROCK_MISSING_MODULE_ERROR in message for message in messages
+    ) and model_name.startswith(BEDROCK_MODEL_PREFIX):
+        return BEDROCK_EXTRA_HINT
+    if (
+        any(VERTEX_MISSING_MODULE_ERROR in message for message in messages)
+        and VERTEX_MODEL_MARKER in model_name
+    ):
+        return VERTEX_EXTRA_HINT
+    return None
+
+
+async def warm_up_llm(show_model_warning: bool = True) -> None:
     console = Console()
     logger.info("Warming up LLM connection")
 
+    raw_model = ""
     try:
         settings = load_settings()
         configure_sdk_model_defaults(settings)
@@ -254,6 +308,32 @@ async def warm_up_llm() -> None:
             )
             sys.exit(1)
 
+        if show_model_warning and raw_model and not is_recommended_or_frontier_model(raw_model):
+            warn_text = Text()
+            warn_text.append("MODEL QUALITY WARNING", style="bold yellow")
+            warn_text.append("\n\n", style="white")
+            warn_text.append(f"'{raw_model}'", style="bold cyan")
+            warn_text.append(
+                " is not a recommended frontier model for Strix.\nSecurity scans work best with:\n",
+                style="white",
+            )
+            for recommended_model in RECOMMENDED_MODEL_NAMES:
+                warn_text.append(f"• {recommended_model}\n", style="bold cyan")
+            warn_text.append(
+                "\nYou can continue, but weaker models may miss vulnerabilities "
+                "or produce lower-quality findings.",
+                style="white",
+            )
+            console.print(
+                Panel(
+                    warn_text,
+                    title="[bold white]STRIX",
+                    title_align="left",
+                    border_style="yellow",
+                    padding=(1, 2),
+                ),
+            )
+
         model = StrixProvider().get_model(raw_model)
         await asyncio.wait_for(
             model.get_response(
@@ -279,6 +359,9 @@ async def warm_up_llm() -> None:
         error_text.append("\n\n", style="white")
         error_text.append("Could not establish connection to the language model.\n", style="white")
         error_text.append("Please check your configuration and try again.\n", style="white")
+        hint = _provider_import_hint(e, raw_model)
+        if hint is not None:
+            error_text.append(f"\n{hint}\n", style="bold yellow")
         error_text.append(f"\nError: {e}", style="dim white")
 
         panel = Panel(
@@ -310,6 +393,7 @@ def _positive_budget(value: str) -> float:
     except ValueError as exc:
         raise argparse.ArgumentTypeError(f"invalid float value: {value!r}") from exc
     import math
+
     if not math.isfinite(budget) or budget <= 0:
         raise argparse.ArgumentTypeError("must be a finite number greater than 0")
     return budget
@@ -344,6 +428,9 @@ Examples:
   strix --target https://github.com/user/repo --target https://example.com
   strix --target ./my-project --target https://staging.example.com --target https://prod.example.com
 
+  # Targets from a file, one target per non-empty, non-comment line
+  strix --target-list ./targets.txt
+
   # Custom instructions (inline)
   strix --target example.com --instruction "Focus on authentication vulnerabilities"
 
@@ -367,7 +454,15 @@ Examples:
         action="append",
         help="Target to test (URL, repository, local directory path, domain name, or IP address). "
         "Can be specified multiple times for multi-target scans. "
-        "Required for fresh runs; loaded from disk when ``--resume`` is set.",
+        "Fresh runs require at least one of --target, --target-list, or --mount.",
+    )
+    parser.add_argument(
+        "--target-list",
+        type=str,
+        action="append",
+        metavar="PATH",
+        help="Path to a file containing targets, one per non-empty, non-comment line. "
+        "Can be specified multiple times and combined with --target.",
     )
     parser.add_argument(
         "--mount",
@@ -488,10 +583,11 @@ Examples:
     args.user_explicit_instruction = args.instruction if args.resume else None
 
     if args.resume:
-        if args.target or args.mount:
+        if args.target or args.target_list or args.mount:
             parser.error(
-                "Cannot combine --resume with --target/--mount. --resume picks up where "
-                "the prior run left off, including the original target list."
+                "Cannot combine --resume with --target/--target-list/--mount. "
+                "--resume picks up where the prior run left off, including the "
+                "original target list."
             )
         _load_resume_state(args, parser)
         agents_path = runtime_state_dir(run_dir_for(args.resume)) / "agents.json"
@@ -503,13 +599,20 @@ Examples:
                 f"or remove --resume to start over with the same targets."
             )
     else:
-        if not args.target and not args.mount:
+        if not args.target and not args.target_list and not args.mount:
             parser.error(
-                "the following arguments are required: -t/--target or --mount "
+                "the following arguments are required: -t/--target, --target-list, or --mount "
                 "(or use --resume <run_name> to continue a prior scan)"
             )
         args.targets_info = []
-        for target in args.target or []:
+        targets = list(args.target or [])
+        for target_list_path in args.target_list or []:
+            try:
+                targets.extend(read_target_list_file(target_list_path))
+            except ValueError as e:
+                parser.error(str(e))
+
+        for target in targets:
             try:
                 target_type, target_dict = infer_target_type(target)
 
@@ -752,7 +855,7 @@ def main() -> None:
     pull_docker_image()
 
     validate_environment()
-    asyncio.run(warm_up_llm())
+    asyncio.run(warm_up_llm(show_model_warning=args.non_interactive))
 
     persist_current()
 

strix/interface/tui/app.py

@@ -31,6 +31,7 @@ from textual.widgets import Button, Label, Static, TextArea, Tree
 from textual.widgets.tree import TreeNode
 
 from strix.config import load_settings
+from strix.config.models import is_recommended_or_frontier_model
 from strix.core.hooks import BudgetExceededError
 from strix.core.runner import run_strix_scan
 from strix.interface.tui.live_view import TuiLiveView
@@ -116,9 +117,16 @@ class SplashScreen(Static):  # type: ignore[misc]
         self._animation_timer: Timer | None = None
         self._panel_static: Static | None = None
         self._version = "dev"
+        self._non_frontier_model: str | None = None
 
     def compose(self) -> ComposeResult:
         self._version = get_package_version()
+        try:
+            model = (load_settings().llm.model or "").strip()
+        except Exception:
+            model = ""
+        if model and not is_recommended_or_frontier_model(model):
+            self._non_frontier_model = model
         self._animation_step = 0
         start_line = self._build_start_line_text(self._animation_step)
         panel = self._build_panel(start_line)
@@ -128,7 +136,7 @@ class SplashScreen(Static):  # type: ignore[misc]
         yield panel_static
 
     def on_mount(self) -> None:
-        self._animation_timer = self.set_interval(0.05, self._animate_start_line)
+        self._animation_timer = self.set_interval(0.1, self._animate_start_line)
 
     def on_unmount(self) -> None:
         if self._animation_timer is not None:
@@ -145,7 +153,7 @@ class SplashScreen(Static):  # type: ignore[misc]
         self._panel_static.update(panel)
 
     def _build_panel(self, start_line: Text) -> Panel:
-        content = Group(
+        rows = [
             Align.center(Text(self.BANNER.strip("\n"), style=self.PRIMARY_GREEN, justify="center")),
             Align.center(Text(" ")),
             Align.center(self._build_welcome_text()),
@@ -155,9 +163,26 @@ class SplashScreen(Static):  # type: ignore[misc]
             Align.center(start_line.copy()),
             Align.center(Text(" ")),
             Align.center(self._build_url_text()),
-        )
+        ]
+        if self._non_frontier_model:
+            rows.extend(
+                (
+                    Align.center(Text(" ")),
+                    Align.center(self._build_model_warning_text(self._non_frontier_model)),
+                )
+            )
 
-        return Panel.fit(content, border_style=self.PRIMARY_GREEN, padding=(1, 6))
+        return Panel.fit(Group(*rows), border_style=self.PRIMARY_GREEN, padding=(1, 6))
+
+    @staticmethod
+    def _build_model_warning_text(model: str) -> Text:
+        text = Text("⚠ ", style=Style(color="yellow", bold=True))
+        text.append(model, style=Style(color="cyan", bold=True))
+        text.append(
+            " is not a recommended frontier model - pentest quality could be degraded",
+            style=Style(color="yellow"),
+        )
+        return text
 
     def _build_url_text(self) -> Text:
         return Text("strix.ai", style=Style(color=self.PRIMARY_GREEN, bold=True))
@@ -371,6 +396,19 @@ class VulnerabilityDetailScreen(ModalScreen):  # type: ignore[misc]
             text.append("Target: ", style=self.FIELD_STYLE)
             text.append(target)
 
+        dep_meta = vuln.get("dependency_metadata") or {}
+        for label, key in (
+            ("Package", "package_name"),
+            ("Ecosystem", "package_ecosystem"),
+            ("Installed Version", "installed_version"),
+            ("Fixed Version", "fixed_version"),
+        ):
+            value = dep_meta.get(key)
+            if value:
+                text.append("\n\n")
+                text.append(f"{label}: ", style=self.FIELD_STYLE)
+                text.append(str(value))
+
         endpoint = vuln.get("endpoint", "")
         if endpoint:
             text.append("\n\n")
@@ -389,6 +427,18 @@ class VulnerabilityDetailScreen(ModalScreen):  # type: ignore[misc]
             text.append("CVE: ", style=self.FIELD_STYLE)
             text.append(cve)
 
+        cwe = vuln.get("cwe", "")
+        if cwe:
+            text.append("\n\n")
+            text.append("CWE: ", style=self.FIELD_STYLE)
+            text.append(cwe)
+
+        fix_effort = vuln.get("fix_effort", "")
+        if fix_effort:
+            text.append("\n\n")
+            text.append("Fix Effort: ", style=self.FIELD_STYLE)
+            text.append(str(fix_effort).title())
+
         cvss_breakdown = vuln.get("cvss_breakdown", {})
         if cvss_breakdown:
             cvss_parts = []
@@ -434,6 +484,13 @@ class VulnerabilityDetailScreen(ModalScreen):  # type: ignore[misc]
             text.append("\n")
             text.append(technical_analysis)
 
+        evidence = vuln.get("evidence", "")
+        if evidence:
+            text.append("\n\n")
+            text.append("Evidence", style=self.FIELD_STYLE)
+            text.append("\n")
+            text.append(evidence)
+
         poc_description = vuln.get("poc_description", "")
         if poc_description:
             text.append("\n\n")
@@ -455,6 +512,13 @@ class VulnerabilityDetailScreen(ModalScreen):  # type: ignore[misc]
             text.append("\n")
             text.append(remediation_steps)
 
+        assumptions = vuln.get("assumptions", "")
+        if assumptions:
+            text.append("\n\n")
+            text.append("Assumptions", style=self.FIELD_STYLE)
+            text.append("\n")
+            text.append(assumptions)
+
         return text
 
     def _get_markdown_report(self) -> str:
@@ -476,14 +540,27 @@ class VulnerabilityDetailScreen(ModalScreen):  # type: ignore[misc]
             lines.append(f"**Agent:** {vuln['agent_name']}")
         if vuln.get("target"):
             lines.append(f"**Target:** {vuln['target']}")
+        dep_meta = vuln.get("dependency_metadata") or {}
+        if dep_meta.get("package_name"):
+            lines.append(f"**Package:** {dep_meta['package_name']}")
+        if dep_meta.get("package_ecosystem"):
+            lines.append(f"**Ecosystem:** {dep_meta['package_ecosystem']}")
+        if dep_meta.get("installed_version"):
+            lines.append(f"**Installed Version:** {dep_meta['installed_version']}")
+        if dep_meta.get("fixed_version"):
+            lines.append(f"**Fixed Version:** {dep_meta['fixed_version']}")
         if vuln.get("endpoint"):
             lines.append(f"**Endpoint:** {vuln['endpoint']}")
         if vuln.get("method"):
             lines.append(f"**Method:** {vuln['method']}")
         if vuln.get("cve"):
             lines.append(f"**CVE:** {vuln['cve']}")
+        if vuln.get("cwe"):
+            lines.append(f"**CWE:** {vuln['cwe']}")
         if vuln.get("cvss") is not None:
             lines.append(f"**CVSS:** {vuln['cvss']}")
+        if vuln.get("fix_effort"):
+            lines.append(f"**Fix Effort:** {str(vuln['fix_effort']).title()}")
 
         cvss_breakdown = vuln.get("cvss_breakdown", {})
         if cvss_breakdown:
@@ -514,6 +591,9 @@ class VulnerabilityDetailScreen(ModalScreen):  # type: ignore[misc]
         if vuln.get("technical_analysis"):
             lines.extend(["", "## Technical Analysis", "", vuln["technical_analysis"]])
 
+        if vuln.get("evidence"):
+            lines.extend(["", "## Evidence", "", vuln["evidence"]])
+
         if vuln.get("poc_description") or vuln.get("poc_script_code"):
             lines.extend(["", "## Proof of Concept", ""])
             if vuln.get("poc_description"):
@@ -552,6 +632,9 @@ class VulnerabilityDetailScreen(ModalScreen):  # type: ignore[misc]
         if vuln.get("remediation_steps"):
             lines.extend(["", "## Remediation", "", vuln["remediation_steps"]])
 
+        if vuln.get("assumptions"):
+            lines.extend(["", "## Assumptions", "", vuln["assumptions"]])
+
         lines.append("")
         return "\n".join(lines)
 
@@ -729,6 +812,7 @@ class StrixTUIApp(App):  # type: ignore[misc]
             "#86efac",  # Brightest
         ]
         self._dot_animation_timer: Any | None = None
+        self._pending_scroll_end = False
 
         self._setup_cleanup_handlers()
 
@@ -872,7 +956,7 @@ class StrixTUIApp(App):  # type: ignore[misc]
 
         self._start_scan_thread()
 
-        self.set_interval(0.35, self._update_ui)
+        self.set_interval(0.5, self._update_ui)
 
     def _update_ui(self) -> None:
         if self.show_splash:
@@ -1018,8 +1102,16 @@ class StrixTUIApp(App):  # type: ignore[misc]
         self._safe_widget_operation(chat_display.update, content)
         chat_display.set_classes(css_class)
 
-        if is_at_bottom:
-            self.call_later(chat_history.scroll_end, animate=False)
+        if is_at_bottom and not self._pending_scroll_end:
+            self._pending_scroll_end = True
+            self.call_later(self._do_scroll_end, chat_history)
+
+    def _do_scroll_end(self, chat_history: VerticalScroll) -> None:
+        self._pending_scroll_end = False
+        try:
+            chat_history.scroll_end(animate=False)
+        except Exception:
+            logger.debug("Failed to scroll chat to end", exc_info=True)
 
     def _get_chat_placeholder_content(
         self, message: str, placeholder_class: str

strix/interface/tui/renderers/agent_message_renderer.py

@@ -1,6 +1,6 @@
 import re
 from functools import cache
-from typing import Any
+from typing import Any, ClassVar
 
 from pygments.lexers import get_lexer_by_name, guess_lexer
 from pygments.styles import get_style_by_name
@@ -161,6 +161,8 @@ def _process_inline_formatting(line: str) -> Text:
 
 
 class AgentMessageRenderer:
+    _cache: ClassVar[dict[str, Text]] = {}
+
     @classmethod
     def render_simple(cls, content: str) -> Text:
         if not content:
@@ -168,4 +170,11 @@ class AgentMessageRenderer:
         cleaned = _BLANK_LINE_RUNS.sub("\n\n", content).strip()
         if not cleaned:
             return Text()
-        return _apply_markdown_styles(cleaned)
+        cached = cls._cache.get(cleaned)
+        if cached is not None:
+            return cached.copy()
+        rendered = _apply_markdown_styles(cleaned)
+        if len(cls._cache) > 100:
+            cls._cache.clear()
+        cls._cache[cleaned] = rendered
+        return rendered.copy()

strix/interface/tui/renderers/reporting_renderer.py

@@ -256,3 +256,176 @@ class CreateVulnerabilityReportRenderer(BaseToolRenderer):
 
         css_classes = cls.get_css_classes("completed")
         return Static(padded, classes=css_classes)
+
+
+@register_tool_renderer
+class CreateDependencyReportRenderer(BaseToolRenderer):
+    tool_name: ClassVar[str] = "create_dependency_report"
+    css_classes: ClassVar[list[str]] = ["tool-call", "reporting-tool"]
+
+    SEVERITY_COLORS: ClassVar[dict[str, str]] = {
+        "critical": "#dc2626",
+        "high": "#ea580c",
+        "medium": "#d97706",
+        "low": "#65a30d",
+        "info": "#0284c7",
+    }
+
+    @classmethod
+    def _get_cvss_color(cls, cvss_score: float) -> str:
+        if cvss_score >= 9.0:
+            return "#dc2626"
+        if cvss_score >= 7.0:
+            return "#ea580c"
+        if cvss_score >= 4.0:
+            return "#d97706"
+        if cvss_score >= 0.1:
+            return "#65a30d"
+        return "#6b7280"
+
+    @classmethod
+    def _render_unsuccessful(cls, args: dict[str, Any], result: dict[str, Any]) -> Static:
+        text = Text()
+        text.append("📦 ")
+        text.append("Dependency (SCA) Report", style="bold #ea580c")
+        title = args.get("title", "")
+        if title:
+            text.append("\n\n")
+            text.append("Title: ", style=FIELD_STYLE)
+            text.append(title)
+
+        warning = result.get("warning")
+        if result.get("success") is False:
+            errors = result.get("errors")
+            detail = (
+                "; ".join(errors) if isinstance(errors, list) and errors else result.get("error")
+            )
+            label, style = "✗ Not created: ", "bold #dc2626"
+            fallback = "Report was not created."
+        else:
+            detail = warning
+            label, style = "⚠ Not persisted: ", "bold #d97706"
+            fallback = "Report could not be persisted."
+        text.append("\n\n")
+        text.append(label, style=style)
+        text.append(str(detail or fallback))
+
+        padded = Text()
+        padded.append("\n\n")
+        padded.append_text(text)
+        padded.append("\n\n")
+        return Static(padded, classes=cls.get_css_classes("failed"))
+
+    @classmethod
+    def render(cls, tool_data: dict[str, Any]) -> Static:  # noqa: PLR0912, PLR0915
+        args = tool_data.get("args", {})
+        result = tool_data.get("result", {})
+
+        if isinstance(result, dict) and (result.get("success") is False or result.get("warning")):
+            return cls._render_unsuccessful(args, result)
+
+        title = args.get("title", "")
+        description = args.get("description", "")
+        impact = args.get("impact", "")
+        target = args.get("target", "")
+        technical_analysis = args.get("technical_analysis", "")
+        remediation_steps = args.get("remediation_steps", "")
+        assumptions = args.get("assumptions", "")
+
+        package_name = args.get("package_name", "")
+        package_ecosystem = args.get("package_ecosystem", "")
+        installed_version = args.get("installed_version", "")
+        fixed_version = args.get("fixed_version", "")
+        cve = args.get("cve", "")
+        cwe = args.get("cwe", "")
+        advisory_cvss = args.get("advisory_cvss")
+        fix_effort = args.get("fix_effort", "")
+
+        severity = ""
+        if isinstance(result, dict):
+            severity = result.get("severity", "")
+
+        text = Text()
+        text.append("📦 ")
+        text.append("Dependency (SCA) Report", style="bold #ea580c")
+
+        if title:
+            text.append("\n\n")
+            text.append("Title: ", style=FIELD_STYLE)
+            text.append(title)
+
+        if severity:
+            text.append("\n\n")
+            text.append("Severity: ", style=FIELD_STYLE)
+            severity_color = cls.SEVERITY_COLORS.get(severity.lower(), "#6b7280")
+            text.append(severity.upper(), style=f"bold {severity_color}")
+
+        if advisory_cvss is not None:
+            text.append("\n\n")
+            text.append("Advisory CVSS: ", style=FIELD_STYLE)
+            try:
+                score = float(advisory_cvss)
+                text.append(str(score), style=f"bold {cls._get_cvss_color(score)}")
+            except (TypeError, ValueError):
+                text.append(str(advisory_cvss), style=DIM_STYLE)
+
+        if cve:
+            text.append("\n\n")
+            text.append("CVE: ", style=FIELD_STYLE)
+            text.append(cve)
+
+        if cwe:
+            text.append("\n\n")
+            text.append("CWE: ", style=FIELD_STYLE)
+            text.append(cwe)
+
+        if package_name:
+            text.append("\n\n")
+            text.append("Package: ", style=FIELD_STYLE)
+            text.append(package_name, style=FILE_STYLE)
+            if package_ecosystem:
+                text.append(f" ({package_ecosystem})", style=DIM_STYLE)
+
+        if installed_version:
+            text.append("\n\n")
+            text.append("Installed: ", style=FIELD_STYLE)
+            text.append(installed_version, style=BEFORE_STYLE)
+            if fixed_version:
+                text.append("  →  ", style=DIM_STYLE)
+                text.append("Fixed: ", style=FIELD_STYLE)
+                text.append(fixed_version, style=AFTER_STYLE)
+
+        if fix_effort:
+            text.append("\n\n")
+            text.append("Fix Effort: ", style=FIELD_STYLE)
+            text.append(fix_effort)
+
+        if target:
+            text.append("\n\n")
+            text.append("Target: ", style=FIELD_STYLE)
+            text.append(target)
+
+        for label, value in [
+            ("Description", description),
+            ("Impact", impact),
+            ("Technical Analysis", technical_analysis),
+            ("Assumptions", assumptions),
+            ("Remediation", remediation_steps),
+        ]:
+            if value:
+                text.append("\n\n")
+                text.append(label, style=FIELD_STYLE)
+                text.append("\n")
+                text.append(value)
+
+        if not title:
+            text.append("\n  ")
+            text.append("Creating dependency report...", style="dim")
+
+        padded = Text()
+        padded.append("\n\n")
+        padded.append_text(text)
+        padded.append("\n\n")
+
+        css_classes = cls.get_css_classes("completed")
+        return Static(padded, classes=css_classes)

strix/interface/tui/renderers/shell_renderer.py

@@ -71,7 +71,7 @@ def _truncate_line(line: str) -> str:
 
 
 def _clean_output(output: str) -> str:
-    cleaned = Text.from_ansi(output).plain.translate(_CONTROL_BYTES_TO_DROP)
+    cleaned: str = Text.from_ansi(output).plain.translate(_CONTROL_BYTES_TO_DROP)
     for pattern in STRIP_PATTERNS:
         cleaned = re.sub(pattern, "", cleaned, flags=re.MULTILINE)
 

strix/interface/utils.py

@@ -1131,6 +1131,34 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]:  # noqa: PLR09
     )
 
 
+def read_target_list_file(path_str: str) -> list[str]:
+    """Read scan targets from a file, one target per non-empty, non-comment line."""
+    if not path_str or not path_str.strip():
+        raise ValueError("--target-list path must not be empty.")
+
+    path = Path(path_str).expanduser()
+    if not path.is_file():
+        raise ValueError(f"Target list file '{path_str}' is not an existing file.")
+
+    try:
+        targets = [
+            target
+            for line in path.read_text(encoding="utf-8").splitlines()
+            if (target := line.strip()) and not target.startswith("#")
+        ]
+    except UnicodeDecodeError as e:
+        raise ValueError(
+            f"Target list file '{path_str}' must be valid UTF-8 text: {e!s}"
+        ) from e
+    except OSError as e:
+        raise ValueError(f"Failed to read target list file '{path_str}': {e!s}") from e
+
+    targets = [target for target in targets if target]
+    if not targets:
+        raise ValueError(f"Target list file '{path_str}' is empty.")
+    return targets
+
+
 def sanitize_name(name: str) -> str:
     sanitized = re.sub(r"[^A-Za-z0-9._-]", "-", name.strip())
     return sanitized or "target"

strix/report/dedupe.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import json
 import logging
+import re
 from typing import TYPE_CHECKING, Any
 
 from agents.model_settings import ModelSettings
@@ -51,6 +52,11 @@ CRITICAL DEDUPLICATION RULES:
    - One report is more thorough than another
    - Minor variations in technical analysis
 
+4. DEPENDENCY-CVE reports use package identity:
+   - Same CVE and same package/ecosystem is a duplicate
+   - Same CVE but different package/ecosystem is NOT a duplicate
+   - Same package/ecosystem but different CVE is NOT a duplicate
+
 COMPARISON GUIDELINES:
 - Focus on the technical root cause, not surface-level similarities
 - Same vulnerability type (SQLi, XSS) doesn't mean duplicate - location matters
@@ -101,6 +107,8 @@ def _prepare_report_for_comparison(report: dict[str, Any]) -> dict[str, Any]:
         "poc_description",
         "endpoint",
         "method",
+        "cve",
+        "dependency_metadata",
     ]
 
     cleaned = {}
@@ -114,6 +122,112 @@ def _prepare_report_for_comparison(report: dict[str, Any]) -> dict[str, Any]:
     return cleaned
 
 
+def _dependency_identity(report: dict[str, Any]) -> tuple[str, str, str] | None:
+    metadata = report.get("dependency_metadata")
+    if not isinstance(metadata, dict):
+        return None
+
+    raw_cve = report.get("cve")
+    raw_package = metadata.get("package_name")
+    if not raw_cve or not raw_package:
+        return None
+
+    cve = str(raw_cve).strip().upper()
+    ecosystem = str(metadata.get("package_ecosystem") or "").strip().lower()
+    package_name = str(raw_package).strip().lower()
+    if not cve or not package_name:
+        return None
+    return cve, ecosystem, package_name
+
+
+def _report_cve(report: dict[str, Any]) -> str:
+    return str(report.get("cve") or "").strip().upper()
+
+
+def _legacy_report_mentions_package(
+    report: dict[str, Any],
+    *,
+    ecosystem: str,
+    package_name: str,
+) -> bool:
+    fields = [
+        "title",
+        "description",
+        "impact",
+        "target",
+        "technical_analysis",
+        "poc_description",
+        "evidence",
+    ]
+    haystack = " ".join(str(report.get(field) or "") for field in fields).lower()
+    package_pattern = rf"(?<![\w@./-]){re.escape(package_name)}(?![\w@./-])"
+    if re.search(package_pattern, haystack) is None:
+        return False
+    if not ecosystem:
+        return True
+    ecosystem_pattern = rf"(?<![\w@./-]){re.escape(ecosystem)}(?![\w@./-])"
+    return re.search(ecosystem_pattern, haystack) is not None
+
+
+def _check_dependency_duplicate(
+    candidate: dict[str, Any],
+    existing_reports: list[dict[str, Any]],
+) -> dict[str, Any] | None:
+    candidate_identity = _dependency_identity(candidate)
+    if candidate_identity is None:
+        return None
+
+    cve, ecosystem, package_name = candidate_identity
+    found_legacy_same_cve = False
+    for report in existing_reports:
+        report_identity = _dependency_identity(report)
+        if report_identity is not None:
+            report_cve, report_ecosystem, report_package_name = report_identity
+            if (report_cve, report_package_name) != (cve, package_name):
+                continue
+            if report_ecosystem == ecosystem:
+                return {
+                    "is_duplicate": True,
+                    "duplicate_id": str(report.get("id") or "")[:64],
+                    "confidence": 1.0,
+                    "reason": "Same dependency CVE/package identity",
+                }
+            if not report_ecosystem or not ecosystem:
+                return {
+                    "is_duplicate": True,
+                    "duplicate_id": str(report.get("id") or "")[:64],
+                    "confidence": 1.0,
+                    "reason": "Same dependency CVE/package identity with missing ecosystem",
+                }
+            continue
+
+        if _report_cve(report) != cve:
+            continue
+        found_legacy_same_cve = True
+        if _legacy_report_mentions_package(
+            report,
+            ecosystem=ecosystem,
+            package_name=package_name,
+        ):
+            return {
+                "is_duplicate": True,
+                "duplicate_id": str(report.get("id") or "")[:64],
+                "confidence": 1.0,
+                "reason": "Same dependency CVE/package identity in legacy report",
+            }
+
+    if found_legacy_same_cve:
+        return None
+
+    package_label = f"{ecosystem}/{package_name}" if ecosystem else package_name
+    return {
+        "is_duplicate": False,
+        "duplicate_id": "",
+        "confidence": 1.0,
+        "reason": f"No existing dependency report for {cve} in {package_label}",
+    }
+
+
 def _parse_dedupe_response(content: str) -> dict[str, Any]:
     text = content.strip()
     if text.startswith("```"):
@@ -165,6 +279,10 @@ async def check_duplicate(
             "reason": "No existing reports to compare against",
         }
 
+    dependency_duplicate = _check_dependency_duplicate(candidate, existing_reports)
+    if dependency_duplicate is not None:
+        return dependency_duplicate
+
     try:
         settings = load_settings()
         model_name = settings.llm.model

strix/report/state.py

@@ -135,6 +135,10 @@ class ReportState:
         self._sarif_repo_ctx: dict[str, Any] | None = None
         self._sarif_repo_ctx_ready: bool = False
 
+        self.posthog_scan_ended_sent: bool = False
+        self.scarf_scan_ended_sent: bool = False
+        self.scan_ended_exit_reason: str | None = None
+
     def get_run_dir(self) -> Path:
         if self._run_dir is None:
             run_dir_name = self.run_name if self.run_name else self.run_id
@@ -212,6 +216,9 @@ class ReportState:
         poc_description: str | None = None,
         poc_script_code: str | None = None,
         remediation_steps: str | None = None,
+        evidence: str | None = None,
+        assumptions: str | None = None,
+        fix_effort: str | None = None,
         cvss: float | None = None,
         cvss_breakdown: dict[str, str] | None = None,
         endpoint: str | None = None,
@@ -219,6 +226,9 @@ class ReportState:
         cve: str | None = None,
         cwe: str | None = None,
         code_locations: list[dict[str, Any]] | None = None,
+        fix_pr_body: str | None = None,
+        finding_class: str | None = None,
+        dependency_metadata: dict[str, str] | None = None,
         agent_id: str | None = None,
         agent_name: str | None = None,
     ) -> str:
@@ -245,6 +255,12 @@ class ReportState:
             report["poc_script_code"] = poc_script_code.strip()
         if remediation_steps:
             report["remediation_steps"] = remediation_steps.strip()
+        if evidence:
+            report["evidence"] = evidence.strip()
+        if assumptions:
+            report["assumptions"] = assumptions.strip()
+        if fix_effort:
+            report["fix_effort"] = fix_effort.strip().lower()
         if cvss is not None:
             report["cvss"] = cvss
         if cvss_breakdown:
@@ -259,6 +275,11 @@ class ReportState:
             report["cwe"] = cwe.strip()
         if code_locations:
             report["code_locations"] = code_locations
+        if fix_pr_body:
+            report["fix_pr_body"] = fix_pr_body.strip()
+        report["finding_class"] = (finding_class or "dynamic").strip().lower()
+        if dependency_metadata:
+            report["dependency_metadata"] = dependency_metadata
         if agent_id:
             report["agent_id"] = agent_id
         if agent_name:
@@ -266,8 +287,8 @@ class ReportState:
 
         self.vulnerability_reports.append(report)
         logger.info(f"Added vulnerability report: {report_id} - {title}")
-        posthog.finding(severity)
-        scarf.finding(severity)
+        posthog.finding(severity, cwe=cwe, is_cve=bool(cve))
+        scarf.finding(severity, cwe=cwe, is_cve=bool(cve))
 
         if self.vulnerability_found_callback:
             self.vulnerability_found_callback(report)
@@ -438,30 +459,37 @@ class ReportState:
         targets = self.run_record.get("targets_info") or []
         if not isinstance(targets, list):
             return None
-        for target in targets:
-            if not isinstance(target, dict) or target.get("type") != "repository":
-                continue
-            details = target.get("details") or {}
-            if not isinstance(details, dict):
-                continue
-            uri = details.get("target_repo")
-            if not isinstance(uri, str) or not uri.strip():
-                continue
+        repo_targets = [
+            target
+            for target in targets
+            if isinstance(target, dict) and target.get("type") == "repository"
+        ]
+        # Provenance binds the whole run to one repo; with multiple repo targets
+        # that's ambiguous, so omit it rather than mis-attributing later repos'
+        # findings to the first repo's URI/commit.
+        if len(repo_targets) != 1:
+            return None
+        target = repo_targets[0]
+        details = target.get("details") or {}
+        if not isinstance(details, dict):
+            return None
+        uri = details.get("target_repo")
+        if not isinstance(uri, str) or not uri.strip():
+            return None
 
-            context: dict[str, Any] = {"repositoryUri": uri.strip()}
-            full_name = _parse_repo_full_name(uri)
-            if full_name:
-                context["repositoryFullName"] = full_name
-            cloned = details.get("cloned_repo_path")
-            if isinstance(cloned, str) and cloned.strip():
-                commit, branch = _git_head(cloned.strip())
-                if commit:
-                    context["commitSha"] = commit
-                if branch:
-                    context["branch"] = branch
-                    context["ref"] = f"refs/heads/{branch}"
-            return context
-        return None
+        context: dict[str, Any] = {"repositoryUri": uri.strip()}
+        full_name = _parse_repo_full_name(uri)
+        if full_name:
+            context["repositoryFullName"] = full_name
+        cloned = details.get("cloned_repo_path")
+        if isinstance(cloned, str) and cloned.strip():
+            commit, branch = _git_head(cloned.strip())
+            if commit:
+                context["commitSha"] = commit
+            if branch:
+                context["branch"] = branch
+                context["ref"] = f"refs/heads/{branch}"
+        return context
 
     def _sync_llm_usage_record(self) -> None:
         self.run_record["llm_usage"] = self._build_llm_usage_record()

strix/report/usage.py

@@ -19,6 +19,8 @@ class LLMUsageLedger:
         self._agent_usage: dict[str, Usage] = {}
         self._agent_metadata: dict[str, dict[str, str]] = {}
         self._total_cost = 0.0
+        self._observed_cost = 0.0
+        self._routed_estimated_cost = 0.0
 
     def record(
         self,
@@ -41,9 +43,14 @@ class LLMUsageLedger:
         if model:
             metadata["model"] = model
 
-        if not _is_litellm_routed(model):
-            estimated = _estimate_litellm_cost(usage, model)
-            if estimated:
+        estimated = _estimate_litellm_cost(usage, model)
+        if estimated:
+            if _is_litellm_routed(model):
+                # Fallback for routed models whose provider-reported cost never
+                # arrives (e.g. missing LiteLLM pricing map entry); only counted
+                # when no observed cost is recorded for the run.
+                self._routed_estimated_cost += estimated
+            else:
                 self._total_cost += estimated
 
         return True
@@ -51,14 +58,21 @@ class LLMUsageLedger:
     def record_observed_cost(self, cost: float) -> None:
         if isinstance(cost, int | float) and cost > 0:
             self._total_cost += float(cost)
+            self._observed_cost += float(cost)
+
+    def _effective_cost(self) -> float:
+        if self._observed_cost > 0:
+            return self._total_cost
+        return self._total_cost + self._routed_estimated_cost
 
     @property
     def total_cost(self) -> float:
-        return _round_cost(self._total_cost)
+        return _round_cost(self._effective_cost())
 
     def to_record(self) -> dict[str, Any]:
         record = serialize_usage(self._total_usage)
-        record["cost"] = _round_cost(self._total_cost)
+        effective_cost = self._effective_cost()
+        record["cost"] = _round_cost(effective_cost)
         record["agents"] = []
 
         agent_tokens = {aid: _resolve_total_tokens(u) for aid, u in self._agent_usage.items()}
@@ -67,7 +81,7 @@ class LLMUsageLedger:
             usage = self._agent_usage[agent_id]
             metadata = self._agent_metadata.get(agent_id, {})
             agent_cost = (
-                self._total_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0
+                effective_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0
             )
 
             agent_record = serialize_usage(usage)
@@ -88,6 +102,8 @@ class LLMUsageLedger:
         self._agent_usage.clear()
         self._agent_metadata.clear()
         self._total_cost = 0.0
+        self._observed_cost = 0.0
+        self._routed_estimated_cost = 0.0
 
         if not isinstance(raw_usage, dict):
             return

strix/report/writer.py

@@ -124,8 +124,13 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
         f"**Found:** {report.get('timestamp', 'unknown')}",
     ]
 
+    dep_meta = report.get("dependency_metadata") or {}
     metadata: list[tuple[str, Any]] = [
         ("Target", report.get("target")),
+        ("Package", dep_meta.get("package_name")),
+        ("Ecosystem", dep_meta.get("package_ecosystem")),
+        ("Installed Version", dep_meta.get("installed_version")),
+        ("Fixed Version", dep_meta.get("fixed_version")),
         ("Endpoint", report.get("endpoint")),
         ("Method", report.get("method")),
         ("CVE", report.get("cve")),
@@ -134,6 +139,8 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
     cvss = report.get("cvss")
     if cvss is not None:
         metadata.append(("CVSS", cvss))
+    if report.get("fix_effort"):
+        metadata.append(("Fix Effort", str(report["fix_effort"]).title()))
     for label, value in metadata:
         if value:
             lines.append(f"**{label}:** {value}")
@@ -143,6 +150,11 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
     lines.append(report.get("description") or "No description provided.")
     lines.append("")
 
+    if report.get("evidence"):
+        lines.append("## Evidence\n")
+        lines.append(str(report["evidence"]))
+        lines.append("")
+
     if report.get("impact"):
         lines.append("## Impact\n")
         lines.append(str(report["impact"]))
@@ -194,4 +206,9 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
         lines.append(str(report["remediation_steps"]))
         lines.append("")
 
+    if report.get("assumptions"):
+        lines.append("## Assumptions\n")
+        lines.append(str(report["assumptions"]))
+        lines.append("")
+
     return "\n".join(lines)

strix/runtime/caido_bootstrap.py

@@ -10,6 +10,7 @@ exposed-port URL for all subsequent SDK calls.
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
 import logging
 from typing import TYPE_CHECKING
@@ -79,23 +80,72 @@ async def _login_as_guest(
     raise RuntimeError(f"loginAsGuest failed after {attempts} attempts: {last_err}")
 
 
-async def bootstrap_caido(
+async def _aclose_quietly(client: Client) -> None:
+    """Best-effort close of a client whose setup failed; never raises."""
+    with contextlib.suppress(Exception):
+        await client.aclose()
+
+
+async def _connect_client(
     session: BaseSandboxSession,
     *,
     host_url: str,
     container_url: str,
 ) -> Client:
-    """Connect to the in-container Caido sidecar and select a fresh project."""
-    logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url)
-
     access_token = await _login_as_guest(session, container_url=container_url)
-
     client = Client(host_url, auth=TokenAuthOptions(token=access_token))
     await client.connect()
+    return client
 
-    project = await client.project.create(
-        CreateProjectOptions(name="sandbox", temporary=True),
-    )
-    await client.project.select(project.id)
+
+async def bootstrap_caido(
+    session: BaseSandboxSession,
+    *,
+    host_url: str,
+    container_url: str,
+) -> tuple[Client, str]:
+    """Connect to the in-container Caido sidecar and select a fresh project.
+
+    Returns the connected client and the id of the temporary project it
+    selected. The project id lets :func:`reconnect_caido` rebuild a dead
+    transport while staying on the *same* project (and its captured traffic)
+    instead of creating a new empty one.
+    """
+    logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url)
+
+    client = await _connect_client(session, host_url=host_url, container_url=container_url)
+    try:
+        project = await client.project.create(
+            CreateProjectOptions(name="sandbox", temporary=True),
+        )
+        await client.project.select(project.id)
+    except BaseException:
+        # Don't leak the connected transport if project setup fails.
+        await _aclose_quietly(client)
+        raise
     logger.info("Caido project selected: %s", project.id)
+    return client, str(project.id)
+
+
+async def reconnect_caido(
+    session: BaseSandboxSession,
+    *,
+    host_url: str,
+    container_url: str,
+    project_id: str,
+) -> Client:
+    """Rebuild a Caido client after its transport died, keeping the project.
+
+    Re-authenticates, reconnects, and re-selects the existing project so the
+    caller keeps access to the traffic captured before the disconnect.
+    """
+    logger.info("Reconnecting Caido client (host=%s, project=%s)", host_url, project_id)
+    client = await _connect_client(session, host_url=host_url, container_url=container_url)
+    try:
+        await client.project.select(project_id)
+    except BaseException:
+        # A missing/unavailable project must not leave the freshly-connected
+        # transport dangling — otherwise every retry leaks another one.
+        await _aclose_quietly(client)
+        raise
     return client

strix/runtime/docker_client.py

@@ -24,31 +24,134 @@ from __future__ import annotations
 
 import contextlib
 import logging
+import os
 import uuid
-from typing import Any
+from typing import Any, cast
 
+from agents.sandbox.errors import ExposedPortUnavailableError
 from agents.sandbox.manifest import Manifest
 from agents.sandbox.sandboxes.docker import (
     DockerSandboxClient,
+    DockerSandboxSession,
     _build_docker_volume_mounts,
     _docker_port_key,
     _manifest_requires_fuse,
     _manifest_requires_sys_admin,
 )
 from agents.sandbox.session.sandbox_session import SandboxSession
+from agents.sandbox.types import ExposedPortEndpoint
 from docker import errors as docker_errors  # type: ignore[import-untyped, unused-ignore]
 from docker.models.containers import Container  # type: ignore[import-untyped, unused-ignore]
+from docker.types import LogConfig  # type: ignore[import-untyped, unused-ignore]
 from docker.types import Mount as DockerSDKMount  # type: ignore[import-untyped, unused-ignore]
 from docker.utils import parse_repository_tag  # type: ignore[import-untyped, unused-ignore]
+from requests.exceptions import RequestException
 
 
 logger = logging.getLogger(__name__)
 
 
+_SANDBOX_NETWORK_ENV = "STRIX_DOCKER_SANDBOX_NETWORK"
+
+
+def _sandbox_network() -> str | None:
+    value = os.environ.get(_SANDBOX_NETWORK_ENV, "").strip()
+    return value or None
+
+
+def _apply_sandbox_network(create_kwargs: dict[str, Any]) -> None:
+    network = _sandbox_network()
+    if network:
+        create_kwargs["network"] = network
+        create_kwargs.pop("ports", None)
+
+
+def _apply_resource_limits(create_kwargs: dict[str, Any]) -> None:
+    """Apply optional cgroup resource caps from the environment. Unset/blank
+    values leave docker's default (unbounded), so this is opt-in per host."""
+    mem_limit = os.environ.get("STRIX_SANDBOX_MEM_LIMIT", "").strip()
+    if mem_limit:
+        create_kwargs["mem_limit"] = mem_limit
+
+    shm_size = os.environ.get("STRIX_SANDBOX_SHM_SIZE", "").strip()
+    if shm_size:
+        create_kwargs["shm_size"] = shm_size
+
+    cpus = os.environ.get("STRIX_SANDBOX_CPUS", "").strip()
+    if cpus:
+        with contextlib.suppress(ValueError, OverflowError):
+            nano_cpus = int(float(cpus) * 1_000_000_000)
+            if 0 < nano_cpus <= 2**63 - 1:
+                create_kwargs["nano_cpus"] = nano_cpus
+
+    pids_limit = os.environ.get("STRIX_SANDBOX_PIDS_LIMIT", "").strip()
+    if pids_limit:
+        with contextlib.suppress(ValueError):
+            create_kwargs["pids_limit"] = int(pids_limit)
+
+
+def _apply_log_limits(create_kwargs: dict[str, Any]) -> None:
+    """Bound the container's json-file log so a runaway process in the sandbox
+    (e.g. a tool that busy-loops writing to stdout) cannot fill the host disk
+    and take the Docker daemon down with it.
+
+    Unlike the cgroup caps above, this defaults **on** — docker's own default
+    is an unbounded json-file, which is unsafe for an autonomous agent that
+    executes arbitrary commands. ``max-file`` rotation means the on-disk cap is
+    ``max-size * max-file``. Set ``STRIX_SANDBOX_LOG_MAX_SIZE`` to ``0``/``off``
+    to opt back out to docker's default."""
+    max_size = os.environ.get("STRIX_SANDBOX_LOG_MAX_SIZE", "50m").strip()
+    if max_size.lower() in ("0", "off", "none", "unlimited"):
+        return
+    max_file = os.environ.get("STRIX_SANDBOX_LOG_MAX_FILE", "3").strip() or "3"
+    create_kwargs["log_config"] = LogConfig(
+        type=LogConfig.types.JSON,
+        config={"max-size": max_size, "max-file": max_file},
+    )
+
+
+class StrixDockerSandboxSession(DockerSandboxSession):
+    sandbox_network: str = ""
+
+    async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
+        try:
+            self._container.reload()
+        except docker_errors.APIError as e:
+            raise ExposedPortUnavailableError(
+                port=port,
+                exposed_ports=self.state.exposed_ports,
+                reason="backend_unavailable",
+                context={
+                    "backend": "docker",
+                    "detail": "container_reload_failed",
+                    "network": self.sandbox_network,
+                },
+                cause=e,
+            ) from e
+
+        attrs = getattr(self._container, "attrs", {}) or {}
+        networks = attrs.get("NetworkSettings", {}).get("Networks", {})
+        endpoint = networks.get(self.sandbox_network) or {}
+        ip = endpoint.get("IPAddress") or endpoint.get("GlobalIPv6Address")
+        if not isinstance(ip, str) or not ip:
+            raise ExposedPortUnavailableError(
+                port=port,
+                exposed_ports=self.state.exposed_ports,
+                reason="backend_unavailable",
+                context={
+                    "backend": "docker",
+                    "detail": "container_not_on_network",
+                    "network": self.sandbox_network,
+                },
+            )
+        host = f"[{ip}]" if ":" in ip else ip
+        return ExposedPortEndpoint(host=host, port=port, tls=False)
+
+
 class StrixDockerSandboxClient(DockerSandboxClient):
     # Host directories to bind-mount into the container, set by the docker
     # backend before ``create()``. Each item is ``{source, target, read_only}``.
-    strix_bind_mounts: list[dict[str, Any]] = []  # overridden per-instance in backends.py
+    strix_bind_mounts: list[dict[str, Any]] | None = None
 
     async def _create_container(
         self,
@@ -116,6 +219,10 @@ class StrixDockerSandboxClient(DockerSandboxClient):
         extra_hosts = create_kwargs.setdefault("extra_hosts", {})
         extra_hosts["host.docker.internal"] = "host-gateway"
 
+        _apply_sandbox_network(create_kwargs)
+        _apply_resource_limits(create_kwargs)
+        _apply_log_limits(create_kwargs)
+
         # Strix injection: host bind mounts (e.g. large repos passed via --mount)
         # that bypass the SDK's file-by-file LocalDir copy.
         bind_mounts = getattr(self, "strix_bind_mounts", ())
@@ -145,9 +252,27 @@ class StrixDockerSandboxClient(DockerSandboxClient):
         )
         return container
 
+    async def create(self, **kwargs: Any) -> SandboxSession:
+        session = await super().create(**kwargs)
+        network = _sandbox_network()
+        inner = session._inner
+        if network and isinstance(inner, DockerSandboxSession):
+            inner.__class__ = StrixDockerSandboxSession
+            cast("StrixDockerSandboxSession", inner).sandbox_network = network
+        return session
+
     async def delete(self, session: SandboxSession) -> SandboxSession:
         container_id = getattr(getattr(session._inner, "state", None), "container_id", None)
         if container_id:
-            with contextlib.suppress(docker_errors.NotFound, docker_errors.APIError):
+            # Best-effort kill: NotFound/APIError cover a gone or unhappy
+            # container. RequestException covers a torn-down daemon socket —
+            # containers.get() -> inspect_container raises requests'
+            # ConnectionError, which is a sibling of docker.errors.APIError
+            # under requests.RequestException (not a subclass), so it escapes
+            # an APIError-only suppress and surfaces a full traceback even
+            # though this teardown is meant to be best-effort.
+            with contextlib.suppress(
+                docker_errors.NotFound, docker_errors.APIError, RequestException
+            ):
                 self.docker_client.containers.get(container_id).kill()
         return await super().delete(session)

strix/runtime/local_dir_staging.py

@@ -0,0 +1,120 @@
+"""Symlink-safe staging for ``LocalDir`` manifest uploads.
+
+The sandbox SDK's ``LocalDir`` walker refuses to copy symlinks at all — it
+raises ``LocalDirReadError(reason="symlink_not_supported")`` on the first one
+as a path-escape / TOCTOU safeguard. Real source trees (especially JS/TS
+monorepos with workspace or shared-config links) routinely commit symlinks, so
+handing such a tree straight to ``LocalDir`` aborts the upload before the agent
+even starts.
+
+:func:`stage_symlink_safe_dir` returns a path that is always safe to hand to
+``LocalDir``:
+
+* a tree with no symlinks is used as-is (no copy);
+* otherwise the tree is copied into a temp directory with symlinks resolved:
+
+  - a link whose target stays inside the tree is *dereferenced* (its target
+    content is materialized in place), so the agent still sees the file;
+  - a link that escapes the tree, dangles, or forms a cycle is *dropped* and
+    never followed. Refusing to follow out-of-tree links preserves the walker's
+    path-escape safety and keeps host/out-of-tree content from leaking into the
+    (hostile) sandbox.
+
+Regular files are hard-linked when possible (falling back to a copy across
+devices), so the staged tree adds negligible disk for the non-symlink bulk.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import shutil
+import tempfile
+from pathlib import Path
+
+
+logger = logging.getLogger(__name__)
+
+_STAGING_PREFIX = "strix-localdir-"
+
+
+def _is_within(target: Path, root: Path) -> bool:
+    """Return whether ``target`` is ``root`` itself or nested under it."""
+    if target == root:
+        return True
+    try:
+        target.relative_to(root)
+    except ValueError:
+        return False
+    return True
+
+
+def tree_has_symlink(root: Path) -> bool:
+    """Return whether ``root`` contains any symlink (file or directory)."""
+    for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
+        base = Path(dirpath)
+        for name in (*dirnames, *filenames):
+            if (base / name).is_symlink():
+                return True
+    return False
+
+
+def _link_or_copy(src: Path, dst: Path) -> None:
+    """Hard-link ``src`` to ``dst``, falling back to a content copy."""
+    try:
+        os.link(src, dst)
+    except OSError:
+        shutil.copy2(src, dst, follow_symlinks=True)
+
+
+def _stage_dir(src: Path, dst: Path, root: Path, seen: frozenset[Path]) -> None:
+    dst.mkdir(parents=True, exist_ok=True)
+    for entry in os.scandir(src):
+        entry_path = Path(entry.path)
+        dest_path = dst / entry.name
+
+        if entry.is_symlink():
+            target = Path(os.path.realpath(entry_path))
+            if not _is_within(target, root):
+                logger.warning("staging: dropping out-of-tree symlink %s -> %s", entry_path, target)
+                continue
+            if not target.exists():
+                logger.warning("staging: dropping dangling symlink %s", entry_path)
+                continue
+            if target in seen:
+                logger.warning("staging: dropping cyclic symlink %s -> %s", entry_path, target)
+                continue
+            if target.is_dir():
+                _stage_dir(target, dest_path, root, seen | {target})
+            else:
+                _link_or_copy(target, dest_path)
+        elif entry.is_dir(follow_symlinks=False):
+            _stage_dir(entry_path, dest_path, root, seen)
+        elif entry.is_file(follow_symlinks=False):
+            _link_or_copy(entry_path, dest_path)
+        else:
+            # Sockets, FIFOs, devices — not part of a source tree; skip.
+            logger.debug("staging: skipping non-regular entry %s", entry_path)
+
+
+def stage_symlink_safe_dir(src_root: Path) -> tuple[Path, Path | None]:
+    """Return ``(upload_path, staged_temp)`` for uploading ``src_root``.
+
+    ``upload_path`` is safe to hand to ``LocalDir``. When the tree contains no
+    symlinks it is ``src_root`` itself and ``staged_temp`` is ``None``.
+    Otherwise a symlink-safe copy is materialized in a temp directory and both
+    returned values point at it; the caller owns removing ``staged_temp`` once
+    the upload completes.
+    """
+    root = src_root.resolve()
+    if not tree_has_symlink(root):
+        return root, None
+
+    staged = Path(tempfile.mkdtemp(prefix=_STAGING_PREFIX))
+    try:
+        _stage_dir(root, staged, root, frozenset({root}))
+    except OSError:
+        shutil.rmtree(staged, ignore_errors=True)
+        raise
+    logger.info("staging: materialized symlink-safe copy of %s at %s", root, staged)
+    return staged, staged

strix/runtime/session_manager.py

@@ -3,15 +3,22 @@
 from __future__ import annotations
 
 import logging
+import shutil
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from agents.sandbox.entries import BaseEntry, LocalDir
 from agents.sandbox.manifest import Environment, Manifest
 
 from strix.config import load_settings
 from strix.runtime.backends import get_backend
-from strix.runtime.caido_bootstrap import bootstrap_caido
+from strix.runtime.caido_bootstrap import bootstrap_caido, reconnect_caido
+from strix.runtime.local_dir_staging import stage_symlink_safe_dir
+from strix.tools.proxy.caido_api import SharedCaidoClient
+
+
+if TYPE_CHECKING:
+    from caido_sdk_client import Client as CaidoClient
 
 
 logger = logging.getLogger(__name__)
@@ -29,16 +36,20 @@ _WORKSPACE_ROOT = "/workspace"
 
 def build_session_entries(
     local_sources: list[dict[str, Any]],
-) -> tuple[dict[str | Path, BaseEntry], list[dict[str, Any]]]:
+) -> tuple[dict[str | Path, BaseEntry], list[dict[str, Any]], list[Path]]:
     """Split local sources into copied manifest entries and host bind mounts.
 
     Sources flagged ``mount`` are bind-mounted read-only at
     ``/workspace/<workspace_subdir>`` (not added to the manifest, so the SDK
     does not stream them in file-by-file). Every other source becomes a
-    ``LocalDir`` entry copied into the container as before.
+    ``LocalDir`` entry copied into the container as before. Trees containing
+    symlinks (which the SDK's ``LocalDir`` walker refuses outright) are first
+    staged into a symlink-safe temp copy; those temp dirs are returned so the
+    caller can remove them once the upload completes.
     """
     entries: dict[str | Path, BaseEntry] = {}
     bind_mounts: list[dict[str, Any]] = []
+    staged_dirs: list[Path] = []
     for src in local_sources:
         ws_subdir = src.get("workspace_subdir") or ""
         host_path = src.get("source_path") or ""
@@ -54,8 +65,11 @@ def build_session_entries(
                 }
             )
         else:
-            entries[ws_subdir] = LocalDir(src=resolved)
-    return entries, bind_mounts
+            upload_path, staged = stage_symlink_safe_dir(resolved)
+            if staged is not None:
+                staged_dirs.append(staged)
+            entries[ws_subdir] = LocalDir(src=upload_path)
+    return entries, bind_mounts, staged_dirs
 
 
 async def create_or_reuse(
@@ -75,7 +89,7 @@ async def create_or_reuse(
         logger.info("Reusing existing sandbox session for scan %s", scan_id)
         return cached
 
-    entries, bind_mounts = build_session_entries(local_sources)
+    entries, bind_mounts, staged_dirs = build_session_entries(local_sources)
 
     # Caido runs as an in-container sidecar; HTTP(S) traffic from any
     # process started via ``session.exec`` (the SDK's Shell tool, etc.)
@@ -106,27 +120,40 @@ async def create_or_reuse(
         backend_name,
         image,
     )
-    client, session = await backend(
-        image=image,
-        manifest=manifest,
-        exposed_ports=(_CONTAINER_CAIDO_PORT,),
-        bind_mounts=bind_mounts,
-    )
+    try:
+        client, session = await backend(
+            image=image,
+            manifest=manifest,
+            exposed_ports=(_CONTAINER_CAIDO_PORT,),
+            bind_mounts=bind_mounts,
+        )
+    finally:
+        for staged in staged_dirs:
+            shutil.rmtree(staged, ignore_errors=True)
 
     caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
-    host_caido_url = f"http://{caido_endpoint.host}:{caido_endpoint.port}"
+    scheme = "https" if caido_endpoint.tls else "http"
+    host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
     logger.debug("Caido host endpoint resolved: %s", host_caido_url)
 
-    caido_client = await bootstrap_caido(
+    caido_client, caido_project_id = await bootstrap_caido(
         session,
         host_url=host_caido_url,
         container_url=container_caido_url,
     )
 
+    async def _reconnect_caido() -> CaidoClient:
+        return await reconnect_caido(
+            session,
+            host_url=host_caido_url,
+            container_url=container_caido_url,
+            project_id=caido_project_id,
+        )
+
     bundle = {
         "client": client,
         "session": session,
-        "caido_client": caido_client,
+        "caido_client": SharedCaidoClient(caido_client, _reconnect_caido),
     }
     _SESSION_CACHE[scan_id] = bundle
     logger.info("Sandbox session for scan %s ready and cached", scan_id)
@@ -153,11 +180,19 @@ async def cleanup(scan_id: str) -> None:
         except Exception:  # noqa: BLE001
             logger.debug("cleanup(%s): caido_client.aclose() raised", scan_id, exc_info=True)
 
+    client = bundle["client"]
     try:
-        await bundle["client"].delete(bundle["session"])
+        await client.delete(bundle["session"])
         logger.info("Cleaned up sandbox session for scan %s", scan_id)
     except Exception:
         logger.exception(
             "cleanup(%s): client.delete raised; container may need manual reaping",
             scan_id,
         )
+
+    docker_client = getattr(client, "docker_client", None)
+    if docker_client is not None:
+        try:
+            docker_client.close()
+        except Exception:  # noqa: BLE001
+            logger.debug("cleanup(%s): docker_client.close() raised", scan_id, exc_info=True)

strix/skills/README.md

@@ -41,6 +41,7 @@ The skills are dynamically injected into the agent's system prompt, allowing it
 Notable source-aware skills:
 - `source_aware_whitebox` (coordination): white-box orchestration playbook
 - `source_aware_sast` (custom): semgrep/AST/secrets/supply-chain static triage workflow
+- `dependency_cve_scanning` (custom): trivy-based SCA workflow for reporting known dependency CVEs via `create_dependency_report`
 
 ---
 

strix/skills/__init__.py

@@ -1,7 +1,11 @@
 import logging
 import re
+import threading
+from collections import Counter
 from collections.abc import Iterator
+from pathlib import Path
 
+from strix.telemetry import posthog, scarf
 from strix.utils.resource_paths import get_strix_resource_path
 
 
@@ -10,20 +14,82 @@ logger = logging.getLogger(__name__)
 _FRONTMATTER_PATTERN = re.compile(r"^---\s*\n.*?\n---\s*\n", re.DOTALL)
 
 _INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordination"})
+_ROOT_SKILL_CATEGORY = "root"
+
+_EXTRA_SKILL_DIRS: list[Path] = []
+
+
+def register_skill_dir(path: str | Path) -> None:
+    """Add a directory searched for skills ahead of the built-in set.
+
+    The directory uses the same layout as the packaged skills
+    (``<root>/<category>/<name>.md``). Skills found in a registered
+    directory shadow packaged skills with the same relative path, so
+    callers can both add new skills and override existing ones without
+    editing the package. The most recently registered directory has the
+    highest precedence.
+    """
+    resolved = Path(path)
+    if resolved not in _EXTRA_SKILL_DIRS:
+        _EXTRA_SKILL_DIRS.append(resolved)
+        logger.info("Registered extra skill dir: %s", resolved)
+
+
+def registered_skill_dirs() -> tuple[Path, ...]:
+    """Return registered extra skill directories, highest precedence first."""
+    return tuple(reversed(_EXTRA_SKILL_DIRS))
+
+
+def skill_search_dirs() -> tuple[Path, ...]:
+    """All existing skill roots, highest precedence first (built-in last)."""
+    roots = [d for d in registered_skill_dirs() if d.is_dir()]
+    builtin = get_strix_resource_path("skills")
+    if builtin.is_dir():
+        roots.append(builtin)
+    return tuple(roots)
 
 
 def _iter_user_skill_files() -> Iterator[tuple[str, str]]:
     """Yield ``(category_name, skill_name)`` for every user-selectable skill."""
-    skills_dir = get_strix_resource_path("skills")
-    if not skills_dir.exists():
-        return
-    for category_dir in sorted(skills_dir.iterdir()):
-        if not category_dir.is_dir() or category_dir.name.startswith("__"):
-            continue
-        if category_dir.name in _INTERNAL_SKILL_CATEGORIES:
-            continue
-        for file_path in sorted(category_dir.glob("*.md")):
-            yield category_dir.name, file_path.stem
+    seen: set[tuple[str, str]] = set()
+    for skills_dir in skill_search_dirs():
+        for file_path in sorted(skills_dir.glob("*.md")):
+            if file_path.name.startswith("__") or file_path.name == "README.md":
+                continue
+            key = (_ROOT_SKILL_CATEGORY, file_path.stem)
+            if key in seen:
+                continue
+            seen.add(key)
+            yield key
+
+        for category_dir in sorted(skills_dir.iterdir()):
+            if not category_dir.is_dir() or category_dir.name.startswith("__"):
+                continue
+            if category_dir.name in _INTERNAL_SKILL_CATEGORIES:
+                continue
+            for file_path in sorted(category_dir.glob("*.md")):
+                key = (category_dir.name, file_path.stem)
+                if key in seen:
+                    continue
+                seen.add(key)
+                yield key
+
+
+def _is_selectable_root_skill_file(file_path: Path) -> bool:
+    return file_path.suffix == ".md" and not (
+        file_path.name.startswith("__") or file_path.name == "README.md"
+    )
+
+
+def _qualified_skill_file(skills_dir: Path, category: str, name: str) -> Path | None:
+    if category == _ROOT_SKILL_CATEGORY:
+        candidate = skills_dir / f"{name}.md"
+        if candidate.exists() and _is_selectable_root_skill_file(candidate):
+            return candidate
+        return None
+
+    candidate = skills_dir / category / f"{name}.md"
+    return candidate if candidate.exists() else None
 
 
 def get_all_skill_names() -> set[str]:
@@ -31,6 +97,54 @@ def get_all_skill_names() -> set[str]:
     return {name for _, name in _iter_user_skill_files()}
 
 
+def _get_all_skill_keys() -> set[str]:
+    keys: set[str] = set()
+    for category, name in _iter_user_skill_files():
+        keys.add(f"{category}/{name}")
+    return keys
+
+
+def _get_ambiguous_skill_names() -> set[str]:
+    counts = Counter(name for _, name in _iter_user_skill_files())
+    return {name for name, count in counts.items() if count > 1}
+
+
+def _qualified_skill_files(skill_name: str) -> list[Path]:
+    category, _, name = skill_name.partition("/")
+    for skills_dir in skill_search_dirs():
+        candidate = _qualified_skill_file(skills_dir, category, name)
+        if candidate is not None:
+            return [candidate]
+    return []
+
+
+def _bare_skill_files(skill_name: str) -> list[Path]:
+    seen: set[tuple[str, str]] = set()
+    candidates: list[Path] = []
+    for skills_dir in skill_search_dirs():
+        for category_dir in sorted(skills_dir.iterdir()):
+            if not category_dir.is_dir() or category_dir.name.startswith("__"):
+                continue
+            if category_dir.name in _INTERNAL_SKILL_CATEGORIES:
+                continue
+            key = (category_dir.name, skill_name)
+            if key in seen:
+                continue
+            candidate = category_dir / f"{skill_name}.md"
+            if candidate.exists():
+                seen.add(key)
+                candidates.append(candidate)
+
+        key = (_ROOT_SKILL_CATEGORY, skill_name)
+        if key in seen:
+            continue
+        root_candidate = _qualified_skill_file(skills_dir, _ROOT_SKILL_CATEGORY, skill_name)
+        if root_candidate is not None:
+            seen.add(key)
+            candidates.append(root_candidate)
+    return candidates
+
+
 def get_available_skills() -> dict[str, list[str]]:
     grouped: dict[str, list[str]] = {}
     for category, name in _iter_user_skill_files():
@@ -52,48 +166,63 @@ def validate_requested_skills(skill_list: list[str], max_skills: int = 5) -> str
     if not skill_list:
         return None
     available = get_all_skill_names()
-    invalid = sorted({s for s in skill_list if s not in available})
+    available_keys = _get_all_skill_keys()
+    invalid = sorted({s for s in skill_list if s not in available and s not in available_keys})
     if invalid:
         return f"Invalid skill name(s): {invalid}. Available skills: {sorted(available)}"
+    ambiguous = sorted({s for s in skill_list if "/" not in s} & _get_ambiguous_skill_names())
+    if ambiguous:
+        return (
+            f"Ambiguous skill name(s): {ambiguous}. Use category-qualified names from: "
+            f"{sorted(available_keys)}"
+        )
     return None
 
 
+def _track_skill_loaded(skill_name: str, file_path: Path) -> None:
+    builtin = get_strix_resource_path("skills")
+    if not file_path.is_relative_to(builtin):
+        skill_name = "custom"
+
+    def _send() -> None:
+        posthog.skill_loaded(skill_name)
+        scarf.skill_loaded(skill_name)
+
+    threading.Thread(target=_send, daemon=True).start()
+
+
+def _candidate_skill_files(skill_name: str) -> list[Path]:
+    """Resolve *skill_name* to effective matching files."""
+    if "/" in skill_name:
+        return _qualified_skill_files(skill_name)
+    return _bare_skill_files(skill_name)
+
+
 def load_skills(skill_names: list[str]) -> dict[str, str]:
     """Load skill markdown bodies (frontmatter stripped) by name.
 
-    Skill files live at ``strix/skills/<category>/<name>.md``. Names
-    can be ``"name"`` (any category), ``"category/name"``, or a bare
-    file at the skills root. Missing skills are logged and skipped.
+    Skill files live at ``strix/skills/<category>/<name>.md`` (or any
+    directory added via :func:`register_skill_dir`, searched first).
+    Names can be ``"name"`` (any category), ``"category/name"``, or a
+    bare file at the skills root. Missing skills are logged and skipped.
     """
-    skills_dir = get_strix_resource_path("skills")
-    if not skills_dir.exists():
+    search_dirs = skill_search_dirs()
+    if not search_dirs:
         return {}
 
-    by_category: dict[str, str] = {}
-    for category_dir in skills_dir.iterdir():
-        if not category_dir.is_dir() or category_dir.name.startswith("__"):
-            continue
-        for file_path in category_dir.glob("*.md"):
-            by_category[file_path.stem] = f"{category_dir.name}/{file_path.stem}.md"
-
     skill_content: dict[str, str] = {}
     for skill_name in skill_names:
-        rel_path: str | None
-        if "/" in skill_name:
-            rel_path = f"{skill_name}.md"
-        elif skill_name in by_category:
-            rel_path = by_category[skill_name]
-        elif (skills_dir / f"{skill_name}.md").exists():
-            rel_path = f"{skill_name}.md"
-        else:
-            rel_path = None
-
-        if rel_path is None or not (skills_dir / rel_path).exists():
+        candidates = _candidate_skill_files(skill_name)
+        if not candidates:
             logger.warning("Skill not found: %s", skill_name)
             continue
+        if len(candidates) > 1:
+            logger.warning("Ambiguous skill name %s; use a category-qualified name", skill_name)
+            continue
+        file_path = candidates[0]
 
         try:
-            content = (skills_dir / rel_path).read_text(encoding="utf-8")
+            content = file_path.read_text(encoding="utf-8")
         except (OSError, ValueError) as e:
             logger.warning("Failed to load skill %s: %s", skill_name, e)
             continue
@@ -101,6 +230,7 @@ def load_skills(skill_names: list[str]) -> dict[str, str]:
         var_name = skill_name.split("/")[-1]
         skill_content[var_name] = _FRONTMATTER_PATTERN.sub("", content).lstrip()
         logger.debug("Loaded skill: %s -> %s", skill_name, var_name)
+        _track_skill_loaded(var_name, file_path)
 
     logger.debug("load_skills: %d skill(s) resolved", len(skill_content))
     return skill_content

strix/skills/cloud/gcp.md

@@ -0,0 +1,194 @@
+---
+name: gcp
+description: GCP cloud security testing covering IAM misconfigurations, public storage buckets, metadata abuse, and service account privilege escalation
+---
+
+# Google Cloud Platform (GCP)
+
+GCP misconfigurations expose project data, service account keys, and lateral movement paths across Compute, Cloud Storage, Cloud Functions, and GKE. This skill covers direct GCP API testing and post-compromise enumeration from VMs/containers. For SSRF-mediated metadata access, combine with the `ssrf` skill.
+
+## Attack Surface
+
+**Identity**
+- IAM policies: project/folder/org level bindings
+- Service accounts, keys (JSON), Workload Identity, impersonation
+- OAuth scopes on compute instances and Cloud Functions
+
+**Storage & Data**
+- Cloud Storage (GCS) buckets and objects
+- BigQuery datasets, Cloud SQL instances, Firestore (see `firebase_firestore` skill)
+- Secret Manager, Cloud KMS keys
+
+**Compute**
+- Compute Engine VMs, Cloud Run, Cloud Functions, GKE clusters
+- Metadata server at `http://metadata.google.internal/computeMetadata/v1/`
+- Startup scripts, instance templates, custom images
+
+**Management**
+- Cloud Console, gcloud CLI, Deployment Manager, Terraform state buckets
+- Cloud Logging, Error Reporting, Cloud Build triggers
+
+## Reconnaissance
+
+**Credential Discovery**
+- Service account JSON keys in repos, CI/CD, `.env`, backup buckets
+- `GOOGLE_APPLICATION_CREDENTIALS` environment variable
+- Default Compute Engine service account on VMs (often overprivileged)
+- OAuth tokens in browser/local `gcloud` config (`~/.config/gcloud/`)
+
+**Unauthenticated Enumeration**
+
+Avoid `gsutil` for anonymous checks — it can use ambient `gcloud` or application-default credentials and produce false public-bucket findings. Unset `GOOGLE_APPLICATION_CREDENTIALS` and use unauthenticated HTTP instead.
+
+```
+# GCS bucket existence (403 = exists but private, 404 = not found/wrong region)
+curl -I https://storage.googleapis.com/target-bucket/
+
+# Anonymous listing (no Authorization header; confirms allUsers/allAuthenticatedUsers List)
+curl https://storage.googleapis.com/target-bucket/
+
+# Alternate URL forms
+curl -I https://target-bucket.storage.googleapis.com/
+```
+
+**Authenticated Enumeration**
+```
+gcloud auth list
+gcloud config get-value project
+gcloud projects get-iam-policy PROJECT_ID
+gcloud iam service-accounts list
+gcloud storage ls
+gcloud compute instances list
+gcloud container clusters list
+```
+
+## Key Vulnerabilities
+
+### Cloud Storage Misconfigurations
+
+- Public buckets: `allUsers` or `allAuthenticatedUsers` with `roles/storage.objectViewer` or `objectAdmin`
+- Listable buckets revealing object keys: backups, `.env`, `terraform.tfstate`, SA keys
+- Uniform bucket-level access disabled with legacy ACL public-read
+- Signed URL with excessive TTL or overly broad object prefix
+
+**Test:**
+```
+gsutil iam get gs://BUCKET          # requires credentials
+curl https://storage.googleapis.com/BUCKET/   # anonymous listing check
+curl -I https://storage.googleapis.com/BUCKET/sensitive.sql
+```
+
+### IAM Privilege Escalation
+
+Common escalation paths (verify with `gcloud iam` / policy simulator):
+
+| Permission | Escalation |
+|------------|------------|
+| `iam.serviceAccounts.actAs` + `compute.instances.create` | VM with privileged SA |
+| `iam.serviceAccountKeys.create` | Export key for higher-priv SA |
+| `iam.serviceAccounts.setIamPolicy` | Grant yourself roles on SA |
+| `cloudfunctions.functions.create` + `actAs` | Deploy function as privileged SA |
+| `run.services.create` (Cloud Run) + `actAs` | Deploy service with admin SA |
+| `storage.buckets.update` + `setIamPolicy` | Open bucket to public or self |
+
+**Test:**
+```
+gcloud projects get-iam-policy PROJECT --flatten="bindings[].members" --filter="bindings.members:user:YOU"
+gcloud iam roles list --project=PROJECT
+```
+
+### Metadata Server Abuse
+
+From any code execution on a GCP VM, Cloud Run (if metadata accessible), or compromised pod:
+
+```
+curl -H "Metadata-Flavor: Google" \
+  http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token
+
+curl -H "Metadata-Flavor: Google" \
+  http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/email
+```
+
+- Default compute SA may have `editor` role on project (legacy projects)
+- Requested OAuth scopes may allow `cloud-platform` full access
+- Workload Identity misconfiguration in GKE → cross-namespace SA token theft
+
+### GKE Misconfigurations
+
+- Dashboard/UI exposed, anonymous RBAC (see `kubernetes` skill for K8s layer)
+- Workload Identity not enforced; pods use node SA with broad GCP permissions
+- `kubectl` proxy or `kubelet` read-only port exposed
+- Secrets in ConfigMaps; GCR/Artifact Registry images pulling without auth
+
+### Cloud Functions / Cloud Run
+
+- HTTP-triggered functions without authentication (`--allow-unauthenticated`)
+- Environment variables containing API keys (`gcloud functions describe`)
+- Overprivileged runtime service account (`roles/editor`)
+- Event triggers accepting attacker-controlled Pub/Sub messages
+
+### BigQuery & Cloud SQL
+
+- Public datasets (`allUsers` on dataset IAM)
+- Cloud SQL public IP with weak/no password
+- Exported snapshots in public GCS buckets
+
+### Secret Manager & KMS
+
+- `secretmanager.versions.access` granted to unintended principals
+- Secrets replicated to logs via misconfigured Cloud Functions env vars
+- KMS cryptoKey IAM with `allAuthenticatedUsers`
+
+## Advanced Techniques
+
+**Terraform State in GCS**
+- `terraform.tfstate` in listable bucket → all resource addresses, sometimes secrets in plain text
+
+**Service Account Impersonation Chain**
+- `roles/iam.serviceAccountTokenCreator` on target SA → short-lived access tokens
+
+**Org/Fold Policy Gaps**
+- Project-level deny policies not applied; child project inherits permissive folder IAM
+
+## Testing Methodology
+
+1. **Discover credentials** — Keys in code, metadata, SSRF, public buckets
+2. **Identify principal** — `gcloud auth list`, effective project IAM
+3. **Enumerate storage** — Public/listable buckets, sensitive object names
+4. **Escalation paths** — Map `actAs`, key creation, function deploy permissions
+5. **Metadata** — From any shell in GCP workload, fetch SA token and scopes
+6. **GKE layer** — Pivot from GCP IAM to cluster (combine with `kubernetes` skill)
+
+## Validation
+
+1. Demonstrate unauthorized GCS object read/list with bucket URL and object key
+2. Show IAM escalation path with exact role/member binding and resulting access
+3. Prove metadata token theft from compute context with redacted token scope
+4. Document project ID, resource name, and IAM binding root cause
+5. Confirm fix blocks the specific principal/permission/resource combination
+
+## False Positives
+
+- Intentionally public static asset bucket with no sensitive objects
+- Metadata server unreachable from tested context (no RCE/SSRF)
+- SA token from metadata has only `devstorage.read_only` on single bucket (note scope, not full breach)
+- `403` on bucket HEAD indicating existence but not readable content
+
+## Impact
+
+- Mass data exfiltration from GCS/BigQuery/Cloud SQL backups
+- Project or org compromise via SA key theft or IAM escalation
+- Lateral movement from GKE pod to cloud control plane
+- Regulatory exposure (PII in public buckets or exports)
+
+## Pro Tips
+
+1. Always check both `gsutil iam get` and anonymous `curl` — IAM and ACL layers differ
+2. Search public buckets for `*.json` service account keys and `terraform.tfstate`
+3. Default compute SA email: `PROJECT_NUMBER-compute@developer.gserviceaccount.com`
+4. Combine with `kubernetes` skill when target runs on GKE
+5. Firebase-hosted apps often use GCP project underneath — pivot from web to GCP project ID in configs
+
+## Summary
+
+GCP security requires least-privilege IAM, no public data paths, tight metadata/scopes on compute, and protected service account keys. Enumerate from any credential or shell — even read-only GCS access often reveals escalation artifacts.

strix/skills/custom/dependency_cve_scanning.md

@@ -0,0 +1,138 @@
+---
+name: dependency-cve-scanning
+description: Supply-chain / SCA playbook — scan repository lockfiles for known dependency CVEs and report them with create_dependency_report (no dynamic PoC required)
+---
+
+# Dependency / Supply-Chain CVE Scanning (SCA)
+
+Use this skill on white-box / repository scans to make sure a repository pinning a
+**known-vulnerable dependency** is actually reported as a finding, instead of being
+discovered and then silently dropped because it cannot be dynamically exploited.
+
+Known-CVE dependency findings are a first-class deliverable. Report each one with
+the dedicated `create_dependency_report` tool.
+
+## Why this skill exists
+
+A vulnerable dependency pinned in a lockfile (e.g. `lodash@4.17.4` with a known
+prototype-pollution CVE) usually cannot be dynamically PoC'd from the outside —
+the vulnerable code path may not even be reachable from a running endpoint. The
+normal "no report without a dynamic PoC" rule would suppress it. For these
+findings the proof is the **lockfile entry + scanner output + published
+advisory**, not an exploit script. This is the one explicit exception to the
+dynamic-validation rule, and it exists only for `create_dependency_report`.
+
+## Scan procedure
+
+Run from the repo root and store output in the shared artifact directory used by
+the source-aware pass:
+
+```bash
+ART=/workspace/.strix-source-aware
+mkdir -p "$ART"
+
+# Record the vuln DB age so a stale DB is a visible signal, not a silent clean scan.
+trivy version --format json 2>/dev/null | tee "$ART/trivy-version.json"
+# inspect .VulnerabilityDB.UpdatedAt / NextUpdate
+
+# Lockfile/manifest -> known-CVE matching. Try a best-effort DB refresh first so a
+# sandbox with egress gets the freshest CVEs; if the update fails, fall back to the
+# cached DB instead of failing the scan. --offline-scan keeps per-package advisory
+# lookups offline.
+trivy fs --scanners vuln --timeout 30m --offline-scan \
+  --format json --output "$ART/trivy-sca.json" . \
+  || trivy fs --scanners vuln --timeout 30m --offline-scan --skip-db-update \
+       --format json --output "$ART/trivy-sca.json" . \
+  || true
+```
+
+If `.VulnerabilityDB.UpdatedAt` is more than a few weeks old (the sandbox had no
+egress to refresh it), treat it as a scan limitation and note it in the
+`assumptions` of dependency findings — a stale DB that still returns *some* results
+will not trip the "zero results is suspicious" heuristic, so its age is the only
+staleness signal.
+
+Trivy reads the lockfiles/manifests it finds, including:
+`package-lock.json`, `yarn.lock`, `pnpm-lock.yaml`, `poetry.lock`,
+`requirements.txt`, `Pipfile.lock`, `go.mod`/`go.sum`, `Gemfile.lock`,
+`pom.xml`/`gradle.lockfile`, `Cargo.lock`, `composer.lock`, etc.
+
+If trivy returns zero vulnerabilities on a repo with dependencies, treat it as
+suspicious: confirm the vuln DB is present (`trivy-version.json`) and that
+lockfiles exist.
+
+## Interpreting results
+
+For each entry under `.Results[].Vulnerabilities[]` in `trivy-sca.json`, collect:
+
+- `VulnerabilityID` — the CVE (or GHSA; prefer the CVE if both are present)
+- `PkgName` and `InstalledVersion` — the affected package + pinned version
+- `FixedVersion` — the version that resolves it
+- `Target` — the lockfile path it came from
+- `.Results[].Type` (e.g. `npm`, `pip`, `gomod`, `pom`, `gemspec`, `cargo`) — the
+  package ecosystem; normalize to the registry name lowercased (`npm`, `pypi`,
+  `go`, `maven`, `rubygems`, `cargo`, `composer`, `nuget`, ...)
+- `CVSS` — the published advisory base score
+- `PrimaryURL` / references — to verify the advisory
+
+Deduplicate by `(CVE, PkgName, InstalledVersion)`. File one
+`create_dependency_report` per CVE — do not batch multiple CVEs into one report.
+
+### Reachability is a confidence modifier, not a gate
+
+Do NOT suppress or downgrade a known CVE just because you could not prove the
+vulnerable code path is reachable. Report it, set `advisory_cvss` from the
+advisory, and use `assumptions` to note reachability (e.g. "the vulnerable
+`template()` API does not appear to be imported in application code, so practical
+exploitability is uncertain"). If you *can* show reachability or chain it into a
+dynamic exploit, do that and report it as a normal dynamic finding with
+`create_vulnerability_report` instead.
+
+## Reporting
+
+Report each confirmed known CVE with the dedicated `create_dependency_report`
+tool (NOT `create_vulnerability_report` — that tool is for dynamically validated
+findings and rejects empty PoC fields):
+
+- Set `cve` to the verified `CVE-YYYY-NNNNN` id (required). If you only have a
+  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).
+- **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).
+  - `installed_version` — `InstalledVersion` (required).
+  - `package_ecosystem` — normalized ecosystem from `.Results[].Type` (lowercased,
+    e.g. `npm`, `pypi`, `go`, `maven`, `rubygems`, `cargo`) (required).
+  - `fixed_version` — `FixedVersion` (leave empty only if no fix is published).
+- Reference the repo-relative `Target` lockfile path in `description` /
+  `technical_analysis` (no leading slash) so the finding is traceable.
+- Put the concrete proof in `description` / `technical_analysis`: package name,
+  installed/affected version, fixed version, lockfile path, and the relevant
+  trivy output excerpt.
+- **Always set `advisory_cvss` to the published advisory base score (0.0–10.0).**
+  Severity is derived *solely* from this number: read it off the advisory (`CVSS`
+  in trivy output, or the NVD/GHSA page) and pass the real value. The tool rejects
+  a call that omits it, because guessing a score both inflates low CVEs and
+  deflates critical ones.
+- Set `cwe` to the most specific `CWE-NNN` when the advisory names one.
+- Do NOT cap severity at LOW just because there is no dynamic reproduction — use
+  the advisory score.
+- Use `assumptions` for reachability/exploitability caveats.
+
+Verify the CVE with `web_search` when available before reporting. Never guess or
+hallucinate a CVE id.
+
+## Anti-patterns
+
+- Do not report a dependency CVE with `create_vulnerability_report`; use
+  `create_dependency_report`.
+- Do not report a finding without a verified CVE id.
+- Do not batch multiple CVEs into one report.
+- Do not omit `advisory_cvss` — the tool rejects it, and it is the single input
+  that determines dependency severity.
+- Do not silently drop a known CVE because it lacks a dynamic PoC — that is the
+  exact failure this skill prevents.
+- Do not downgrade advisory severity for lack of dynamic reproduction.

strix/skills/custom/source_aware_sast.md

@@ -121,6 +121,11 @@ trivy fs --scanners vuln,misconfig --timeout 30m --offline-scan \
   --format json --output /workspace/.strix-source-aware/trivy-fs.json . || true
 ```
 
+Known-CVE dependency findings are the one exception to the "report only after
+dynamic validation" rule below: report each one with `create_dependency_report`
+(not `create_vulnerability_report`), setting `advisory_cvss` from the published
+advisory. `load_skill(["dependency_cve_scanning"])` for the full SCA workflow.
+
 ## JavaScript-Side Coverage
 
 For frontends and Node services, layer these on top of the language-agnostic

strix/skills/technologies/auth0.md

@@ -0,0 +1,188 @@
+---
+name: auth0
+description: Auth0 tenant security testing covering misconfigured rules/actions, scope escalation, MFA bypass, and cross-application token confusion
+---
+
+# Auth0
+
+Auth0 misconfigurations enable account takeover, cross-tenant data access, and privilege escalation through Rules/Actions, loose application settings, weak API authorization, and token acceptance bugs in consuming applications. Test both the Auth0 tenant configuration and how downstream APIs validate Auth0-issued tokens.
+
+## Attack Surface
+
+**Auth0 Components**
+- Applications: SPA, Regular Web, Native, Machine-to-Machine (M2M)
+- APIs (Resource Servers): identifiers, scopes, RBAC, permissions
+- Connections: database, social, enterprise (SAML/OIDC)
+- Rules (legacy) and Actions (post-login, pre-user-registration, credentials exchange)
+- Organizations (multi-tenant B2B), roles, permissions
+- Universal Login, custom domains, custom database scripts
+
+**Token Types**
+- ID Token (OIDC), Access Token (JWT or opaque), Refresh Token
+- Management API tokens, client credentials tokens (M2M)
+- PAR, PKCE flows for public clients
+
+**Management**
+- Auth0 Management API (`/api/v2/`)
+- Tenant settings, attack protection, MFA policies, anomaly detection
+- Logs streaming, hooks, custom prompts
+
+## Reconnaissance
+
+**Tenant Discovery**
+```
+# From app config, JS bundles, mobile apps
+domain: tenant.us.auth0.com / tenant.eu.auth0.com / login.customdomain.com
+client_id, audience, scope values in authorize URLs
+```
+
+**OIDC Discovery**
+```
+GET https://TENANT.auth0.com/.well-known/openid-configuration
+GET https://TENANT.auth0.com/.well-known/jwks.json
+```
+
+**Authenticated Userinfo** (requires bearer access token — unauthenticated requests return 401)
+```
+GET https://TENANT.auth0.com/userinfo
+Authorization: Bearer <access_token>
+```
+
+**Application Fingerprint**
+- Login redirect to `https://TENANT.auth0.com/authorize?client_id=...`
+- `auth0-js`, `@auth0/auth0-spa-js`, `auth0-react` in frontend bundles
+- API `audience` parameter in token requests
+
+**Management API Exposure**
+- Leaked M2M credentials with `read:users`, `update:users`, `create:users` scopes
+- Management API called from browser (CORS misconfiguration)
+
+## Key Vulnerabilities
+
+### Application Configuration
+
+**Callback URL / Origin Misconfigurations**
+- Wildcard or overly broad Allowed Callback URLs: `https://app.com/*`, `http://localhost:*`
+- Allowed Logout URLs, Web Origins, CORS origins too permissive
+- Native app custom scheme hijacking (`com.app://callback`)
+
+**Token Settings**
+- ID Token used as API access token (audience/scope confusion)
+- Refresh token rotation disabled; overly long TTL
+- Signing algorithm downgrade if RS256 not enforced downstream
+
+### API Authorization (Resource Server)
+
+**Missing Scope/RBAC Enforcement**
+- API accepts any valid access token without required `scope` or `permissions` claim
+- RBAC enabled in Auth0 but API doesn't call `/userinfo` or validate `permissions` array
+- Wrong `audience` accepted — token for App A works on App B's API
+
+**Test:**
+```
+# Token for audience A used against API B
+Authorization: Bearer <token_with_audience_A>
+```
+
+### Rules and Actions Abuse
+
+**Post-Login Rule/Action Injection**
+- Rules that add claims based on unvalidated user metadata:
+  ```javascript
+  user.app_metadata.role = 'admin'  // if user can set app_metadata via signup/API
+  ```
+- `context.authorization` manipulation in Actions
+- Secrets in Rule code exposed to tenant admins or via Management API leak
+
+**Signup / Registration Actions**
+- `pre-user-registration` not blocking disposable emails or role self-assignment
+- Social connection account linking without verified email → account takeover
+
+### Organizations (B2B Multi-Tenancy)
+
+- Missing `org_id` validation in API — user from Org A accesses Org B data
+- Invitation flows accepting attacker email domains
+- Organization membership not re-checked after role change
+
+### MFA Bypass
+
+- MFA not enforced on Management API or high-risk applications
+- Remember-browser cookie bypasses step-up for sensitive actions
+- MFA challenge only on Universal Login but API accepts password-grant tokens without MFA
+- Recovery codes/brute-force on enrollment endpoints
+
+### Account Takeover Vectors
+
+- Password reset link not invalidated after use; predictable reset tokens
+- Email verification not required before sensitive actions
+- Change password without re-auth or MFA
+- Linking attacker's social IdP to victim account (same email, unverified)
+
+### Management API
+
+- M2M app with excessive scopes: `delete:users`, `update:users_app_metadata`
+- Management API token in frontend JavaScript or mobile app
+- Rate limiting absent on `/api/v2/users` enumeration
+
+### Custom Database Scripts
+
+- Custom login script with SQL injection in username lookup
+- `get_user` script returning excessive profile fields
+- Scripts with hardcoded credentials or weak hashing
+
+## Advanced Techniques
+
+**Cross-Application Token Confusion**
+- Same `client_secret` reused across environments (dev/prod)
+- Multiple APIs sharing signing keys without `aud` validation
+
+**Resource Owner Password Grant (if enabled)**
+- Legacy grant enabled — direct username/password to token endpoint, bypassing Universal Login MFA
+
+**Impersonation / Delegation**
+- `act_as` or delegation features misconfigured (legacy features in older tenants)
+
+## Testing Methodology
+
+1. **Extract tenant config** — Domain, client_id, audience, scopes from app
+2. **Callback/origin matrix** — Fuzz Allowed Callback URLs and Web Origins
+3. **Token validation** — Swap audiences, strip scopes, expired tokens, wrong signing keys
+4. **Org boundary** — Two org users accessing each other's org-scoped resources
+5. **MFA policy** — Sensitive actions without step-up; API paths bypassing MFA
+6. **Management API** — Hunt for leaked M2M creds; test scope boundaries
+7. **Rules/Actions** — Trace claim injection from `user_metadata` / `app_metadata`
+
+## Validation
+
+1. Demonstrate account takeover or cross-org access with token/callback/metadata abuse
+2. Show API accepting token without required scope/permission/audience
+3. MFA bypass PoC on protected application flow
+4. Document Auth0 setting (Rule, Application config, API RBAC) root cause
+5. Provide authorize → callback → API request chain with evidence
+
+## False Positives
+
+- Callback URL validation rejects all fuzz attempts consistently
+- API validates `aud`, `iss`, `scope`/`permissions` on every request
+- MFA enforced via Auth0 Action on every login for sensitive apps
+- `app_metadata` writable only by admin via Management API, not user signup
+- Organizations feature correctly binds `org_id` in token and API enforces it
+
+## Impact
+
+- Full account takeover across Auth0-connected applications
+- Cross-tenant data breach in B2B org deployments
+- Privilege escalation via metadata/claim injection in Rules
+- Mass user enumeration/modification via Management API abuse
+
+## Pro Tips
+
+1. Always capture full authorize URL — `audience` and `scope` reveal API targets
+2. Decode access token JWT — check `permissions`, `scope`, `org_id`, `https://.../roles` claims
+3. Test dev/stage tenants separately — often weaker callback rules
+4. Pair with `oauth` and `authentication_jwt` skills for flow/token layer testing
+5. Management API M2M creds in CI logs are high-value — search GitHub, buckets, artifacts
+
+## Summary
+
+Auth0 security spans tenant configuration (callbacks, MFA, Rules) and downstream API token validation (`aud`, `scope`, `permissions`, `org_id`). A perfectly configured Universal Login fails if the API accepts tokens without enforcing Auth0's authorization model.

strix/skills/tooling/agent_browser.md

@@ -365,6 +365,23 @@ agent-browser dialog accept "text"    # accept with prompt input
 agent-browser dialog dismiss          # cancel
 ```
 
+## Readiness & recovery
+
+The first `agent-browser open` in a session launches the headless-Chrome
+daemon; later commands reuse it. Distinguish the two failure modes and react
+differently — do **not** blindly re-run the same failing command in a loop:
+
+- **Daemon / connection failure** (`Failed to connect`, `connection refused`,
+  socket missing, `browser not running`): the daemon isn't up or has died. Run
+  `agent-browser doctor` (add `--fix` if it reports repairable problems), then
+  re-open the page. Retrying the original command unchanged will keep failing.
+- **Malformed command** (`Unknown command`, `Ref not found`, bad flag): fix the
+  command itself — re-snapshot for fresh refs, or correct the syntax.
+
+Invoke `agent-browser` directly through `exec_command`; there is no need to wrap
+it in an extra `sh -c "..."` / `bash -lc "..."` layer, which only adds shell
+quoting and startup-file pitfalls.
+
 ## Diagnosing install issues
 
 If a command fails unexpectedly (`Unknown command`, `Failed to connect`,

strix/skills/tooling/katana.md

@@ -24,7 +24,15 @@ High-signal flags:
 - `-p, -parallelism <n>` concurrent input targets
 - `-rl, -rate-limit <n>` request rate limit
 - `-timeout <seconds>` request timeout
+- `-ct, -crawl-duration <s|m|h|d>` maximum time to crawl the target
 - `-retry <n>` retry count
+- `-mdp, -max-domain-pages <n>` cap pages crawled per domain (default: unlimited)
+- `-fsu, -filter-similar` collapse similar URLs (e.g. /users/123 and /users/456)
+- `-fs, -field-scope <dn|rdn|fqdn|regex>` crawl scope (default `rdn` = root domain + ALL subdomains)
+- `-f, -field <url|path|...>` emit only one field (e.g. `-f url` for a plain URL list)
+- `-or, -omit-raw` omit raw request/response from JSONL output
+- `-ob, -omit-body` omit response body from JSONL output
+- `-mrs, -max-response-size <bytes>` cap per-response bytes read (default 4194304)
 - `-ef, -extension-filter <list>` extension exclusions
 - `-tlsi, -tls-impersonate` experimental JA3/TLS impersonation
 - `-hl, -headless` enable hybrid headless crawling
@@ -37,13 +45,13 @@ High-signal flags:
 - `-silent`, `-j, -jsonl`, `-o <file>` output controls
 
 Agent-safe baseline for automation:
-`mkdir -p crawl && katana -u https://target.tld -d 3 -jc -kf robotstxt -c 10 -p 10 -rl 50 -timeout 10 -retry 1 -ef png,jpg,jpeg,gif,svg,css,woff,woff2,ttf,eot,map -silent -j -o crawl/katana.jsonl`
+`mkdir -p crawl && katana -u https://target.tld -d 3 -ct 10m -mdp 2000 -fsu -jc -kf robotstxt -c 10 -p 10 -rl 50 -timeout 10 -retry 1 -ef png,jpg,jpeg,gif,svg,css,woff,woff2,ttf,eot,map -silent -j -o crawl/katana.jsonl`
 
 Common patterns:
 - Fast crawl baseline:
   `katana -u https://target.tld -d 3 -jc -silent`
-- Deeper JS-aware crawl:
-  `katana -u https://target.tld -d 5 -jc -jsl -kf all -c 10 -p 10 -rl 50 -o katana_urls.txt`
+- Deeper JS-aware crawl (narrowed target; keep it time-bounded):
+  `katana -u https://target.tld -d 5 -ct 15m -jc -jsl -kf all -c 10 -p 10 -rl 50 -o katana_urls.txt`
 - Multi-target run with JSONL output:
   `katana -list urls.txt -d 3 -jc -silent -j -o katana.jsonl`
 - Headless crawl with local Chrome:
@@ -59,6 +67,13 @@ Critical correctness rules:
 - For `-kf`, keep depth at least `-d 3` so known files are fully covered.
 - If writing to a file, ensure parent directory exists before `-o`.
 
+Keeping output small (katana has NO default page cap, so plan for volume):
+- Bound scope and volume: `-fs fqdn` (or `-cs`/`-cos` regex) so the crawl doesn't wander across every subdomain, `-mdp <n>` to cap pages per domain, `-fsu` to collapse near-identical URLs, and `-ct`/`-d` to bound time and depth.
+- Shrink each record: default JSONL is verbose. If you only need endpoints, emit a plain URL list with `-f url` instead of `-j`. If you need JSONL, drop the heavy parts with `-or` (omit raw) and `-ob` (omit body), and lower `-mrs` to cap per-response bytes.
+- Reserve `-jsl` / `-kf all` / higher `-d` for a specific narrowed target — they multiply output fast on large sites.
+- Reduce, then delete: once the crawl finishes, extract just what you need (e.g. `katana ... -f url -o urls.txt` or `sort -u` a URL list, or a short note of interesting paths) and remove the raw crawl file/dir. Don't keep large raw crawls around after you've distilled them.
+- Sanity-check size (`du -sh <out>`); if it's outsized for the scope, tighten `-fs`/`-mdp`/`-fsu`/`-d`/`-ct` and re-run rather than keeping it.
+
 Usage rules:
 - Keep `-d`, `-c`, `-p`, and `-rl` explicit for reproducible runs.
 - Use `-ef` early to reduce static-file noise before fuzzing.

strix/skills/tooling/python.md

@@ -7,9 +7,9 @@ description: Run Python through exec_command in the SDK sandbox. Use the image-b
 
 Use `exec_command` for Python. There is no separate Strix Python executor.
 
-Prefer writing reusable scripts to `/workspace/scratch/<name>.py` and
-running them with `python3 /workspace/scratch/<name>.py`. For short
-one-off transformations, `python3 -c` or a small here-document is fine.
+Prefer writing reusable scripts to a `.py` file and running them with
+`python3 <name>.py`. For short one-off transformations, `python3 -c` or a
+small here-document is fine.
 
 The `shell` parameter on `exec_command` is for swapping POSIX shells
 (`bash`/`zsh`/`sh`), not for picking interpreters. Put the interpreter
@@ -84,17 +84,26 @@ automatically, so it shows up in `list_requests` and you can use
 For iterative exploit work, put code in a file:
 
 ```text
-1. Create or edit `/workspace/scratch/exploit.py` with `apply_patch`.
-2. Run it with `exec_command`: `python3 /workspace/scratch/exploit.py`.
+1. Create or edit a task-unique script (e.g. `poc_<task-id>.py`, so it can't
+   clobber a project file or another agent's script) with `apply_patch`.
+2. Run it with `exec_command`: `python3 poc_<task-id>.py`.
 3. Edit and rerun until the proof-of-concept is reliable.
 ```
 
 ## Installing extra packages
 
-The sandbox's Python lives in `/app/.venv`. To add a one-off dependency
-for an exploit script, use `uv` (already in the image and much faster
-than pip):
+The sandbox's Python lives in `/app/.venv`, and it is the active virtualenv
+(`python3` / `pip` already resolve to it). The following common libraries are
+**pre-installed** — import them directly, no install step needed:
+`requests`, `httpx`, `beautifulsoup4` (`bs4`), `lxml`, `pyjwt` (`jwt`),
+`cryptography`.
+
+To add a one-off dependency for an exploit script, use `uv` (already in the
+image and much faster than pip):
 
 ```bash
 uv pip install --python /app/.venv/bin/python <package>
 ```
+
+Plain `pip install <package>` also works because the venv is active. Install
+before you import, so scripts don't fail with `ModuleNotFoundError`.

strix/skills/vulnerabilities/subdomain_takeover.md

@@ -130,6 +130,13 @@ TLS clues: certificate CN/SAN referencing provider default host instead of the c
 3. Optional: issue a DV certificate (legal scope) and reference CT entry as evidence
 4. Demonstrate impact chains (CSP/script-src trust, OAuth redirect acceptance, cookie Domain scoping)
 
+## Severity
+
+- Score severity based on current claimability plus trusted-origin impact, not just a provider-branded error page
+- When evaluating severity, use `web_search` (if available) for the exact provider/product to confirm whether it now enforces subdomain takeover prevention such as TXT/custom-domain ownership verification or reserved-hostname protections; if search is unavailable, do not treat that absence as evidence that the provider prevents claiming
+- If you have positively confirmed the provider currently prevents third-party claiming and you cannot bypass that control, treat the finding as low severity rather than a confirmed takeover — an unconfirmed provider control is not grounds for downgrading
+- Reserve high/critical severity for cases where you can claim the resource or strongly prove claimability and show meaningful impact such as OAuth redirect abuse, cookie scope abuse, CSP trust, email receipt, or NS delegation control. E.g. Elastic Beanstalk takeovers are still generally legitimate.
+
 ## False Positives
 
 - "Unknown domain" pages that are not claimable due to enforced TXT/ownership checks

strix/skills/vulnerabilities/weak_password_detection.md

@@ -0,0 +1,212 @@
+---
+name: weak-password-detection
+description: Weak password detection, credential stuffing, and brute-force testing using common passwords, system-generated credentials, and tooling like Hydra
+---
+
+# Weak Password Detection / Credential Brute-Force
+
+Weak or default credentials remain one of the most prevalent and high-impact vulnerabilities. This skill covers systematic detection of weak passwords through dictionary attacks, credential stuffing, system-generated password prediction, and brute-force tooling.
+
+## Attack Surface
+
+- Login portals (web, API, mobile, SSH, FTP, Telnet, RDP)
+- Admin panels, dashboards, and management interfaces
+- Default or hardcoded credentials in applications and devices
+- Self-registration flows with weak password policies
+- Password reset flows that generate predictable tokens or passwords
+- API key and token authentication with weak secrets
+
+## Reconnaissance
+
+### Identify Authentication Endpoints
+
+- Standard login forms: `/login`, `/signin`, `/auth`, `/authenticate`, `/api/login`
+- Admin panels: `/admin`, `/administrator`, `/manage`, `/console`, `/cpanel`
+- API auth: `/api/v1/token`, `/oauth/token`, `/api/auth`, `/graphql` (login mutations)
+- Service ports: SSH (22), FTP (21), Telnet (23), SMB (445), RDP (3389), MySQL (3306), PostgreSQL (5432), Redis (6379), MongoDB (27017)
+- Mobile app login endpoints and deep-link auth handlers
+
+### Determine Authentication Mechanism
+
+- Form-based (POST with username/password fields)
+- Basic Authentication (Base64 `Authorization: Basic ...`)
+- Bearer token / JWT (password grant flow)
+- API key in header, query parameter, or body
+- Multi-step authentication (username first, then password)
+- CAPTCHA presence and type (reCAPTCHA, hCaptcha, image-based, math)
+- Rate limiting indicators (429 responses, lockout messages, delays)
+
+### Enumerate Valid Usernames
+
+- Error message differentiation: "Invalid username" vs "Invalid password"
+- Registration page username availability checks
+- Password reset flow: response timing or message leakage
+- Public profiles, API responses, or metadata exposing usernames
+- Common patterns: `admin`, `administrator`, `root`, `user`, `test`, `guest`, `support`, `service`, `api`, `dev`, `ops`
+- Email format derivation from company domain patterns
+
+## Key Vulnerabilities
+
+### Weak Password Policies
+
+- No minimum length or complexity requirements
+- Allowing common passwords: `password`, `123456`, `qwerty`, `admin`, `letmein`
+- Not checking against breached password databases (Have I Been Pwned)
+- Case-insensitive password storage
+- No password history enforcement
+- Excessively short maximum length (indicates plaintext or weak hashing)
+
+### Default and Hardcoded Credentials
+
+- Vendor defaults: `admin/admin`, `admin/password`, `root/root`, `guest/guest`
+- Application frameworks: `django/admin`, `tomcat/tomcat`, `weblogic/weblogic`
+- IoT devices, routers, cameras: manufacturer-specific defaults
+- Database defaults: `postgres/postgres`, `sa/sa`, `root/(empty)`
+- Cloud defaults: AWS instance metadata, Azure default service principals
+- Hardcoded in source code, configuration files, or documentation
+
+### Credential Stuffing
+
+- Users reuse passwords across services
+- Breached credential lists (COMB, Collection #1-5, etc.) enable mass account takeover
+- No multi-factor authentication allows direct access with valid credentials
+- Missing breach detection or forced password rotation after known leaks
+
+### Predictable System-Generated Passwords
+
+- Sequential or pattern-based: `Password1`, `Welcome2025!`, `CompanyName123`
+- Time-based generation: passwords derived from registration timestamp
+- Weak randomness: predictable PRNG seeds in password generators
+- Reset tokens that double as temporary passwords with short expiration
+
+### Brute-Force Vulnerabilities
+
+- No rate limiting on login attempts
+- Absent or ineffective account lockout (client-side only, easily bypassed)
+- IP-based blocking without session/user correlation (rotate IPs via proxy)
+- CAPTCHA bypassable or only triggered after excessive attempts
+- Parallel login attempts not tracked (race conditions on attempt counters)
+- Verbose error messages revealing valid usernames
+
+## Advanced Techniques
+
+### Targeted Password Lists
+
+- Generate custom wordlists from:
+  - Company name, product names, and domain components
+  - Geographic location, industry terms
+  - Season + year patterns: `Summer2025!`, `Winter2026@`
+  - Keyboard walks and leet speak variations
+  - Previously breached passwords for the target domain
+- Cewl: `cewl -d 3 -m 5 -w custom.txt https://target.com` to generate from website content
+
+### Credential Stuffing Workflows
+
+- Use breach databases filtered by target domain or related domains
+- Test email:password pairs where email matches target domain
+- Test username:password pairs with common username derivations
+- Validate successful logins without triggering MFA by checking session endpoints
+
+### Multi-Step Authentication Bypass
+
+- Username enumeration → password brute-force on second step
+- Session fixation between steps: manipulate step identifiers
+- Skip steps via direct URL access to later stages
+- Response manipulation to bypass verification checks
+
+### API and Mobile-Specific
+
+- GraphQL login mutations: batch brute-force via array inputs
+- Mobile APIs often lack rate limiting compared to web frontends
+- JWT password grant flows: brute-force against `/token` endpoint
+- OAuth2 password grant: test `grant_type=password` with weak credentials
+
+### Service-Level Brute-Force
+
+- SSH: `hydra -l admin -P passwords.txt ssh://target.com`
+- FTP: `hydra -L users.txt -P passwords.txt ftp://target.com`
+- RDP: `hydra -l administrator -P passwords.txt rdp://target.com`
+- SMB: `hydra -L users.txt -P passwords.txt smb://target.com`
+- Database: MySQL, PostgreSQL, MongoDB, Redis with weak credentials
+- API endpoints: `ffuf` or custom scripts for HTTP-based brute-force
+
+## Tooling
+
+### Hydra (Primary Tool)
+
+- HTTP POST form brute-force:
+  `hydra -l admin -P /usr/share/wordlists/rockyou.txt target.com http-post-form "/login:username=^USER^&password=^PASS^:Invalid credentials"`
+- Basic Auth:
+  `hydra -L users.txt -P passwords.txt target.com http-get -s 8080 /admin`
+- SSH:
+  `hydra -l root -P passwords.txt -t 4 ssh://target.com`
+- FTP:
+  `hydra -L users.txt -P passwords.txt ftp://target.com`
+- Custom headers and cookies:
+  `hydra ... http-post-form "/api/login:json={\"user\":\"^USER^\",\"pass\":\"^PASS^\"}:F=401"`
+
+### ffuf (HTTP Fuzzing)
+
+- Login brute-force with multiple users and passwords:
+  `ffuf -w users.txt:USER -w passwords.txt:PASS -u https://target.com/login -X POST -d "username=USER&password=PASS" -fr "Invalid"`
+- Filter by response size, status code, or regex to identify successes
+
+### Patator (Versatile Brute-Force)
+
+- `patator http_fuzz url=https://target.com/login method=POST body='username=FILE0&password=FILE1' 0=user.txt 1=pass.txt -x ignore:fgrep='Invalid'`
+
+### Custom Python Scripts
+
+- Use `requests` with threading for high-speed API brute-force
+- Implement jitter and proxy rotation to evade rate limiting
+- Parse CSRF tokens dynamically between requests
+
+### Wordlists
+
+- `/usr/share/wordlists/rockyou.txt` (common passwords)
+- `/usr/share/seclists/Passwords/` (organized by category)
+- `/usr/share/seclists/Passwords/Default-Credentials/` (vendor defaults)
+- Custom lists from Cewl, CeWL, or target-specific scraping
+- Breach compilation subsets filtered by target relevance
+
+## Validation
+
+1. Confirm successful login with captured credentials (session token, cookie, or JWT)
+2. Verify account access level: admin vs user privileges
+3. Check if MFA is enforced post-login or can be bypassed
+4. Test credential reuse across other endpoints or services
+5. Document password policy weaknesses that allowed the breach
+6. Verify if the same credentials work on staging, dev, or related domains
+
+## False Positives
+
+- Honey accounts or honeypot responses designed to mislead attackers
+- Temporary lockouts that resolve quickly (distinguish from permanent bans)
+- Different error messages that don't actually indicate valid username enumeration
+- CAPTCHA or WAF blocking that appears as a failed login
+- Rate limiting that returns 429 instead of 401 (adjust timing)
+
+## Impact
+
+- Complete account takeover for affected users
+- Administrative access leading to full system compromise
+- Lateral movement via reused credentials across services
+- Data exfiltration, privilege escalation, and persistence
+- Reputational damage and compliance violations (GDPR, PCI-DSS)
+
+## Pro Tips
+
+1. Always start with default credentials and vendor-specific lists before broad brute-force
+2. Enumerate usernames first; password brute-force without valid users is inefficient
+3. Use small, targeted wordlists before massive lists like rockyou.txt
+4. Monitor for rate limiting and adapt delays; aggressive brute-force causes IP bans and alerts
+5. Test for password spraying (one password, many users) before targeted brute-force
+6. Check for concurrent session limits; successful logins may kick out legitimate users
+7. GraphQL batching can test multiple credentials in a single request, bypassing per-request limits
+8. Document the password policy and recommend minimum standards (length, complexity, breach checking)
+9. When Hydra is unavailable, use ffuf or custom scripts with equivalent logic
+10. Combine with MFA testing: weak passwords plus missing MFA is a critical finding
+
+## Summary
+
+Weak password detection requires systematic enumeration of authentication surfaces, intelligent wordlist selection, and careful brute-force execution. The highest impact often comes from default credentials, password spraying, and credential stuffing rather than exhaustive brute-force. Always validate findings with confirmed logins and assess the full scope of account compromise.

strix/telemetry/README.md

@@ -2,7 +2,7 @@
 
 To help make Strix better for everyone, we collect anonymized data that helps us understand how to better improve our AI security agent for our users, guide the addition of new features, and fix common errors and bugs. This feedback loop is crucial for improving Strix's capabilities and user experience.
 
-We use [PostHog](https://posthog.com), an open-source analytics platform, for data collection and analysis. Our telemetry implementation is fully transparent - you can review the [source code](https://github.com/usestrix/strix/blob/main/strix/telemetry/posthog.py) to see exactly what we track.
+We use [PostHog](https://posthog.com), an open-source analytics platform, for data collection and analysis, along with [Scarf](https://scarf.sh). Our telemetry implementation is fully transparent - you can review the source code ([posthog.py](https://github.com/usestrix/strix/blob/main/strix/telemetry/posthog.py), [scarf.py](https://github.com/usestrix/strix/blob/main/strix/telemetry/scarf.py)) to see exactly what we track.
 
 ### Telemetry Policy
 
@@ -16,7 +16,8 @@ We collect only very **basic** usage data including:
 **System Context:** OS type, architecture, Strix version\
 **Scan Context:** Scan mode (quick/standard/deep), scan type (whitebox/blackbox)\
 **Model Usage:** Which LLM model is being used (not prompts or responses)\
-**Aggregate Metrics:** Vulnerability counts by severity
+**Feature Usage:** Which built-in skills are loaded\
+**Aggregate Metrics:** Vulnerability counts by severity and weakness category (CWE)
 
 ### What We **Never** Collect
 

strix/telemetry/logging.py

@@ -63,6 +63,18 @@ _HANDLER_TAG = "_strix_scan_handler"
 # ``openai.agents`` is the openai-agents SDK's canonical logger root.
 _TRACKED_ROOTS: tuple[str, ...] = ("strix", "openai.agents")
 
+_STDOUT_QUIET_ROOTS: frozenset[str] = frozenset({"openai.agents"})
+
+
+class _StdoutQuietFilter(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        if record.levelno >= logging.WARNING:
+            return True
+        return not any(
+            record.name == root or record.name.startswith(root + ".")
+            for root in _STDOUT_QUIET_ROOTS
+        )
+
 
 def configure_dependency_logging() -> None:
     """Quiet dependency logging/warnings that obscure Strix scan logs."""
@@ -119,6 +131,7 @@ def setup_scan_logging(run_dir: Path, *, debug: bool | None = None) -> Callable[
     stream_handler.setLevel(logging.DEBUG if debug else logging.ERROR)
     stream_handler.setFormatter(formatter)
     stream_handler.addFilter(context_filter)
+    stream_handler.addFilter(_StdoutQuietFilter())
     setattr(stream_handler, _HANDLER_TAG, True)
 
     tracked_loggers = [logging.getLogger(name) for name in _TRACKED_ROOTS]

strix/telemetry/posthog.py

@@ -26,10 +26,10 @@ def _is_enabled() -> bool:
     return load_settings().telemetry.enabled
 
 
-def _send(event: str, properties: dict[str, Any]) -> None:
+def _send(event: str, properties: dict[str, Any]) -> bool:
     if not _is_enabled():
         logger.debug("posthog disabled; skipping event %s", event)
-        return
+        return False
     try:
         payload = {
             "api_key": _POSTHOG_PUBLIC_API_KEY,
@@ -46,8 +46,10 @@ def _send(event: str, properties: dict[str, Any]) -> None:
             pass
     except Exception:  # noqa: BLE001
         logger.debug("posthog send failed for event %s", event, exc_info=True)
+        return False
     else:
         logger.debug("posthog event sent: %s", event)
+        return True
 
 
 def start(
@@ -71,17 +73,34 @@ def start(
     )
 
 
-def finding(severity: str) -> None:
+def finding(severity: str, cwe: str | None = None, is_cve: bool = False) -> None:
     _send(
         "finding_reported",
         {
             **base_props(),
             "severity": severity.lower(),
+            "cwe": (cwe or "").strip().lower() or "unknown",
+            "is_cve": is_cve,
+        },
+    )
+
+
+def skill_loaded(skill_name: str) -> None:
+    _send(
+        "skill_loaded",
+        {
+            **base_props(),
+            "skill": skill_name,
         },
     )
 
 
 def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
+    if report_state.posthog_scan_ended_sent:
+        return
+    if report_state.scan_ended_exit_reason is None:
+        report_state.scan_ended_exit_reason = exit_reason
+
     vulnerabilities_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0}
     for v in report_state.vulnerability_reports:
         sev = v.get("severity", "info").lower()
@@ -110,11 +129,11 @@ def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
     except (TypeError, ValueError, AttributeError):
         pass
 
-    _send(
+    report_state.posthog_scan_ended_sent = _send(
         "scan_ended",
         {
             **base_props(),
-            "exit_reason": exit_reason,
+            "exit_reason": report_state.scan_ended_exit_reason,
             "duration_seconds": round(duration),
             "vulnerabilities_total": len(report_state.vulnerability_reports),
             **{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()},

strix/telemetry/scarf.py

@@ -28,10 +28,10 @@ def _is_enabled() -> bool:
     return load_settings().telemetry.enabled
 
 
-def _send(event: str, properties: dict[str, Any]) -> None:
+def _send(event: str, properties: dict[str, Any]) -> bool:
     if not _is_enabled():
         logger.debug("scarf disabled; skipping event %s", event)
-        return
+        return False
     try:
         props = dict(properties)
         version = str(props.pop("strix_version", get_version()) or "unknown")
@@ -47,8 +47,10 @@ def _send(event: str, properties: dict[str, Any]) -> None:
             pass
     except Exception:  # noqa: BLE001
         logger.debug("scarf send failed for event %s", event, exc_info=True)
+        return False
     else:
         logger.debug("scarf event sent: %s", event)
+        return True
 
 
 def start(
@@ -73,18 +75,36 @@ def start(
     )
 
 
-def finding(severity: str) -> None:
+def finding(severity: str, cwe: str | None = None, is_cve: bool = False) -> None:
     _send(
         "finding_reported",
         {
             **base_props(),
             "session": SESSION_ID,
             "severity": severity.lower(),
+            "cwe": (cwe or "").strip().lower() or "unknown",
+            "is_cve": is_cve,
+        },
+    )
+
+
+def skill_loaded(skill_name: str) -> None:
+    _send(
+        "skill_loaded",
+        {
+            **base_props(),
+            "session": SESSION_ID,
+            "skill": skill_name,
         },
     )
 
 
 def end(report_state: ReportState, exit_reason: str = "completed") -> None:
+    if report_state.scarf_scan_ended_sent:
+        return
+    if report_state.scan_ended_exit_reason is None:
+        report_state.scan_ended_exit_reason = exit_reason
+
     vulnerabilities_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0}
     for v in report_state.vulnerability_reports:
         sev = v.get("severity", "info").lower()
@@ -115,12 +135,12 @@ def end(report_state: ReportState, exit_reason: str = "completed") -> None:
     except (TypeError, ValueError, AttributeError):
         pass
 
-    _send(
+    report_state.scarf_scan_ended_sent = _send(
         "scan_ended",
         {
             **base_props(),
             "session": SESSION_ID,
-            "exit_reason": exit_reason,
+            "exit_reason": report_state.scan_ended_exit_reason,
             "duration_seconds": round(duration),
             "vulnerabilities_total": len(report_state.vulnerability_reports),
             **{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()},

strix/tools/agents_graph/tools.py

@@ -229,7 +229,8 @@ async def wait_for_message(  # noqa: PLR0911
     Use when you have nothing useful to do until a child/peer responds
     — typically after spawning subagents and you want to wait for
     their completion reports. The agent automatically resumes when any
-    message arrives.
+    message arrives, so pick a ``timeout_seconds`` proportional to the
+    work you're awaiting.
 
     **Critical caveats:**
 
@@ -246,9 +247,19 @@ async def wait_for_message(  # noqa: PLR0911
         reason: One-line note shown in graph snapshots while you're
             waiting (helps a human or sibling agent debug who's stuck
             on what).
-        timeout_seconds: Hard cap (default 600s). On timeout the tool
-            returns and you decide whether to keep working or wait
-            again.
+        timeout_seconds: Max seconds to wait (default 600). This is only
+            a cap — the tool returns the INSTANT a message arrives, so a
+            larger value never makes you wait longer when the reply does
+            come. Right-size it to what you're waiting on: a short wait
+            (e.g. 10-60s) for a quick ack or a small/fast subtask, and a
+            longer one (e.g. ~100-200s) only for genuinely long-running
+            work (deep recon, exploitation, a full sub-scan). The cap only
+            bites when the expected message never arrives — so an oversized
+            timeout on a trivial wait just strands you idle until it
+            elapses. On timeout the tool returns and you decide whether to
+            keep working or wait again. (Applies to autonomous multi-agent
+            runs; in interactive/chat sessions the agent instead parks until
+            a message arrives and this cap is not enforced.)
     """
     inner = _ctx(ctx)
     coordinator = coordinator_from_context(inner)
@@ -481,9 +492,10 @@ async def agent_finish(
     3. Stops this subagent's execution.
 
     **Vulnerability findings must already be filed via
-    ``create_vulnerability_report`` before calling this.** The
-    ``findings`` field here is for narrative summary only — it does
-    not register vulns in the scan report.
+    ``create_vulnerability_report`` (or ``create_dependency_report``
+    for known-CVE dependency/supply-chain findings) before calling
+    this.** The ``findings`` field here is for narrative summary only
+    — it does not register vulns in the scan report.
 
     Write the summary as if the parent has no idea what you were
     doing: what did you test, what did you find/confirm/rule out,
@@ -494,8 +506,9 @@ async def agent_finish(
             and specific (URLs, parameters, payloads that worked).
         findings: Optional bullet list of confirmed observations. For
             credit-bearing vulnerabilities, file
-            ``create_vulnerability_report`` first; this is for
-            narrative.
+            ``create_vulnerability_report`` first (or
+            ``create_dependency_report`` for dependency CVEs); this is
+            for narrative.
         success: Whether the assigned subtask was completed
             successfully. Default ``True``.
         report_to_parent: Whether to deliver the completion report to

strix/tools/finish/tool.py

@@ -96,6 +96,15 @@ async def finish_scan(
     2. Writes the four narrative sections to the scan record.
     3. Marks the scan completed and stops execution.
 
+    **This is a terminal action, not a status probe.** Whatever you pass
+    is persisted VERBATIM as the final, customer-facing report and then
+    execution stops. There is no draft mode and no second chance: never
+    submit placeholder, provisional, or "checking if done" text in any
+    field, and never call ``finish_scan`` to poll whether subagents are
+    done (use ``view_agent_graph`` / ``wait_for_message`` for that).
+    Call it exactly ONCE, only when every field holds genuine, finished
+    assessment prose.
+
     **Pre-flight checklist (mandatory — do not skip):**
 
     1. **Call ``view_agent_graph`` first.** Inspect every entry in the
@@ -108,19 +117,39 @@ async def finish_scan(
        Calling ``finish_scan`` while children are alive orphans their
        work and produces an incomplete report.
     2. All vulnerabilities you found are filed via
-       ``create_vulnerability_report`` (un-reported findings are not
-       tracked and not credited).
+       ``create_vulnerability_report`` — or, for known-CVE dependency
+       findings, ``create_dependency_report`` (un-reported findings are
+       not tracked and not credited). A dependency CVE already filed via
+       ``create_dependency_report`` counts as reported; it does NOT need
+       re-filing here and does NOT block finishing.
     3. Don't double-report — one report per distinct vulnerability.
+    4. **Attack-chaining gate.** Do NOT finish until you have genuinely
+       considered chaining the confirmed findings into higher-impact,
+       end-to-end attack paths and tested every plausibly-related
+       combination. You may rule out combinations you can confidently
+       call unrelated — note why instead of padding chains. Any
+       validated chain must already be filed via
+       ``create_vulnerability_report`` — a demonstrated end-to-end chain
+       is a PoC-backed vulnerability, so it uses that tool even when one
+       link is a dependency CVE (the standalone CVE stays in its own
+       ``create_dependency_report``) — and surfaced prominently in
+       ``executive_summary`` / ``technical_analysis``. Finding no real
+       chain after a serious attempt is acceptable; skipping the
+       chaining reasoning, or ignoring a plausibly-related combination,
+       is not.
 
     **Calling this multiple times overwrites the previous report.**
     Make the single call comprehensive.
 
-    **Customer-facing report rules** (this output is rendered into the
-    final PDF the client sees):
+    **Report output rules** (this content may be rendered into generated
+    reports):
 
     - Never mention internal infrastructure: no local/absolute paths
       (``/workspace/...``), no agent names, no sandbox/orchestrator/
       tooling references, no system prompts, no model-internal errors.
+      Never leak internal identifiers (proxy request IDs, internal
+      vulnerability report IDs, or any system-generated IDs) into any
+      field.
     - Tone: formal, third-person, objective, concise. This is a
       consultant deliverable, not an engineering log.
     - Each section has a specific role:
@@ -140,6 +169,74 @@ async def finish_scan(
           (Immediate / Short-term / Medium-term), each with concrete
           remediation steps. End with retest/validation guidance.
 
+    - **Formatting — use markdown in every field.** These fields may be
+      rendered into generated reports, so structure them clearly: lead
+      each section with a short ``# Heading``, use ``**bold**`` for labels/emphasis,
+      ``inline code`` for identifiers/paths/parameters, bullet or
+      numbered lists for enumerations, and fenced code blocks
+      (```` ```language ````) for any code/payload excerpts. Never emit
+      one flat wall of prose or leave code unformatted.
+    - If **zero** vulnerabilities were found, say so plainly and
+      characterize the posture positively; ``technical_analysis`` should
+      summarize the areas tested and confirm no issues, and
+      ``recommendations`` should focus on general hardening.
+
+    Example (abbreviated — mirror this structure, not the wording)::
+
+        executive_summary:
+            # Executive Summary
+
+            An external assessment of the **Acme Customer Portal**
+            identified multiple weaknesses that could lead to
+            unauthorized access to customer data.
+
+            **Overall risk posture:** Elevated.
+
+            **Key findings**
+            - Confirmed SSRF in a URL-preview feature reaching internal
+              network ranges.
+            - Broken tenant isolation enabling cross-tenant data access.
+
+            **Business impact**
+            - Potential exposure of customer records across tenants.
+
+        methodology:
+            # Methodology
+
+            Conducted per the **OWASP WSTG**.
+
+            **Engagement type:** Gray-box external test.
+            **Scope:** `https://app.acme.example`, `.../api/v1/`.
+
+            **Activities:** recon, authn/session review, authorization
+            and tenant-isolation testing, input/SSRF testing.
+
+        technical_analysis:
+            # Technical Analysis
+
+            **Severity model** reflects exploitability x impact.
+
+            1. **SSRF in URL preview** (Critical) — insufficient
+               destination validation; reaches link-local addresses.
+            2. **Broken tenant isolation** (High) — object identifiers
+               accepted without ownership checks.
+
+            **Systemic themes:** authorization enforced inconsistently;
+            no deny-by-default egress policy.
+
+        recommendations:
+            # Recommendations
+
+            **Immediate**
+            1. Remediate SSRF: enforce a destination allowlist,
+               deny-by-default, re-validate on every redirect hop.
+
+            **Short-term**
+            2. Centralize authorization with deny-by-default middleware.
+
+            **Retest & validation:** re-test immediate items to confirm
+            SSRF and tenant-isolation controls hold.
+
     Args:
         executive_summary: Business-level summary for leadership.
         methodology: Frameworks, scope, and approach.

strix/tools/proxy/caido_api.py

@@ -3,7 +3,9 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
+import logging
 import os
 import time
 import urllib.request
@@ -21,9 +23,14 @@ from caido_sdk_client.types import (
 
 
 if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
     from caido_sdk_client import Client as CaidoClient
 
 
+logger = logging.getLogger(__name__)
+
+
 RequestPart = Literal["request", "response"]
 SortBy = Literal[
     "timestamp",
@@ -42,6 +49,19 @@ _SITEMAP_PAGE_SIZE = 30
 
 _DEFAULT_CAIDO_URL = "http://127.0.0.1:48080"
 _CLIENT_CACHE: dict[str, Client] = {}
+_CLIENT_LOCK = asyncio.Lock()
+
+# Substrings that mean the shared client's transport has died or is being used
+# concurrently — recoverable by rebuilding the client and retrying once.
+_CONNECTION_ERROR_MARKERS = (
+    "transport is already connected",
+    "connector is closed",
+    "server disconnected",
+    "session is closed",
+    "cannot write to closing transport",
+    "connection reset",
+    "connection closed",
+)
 _REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = {
     "timestamp": ("req", "created_at"),
     "host": ("req", "host"),
@@ -81,19 +101,116 @@ def _login_as_guest() -> str:
     return str(payload["data"]["loginAsGuest"]["token"]["accessToken"])
 
 
-async def get_client() -> Client:
-    if client := _CLIENT_CACHE.get("default"):
-        return client
-
+async def _new_client() -> Client:
     token = await asyncio.to_thread(_login_as_guest)
     client = Client(caido_url(), auth=TokenAuthOptions(token=token))
     await client.connect()
-    _CLIENT_CACHE["default"] = client
     return client
 
 
+async def _safe_aclose(client: Client | None) -> None:
+    """Close a (possibly dead) client without letting teardown errors escape."""
+    if client is None:
+        return
+    with contextlib.suppress(Exception):
+        await client.aclose()
+
+
+def _is_connection_error(exc: BaseException) -> bool:
+    message = str(exc).lower()
+    if any(marker in message for marker in _CONNECTION_ERROR_MARKERS):
+        return True
+    cause = exc.__cause__ or exc.__context__
+    return cause is not None and cause is not exc and _is_connection_error(cause)
+
+
+async def get_client() -> Client:
+    """Return the shared Caido client, creating it under a lock if needed.
+
+    The lock prevents two concurrent callers from each building a client and
+    racing ``connect()`` on the same transport ("Transport is already
+    connected").
+    """
+    async with _CLIENT_LOCK:
+        client = _CLIENT_CACHE.get("default")
+        if client is None:
+            client = await _new_client()
+            _CLIENT_CACHE["default"] = client
+        return client
+
+
+async def call_with_client[T](
+    fn: Callable[[Client], Awaitable[T]], *, idempotent: bool = True
+) -> T:
+    """Run ``fn`` against the shared client, serialized and reconnect-safe.
+
+    The Caido GraphQL transport is not safe for concurrent use: two in-flight
+    requests race and raise "Transport is already connected". All proxy calls
+    are therefore serialized through ``_CLIENT_LOCK``. If the cached client's
+    transport has since died ("Connector is closed" / "Server disconnected"),
+    the stale client is closed and rebuilt so subsequent calls stop failing
+    against a dead client.
+
+    ``fn`` is only re-run automatically when ``idempotent`` is true. For
+    mutations (replay, scope create/update/delete) a connection error may
+    arrive *after* Caido applied the change, so we heal the client for future
+    calls but re-raise instead of risking a double-apply.
+    """
+    async with _CLIENT_LOCK:
+        client = _CLIENT_CACHE.get("default")
+        if client is None:
+            client = await _new_client()
+            _CLIENT_CACHE["default"] = client
+        try:
+            return await fn(client)
+        except Exception as exc:
+            if not _is_connection_error(exc):
+                raise
+            new_client = await _new_client()
+            _CLIENT_CACHE["default"] = new_client
+            await _safe_aclose(client)
+            if not idempotent:
+                raise
+            return await fn(new_client)
+
+
+class SharedCaidoClient:
+    """Serialized, reconnect-safe wrapper around one host-side Caido client.
+
+    Every agent in a scan shares a single instance (propagated through the
+    shallow-copied run context). ``call`` serializes access — the SDK transport
+    is not concurrency-safe — and, when the transport dies, rebuilds the client
+    via ``reconnect`` (which preserves the Caido project) and closes the dead
+    one, so a transient Caido restart no longer disables proxy tools for the
+    rest of the scan.
+    """
+
+    def __init__(self, client: Client, reconnect: Callable[[], Awaitable[Client]]) -> None:
+        self._client = client
+        self._reconnect = reconnect
+        self._lock = asyncio.Lock()
+
+    async def call[T](self, fn: Callable[[Client], Awaitable[T]], *, idempotent: bool = True) -> T:
+        async with self._lock:
+            try:
+                return await fn(self._client)
+            except Exception as exc:
+                if not _is_connection_error(exc):
+                    raise
+                dead, self._client = self._client, await self._reconnect()
+                await _safe_aclose(dead)
+                if not idempotent:
+                    raise
+                return await fn(self._client)
+
+    async def aclose(self) -> None:
+        async with self._lock:
+            await _safe_aclose(self._client)
+
+
 async def close_client() -> None:
-    client = _CLIENT_CACHE.pop("default", None)
+    async with _CLIENT_LOCK:
+        client = _CLIENT_CACHE.pop("default", None)
     if client is None:
         return
     await client.aclose()
@@ -385,19 +502,23 @@ async def list_requests(
     sort_order: SortOrder = "desc",
     scope_id: str | None = None,
 ) -> Any:
-    return await list_requests_with_client(
-        await get_client(),
-        httpql_filter=httpql_filter,
-        first=first,
-        after=after,
-        sort_by=sort_by,
-        sort_order=sort_order,
-        scope_id=scope_id,
+    return await call_with_client(
+        lambda client: list_requests_with_client(
+            client,
+            httpql_filter=httpql_filter,
+            first=first,
+            after=after,
+            sort_by=sort_by,
+            sort_order=sort_order,
+            scope_id=scope_id,
+        )
     )
 
 
 async def view_request(request_id: str, *, part: RequestPart = "request") -> Any:
-    return await get_request_with_client(await get_client(), request_id, part=part)
+    return await call_with_client(
+        lambda client: get_request_with_client(client, request_id, part=part)
+    )
 
 
 async def repeat_request(
@@ -406,22 +527,28 @@ async def repeat_request(
     modifications: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     mods = modifications or {}
-    result = await get_request_with_client(await get_client(), request_id, part="request")
-    if result is None or result.request.raw is None:
-        raise ValueError(f"Request {request_id} not found")
 
-    original = result.request
-    raw_str = result.request.raw.decode("utf-8", errors="replace")
-    components = parse_raw_request(raw_str)
-    full_url = full_url_from_components(original, components, mods)
-    modified = apply_modifications(components, mods, full_url)
-    connection, raw = build_raw_request(
-        method=modified["method"],
-        url=modified["url"],
-        headers=modified["headers"],
-        body=modified["body"],
-    )
-    return await replay_send_raw(await get_client(), raw=raw, connection=connection)
+    async def _run(client: CaidoClient) -> dict[str, Any]:
+        result = await get_request_with_client(client, request_id, part="request")
+        if result is None or result.request.raw is None:
+            raise ValueError(f"Request {request_id} not found")
+
+        original = result.request
+        raw_str = result.request.raw.decode("utf-8", errors="replace")
+        components = parse_raw_request(raw_str)
+        full_url = full_url_from_components(original, components, mods)
+        modified = apply_modifications(components, mods, full_url)
+        connection, raw = build_raw_request(
+            method=modified["method"],
+            url=modified["url"],
+            headers=modified["headers"],
+            body=modified["body"],
+        )
+        return await replay_send_raw(client, raw=raw, connection=connection)
+
+    # A replay mutates server state; don't auto-retry if the transport dies
+    # mid-send (the request may already have been sent).
+    return await call_with_client(_run, idempotent=False)
 
 
 async def scope_rules(
@@ -432,7 +559,29 @@ async def scope_rules(
     scope_id: str | None = None,
     scope_name: str | None = None,
 ) -> Any:
-    client = await get_client()
+    async def _run(client: CaidoClient) -> Any:
+        return await _scope_rules_with_client(
+            client,
+            action,
+            allowlist=allowlist,
+            denylist=denylist,
+            scope_id=scope_id,
+            scope_name=scope_name,
+        )
+
+    # get/list are read-only and safe to retry; create/update/delete mutate.
+    return await call_with_client(_run, idempotent=action in {"get", "list"})
+
+
+async def _scope_rules_with_client(
+    client: CaidoClient,
+    action: ScopeAction,
+    *,
+    allowlist: list[str] | None = None,
+    denylist: list[str] | None = None,
+    scope_id: str | None = None,
+    scope_name: str | None = None,
+) -> Any:
     if action == "list":
         result = await scope_list(client)
     elif action == "get":
@@ -651,26 +800,30 @@ async def list_sitemap(
     page: int = 1,
     page_size: int = _SITEMAP_PAGE_SIZE,
 ) -> dict[str, Any]:
-    return await list_sitemap_with_client(
-        await get_client(),
-        scope_id=scope_id,
-        parent_id=parent_id,
-        depth=depth,
-        page=page,
-        page_size=page_size,
+    return await call_with_client(
+        lambda client: list_sitemap_with_client(
+            client,
+            scope_id=scope_id,
+            parent_id=parent_id,
+            depth=depth,
+            page=page,
+            page_size=page_size,
+        )
     )
 
 
 async def view_sitemap_entry(entry_id: str) -> dict[str, Any]:
-    return await view_sitemap_entry_with_client(await get_client(), entry_id)
+    return await call_with_client(lambda client: view_sitemap_entry_with_client(client, entry_id))
 
 
 __all__ = [
     "RequestPart",
     "ScopeAction",
+    "SharedCaidoClient",
     "SitemapDepth",
     "SortBy",
     "SortOrder",
+    "call_with_client",
     "close_client",
     "get_client",
     "list_requests",

strix/tools/proxy/tools.py

@@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, Any, Literal
 from agents import RunContextWrapper, function_tool
 
 from strix.tools.proxy import caido_api
+from strix.tools.proxy.caido_api import SharedCaidoClient
 
 
 logger = logging.getLogger(__name__)
@@ -28,7 +29,7 @@ if TYPE_CHECKING:
         SortOrder,
     )
 else:
-    from strix.tools.proxy.caido_api import (  # noqa: TC001
+    from strix.tools.proxy.caido_api import (
         RequestPart,
         SitemapDepth,
         SortBy,
@@ -39,9 +40,18 @@ else:
 ScopeAction = Literal["get", "list", "create", "update", "delete"]
 
 
-def _ctx_client(ctx: RunContextWrapper) -> Client | None:
+def _ctx_proxy(ctx: RunContextWrapper) -> SharedCaidoClient | None:
+    """Return the scan-wide serialized, reconnect-safe Caido client holder.
+
+    All agents in a scan share one :class:`SharedCaidoClient` whose GraphQL
+    transport is not concurrency-safe (parallel calls raise "Transport is
+    already connected"). ``SharedCaidoClient.call`` serializes access and
+    rebuilds the transport if it dies mid-scan. Returns ``None`` when no holder
+    is present (e.g. standalone tool invocation outside a scan run).
+    """
     inner = ctx.context if isinstance(ctx.context, dict) else {}
-    return inner.get("caido_client")
+    proxy = inner.get("caido_client")
+    return proxy if isinstance(proxy, SharedCaidoClient) else None
 
 
 def _to_tool_json(value: Any) -> Any:
@@ -83,6 +93,39 @@ def _err(name: str, exc: Exception) -> str:
     )
 
 
+_HTTPQL_HINT = (
+    "HTTPQL syntax: quote string values and leave integers unquoted; combine "
+    "terms with AND / OR (there is no NOT). Numeric fields (resp.code, req.port, "
+    "id, roundtrip) use eq/ne/gt/gte/lt/lte; text/byte fields (req.host, req.path, "
+    "req.method, req.raw, resp.raw) use cont/ncont/eq/ne/like/nlike/regex/nregex. "
+    "Example: 'resp.code.gte:200 AND resp.code.lt:300 AND req.host.cont:\"api\"'."
+)
+
+
+def _is_httpql_error(exc: Exception) -> bool:
+    message = str(exc).lower()
+    return "httpql" in message or ("filter" in message and "pars" in message)
+
+
+def _httpql_error(exc: Exception, httpql_filter: str | None) -> str:
+    """Return an actionable error for a rejected HTTPQL filter.
+
+    Preserves Caido's exact parser message and echoes the offending query so
+    the agent can self-correct instead of retrying the same broken filter.
+    """
+    logger.info("list_requests rejected HTTPQL filter %r: %s", httpql_filter, exc)
+    return json.dumps(
+        {
+            "success": False,
+            "error": f"Invalid HTTPQL filter: {exc}",
+            "httpql_filter": httpql_filter,
+            "hint": _HTTPQL_HINT,
+        },
+        ensure_ascii=False,
+        default=str,
+    )
+
+
 @function_tool(timeout=120)
 async def list_requests(
     ctx: RunContextWrapper,
@@ -141,19 +184,21 @@ async def list_requests(
         sort_order: ``asc`` or ``desc``.
         scope_id: Restrict to a Caido scope (managed via ``scope_rules``).
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
 
     try:
-        connection = await caido_api.list_requests_with_client(
-            client,
-            httpql_filter=httpql_filter,
-            first=first,
-            after=after,
-            sort_by=sort_by,
-            sort_order=sort_order,
-            scope_id=scope_id,
+        connection = await proxy.call(
+            lambda client: caido_api.list_requests_with_client(
+                client,
+                httpql_filter=httpql_filter,
+                first=first,
+                after=after,
+                sort_by=sort_by,
+                sort_order=sort_order,
+                scope_id=scope_id,
+            )
         )
 
         entries = []
@@ -207,6 +252,8 @@ async def list_requests(
             default=str,
         )
     except Exception as exc:  # noqa: BLE001
+        if httpql_filter and _is_httpql_error(exc):
+            return _httpql_error(exc, httpql_filter)
         return _err("list_requests", exc)
 
 
@@ -244,12 +291,14 @@ async def view_request(
         page: 1-indexed page number (only when no ``search_pattern``).
         page_size: Lines per page.
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
 
     try:
-        result = await caido_api.get_request_with_client(client, request_id, part=part)
+        result = await proxy.call(
+            lambda client: caido_api.get_request_with_client(client, request_id, part=part)
+        )
         if result is None:
             return json.dumps(
                 {"success": False, "error": f"Request {request_id} not found"},
@@ -359,20 +408,15 @@ async def repeat_request(
             - ``body`` — replace the body string entirely.
             - ``cookies`` — dict of cookies to add/update.
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
     mods = modifications or {}
 
-    try:
+    async def _do(client: Client) -> dict[str, Any] | None:
         result = await caido_api.get_request_with_client(client, request_id, part="request")
         if result is None or result.request.raw is None:
-            return json.dumps(
-                {"success": False, "error": f"Request {request_id} not found"},
-                ensure_ascii=False,
-                default=str,
-            )
-
+            return None
         original = result.request
         raw_str = result.request.raw.decode("utf-8", errors="replace")
         components = caido_api.parse_raw_request(raw_str)
@@ -384,7 +428,18 @@ async def repeat_request(
             headers=modified["headers"],
             body=modified["body"],
         )
-        replay = await caido_api.replay_send_raw(client, raw=raw, connection=connection)
+        return await caido_api.replay_send_raw(client, raw=raw, connection=connection)
+
+    try:
+        # A replay mutates target state, so don't auto-retry on a mid-send
+        # transport failure (the request may already have been sent).
+        replay = await proxy.call(_do, idempotent=False)
+        if replay is None:
+            return json.dumps(
+                {"success": False, "error": f"Request {request_id} not found"},
+                ensure_ascii=False,
+                default=str,
+            )
         return _format_replay_tool_result(replay)
     except Exception as exc:  # noqa: BLE001
         return _err("repeat_request", exc)
@@ -437,16 +492,18 @@ async def list_sitemap(
             (recursive subtree). Only meaningful with ``parent_id``.
         page: 1-indexed page (30 entries per page).
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
     try:
-        payload = await caido_api.list_sitemap_with_client(
-            client,
-            scope_id=scope_id,
-            parent_id=parent_id,
-            depth=depth,
-            page=page,
+        payload = await proxy.call(
+            lambda client: caido_api.list_sitemap_with_client(
+                client,
+                scope_id=scope_id,
+                parent_id=parent_id,
+                depth=depth,
+                page=page,
+            )
         )
         return json.dumps(payload, ensure_ascii=False, default=str)
     except Exception as exc:  # noqa: BLE001
@@ -468,11 +525,13 @@ async def view_sitemap_entry(
     Args:
         entry_id: ID from ``list_sitemap`` (or any nested entry).
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
     try:
-        payload = await caido_api.view_sitemap_entry_with_client(client, entry_id)
+        payload = await proxy.call(
+            lambda client: caido_api.view_sitemap_entry_with_client(client, entry_id)
+        )
         return json.dumps(payload, ensure_ascii=False, default=str)
     except Exception as exc:  # noqa: BLE001
         return _err("view_sitemap_entry", exc)
@@ -524,13 +583,13 @@ async def scope_rules(
         scope_id: Required for ``get`` / ``update`` / ``delete``.
         scope_name: Required for ``create`` / ``update``.
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
 
     try:
         if action == "list":
-            scopes = await caido_api.scope_list(client)
+            scopes = await proxy.call(caido_api.scope_list)
             return json.dumps(
                 {"success": True, "scopes": [_to_tool_json(s) for s in scopes]},
                 ensure_ascii=False,
@@ -543,9 +602,11 @@ async def scope_rules(
                     ensure_ascii=False,
                     default=str,
                 )
-            scope = await caido_api.scope_get(client, scope_id)
+            scope = await proxy.call(lambda client: caido_api.scope_get(client, scope_id))
             return json.dumps(
-                {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
+                {"success": True, "scope": _to_tool_json(scope)},
+                ensure_ascii=False,
+                default=str,
             )
         if action == "create":
             if not scope_name:
@@ -554,11 +615,16 @@ async def scope_rules(
                     ensure_ascii=False,
                     default=str,
                 )
-            scope = await caido_api.scope_create(
-                client, name=scope_name, allowlist=allowlist, denylist=denylist
+            scope = await proxy.call(
+                lambda client: caido_api.scope_create(
+                    client, name=scope_name, allowlist=allowlist, denylist=denylist
+                ),
+                idempotent=False,
             )
             return json.dumps(
-                {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
+                {"success": True, "scope": _to_tool_json(scope)},
+                ensure_ascii=False,
+                default=str,
             )
         if action == "update":
             if not scope_id or not scope_name:
@@ -570,11 +636,16 @@ async def scope_rules(
                     ensure_ascii=False,
                     default=str,
                 )
-            scope = await caido_api.scope_update(
-                client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist
+            scope = await proxy.call(
+                lambda client: caido_api.scope_update(
+                    client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist
+                ),
+                idempotent=False,
             )
             return json.dumps(
-                {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
+                {"success": True, "scope": _to_tool_json(scope)},
+                ensure_ascii=False,
+                default=str,
             )
         if not scope_id:
             return json.dumps(
@@ -582,7 +653,7 @@ async def scope_rules(
                 ensure_ascii=False,
                 default=str,
             )
-        await caido_api.scope_delete(client, scope_id)
+        await proxy.call(lambda client: caido_api.scope_delete(client, scope_id), idempotent=False)
         return json.dumps(
             {
                 "success": True,

strix/tools/reporting/tool.py

@@ -148,8 +148,12 @@ _REQUIRED_FIELDS = {
     "poc_description": "PoC description cannot be empty",
     "poc_script_code": "PoC script/code is REQUIRED - provide the actual exploit/payload",
     "remediation_steps": "Remediation steps cannot be empty",
+    "evidence": "Evidence cannot be empty - provide concrete proof of the finding",
+    "assumptions": "Assumptions cannot be empty - state exploitability prerequisites",
 }
 
+_VALID_FIX_EFFORT = frozenset({"trivial", "low", "medium", "high"})
+
 
 async def _do_create(  # noqa: PLR0912
     *,
@@ -161,12 +165,16 @@ async def _do_create(  # noqa: PLR0912
     poc_description: str,
     poc_script_code: str,
     remediation_steps: str,
+    evidence: str,
+    assumptions: str,
+    fix_effort: str,
     cvss_breakdown: dict[str, str],
     endpoint: str | None,
     method: str | None,
     cve: str | None,
     cwe: str | None,
     code_locations: list[dict[str, Any]] | None,
+    fix_pr_body: str | None = None,
     agent_id: str | None = None,
     agent_name: str | None = None,
 ) -> dict[str, Any]:
@@ -180,11 +188,19 @@ async def _do_create(  # noqa: PLR0912
         "poc_description": poc_description,
         "poc_script_code": poc_script_code,
         "remediation_steps": remediation_steps,
+        "evidence": evidence,
+        "assumptions": assumptions,
     }
     for name, msg in _REQUIRED_FIELDS.items():
         if not str(fields.get(name) or "").strip():
             errors.append(msg)
 
+    fix_effort = (fix_effort or "").strip().lower()
+    if fix_effort not in _VALID_FIX_EFFORT:
+        errors.append(
+            f"Invalid fix_effort: {fix_effort!r}. Must be one of: {sorted(_VALID_FIX_EFFORT)}"
+        )
+
     if not isinstance(cvss_breakdown, dict) or not cvss_breakdown:
         errors.append("cvss_breakdown: must be an object with the 8 CVSS metrics")
         cvss_breakdown = {}
@@ -268,6 +284,9 @@ async def _do_create(  # noqa: PLR0912
             poc_description=poc_description,
             poc_script_code=poc_script_code,
             remediation_steps=remediation_steps,
+            evidence=evidence,
+            assumptions=assumptions,
+            fix_effort=fix_effort,
             cvss=cvss_score,
             cvss_breakdown=cvss_breakdown,
             endpoint=endpoint,
@@ -275,6 +294,7 @@ async def _do_create(  # noqa: PLR0912
             cve=cve,
             cwe=cwe,
             code_locations=parsed_locations,
+            fix_pr_body=fix_pr_body,
             agent_id=agent_id if isinstance(agent_id, str) else None,
             agent_name=agent_name if isinstance(agent_name, str) else None,
         )
@@ -309,12 +329,16 @@ async def create_vulnerability_report(
     poc_description: str,
     poc_script_code: str,
     remediation_steps: str,
+    evidence: str,
+    assumptions: str,
+    fix_effort: str,
     cvss_breakdown: dict[str, str],
     endpoint: str | None = None,
     method: str | None = None,
     cve: str | None = None,
     cwe: str | None = None,
     code_locations: list[dict[str, Any]] | None = None,
+    fix_pr_body: str | None = None,
 ) -> str:
     """File a vulnerability report — one report per fully-verified finding.
 
@@ -327,24 +351,46 @@ async def create_vulnerability_report(
     - Suspicions you haven't confirmed with a PoC.
     - Tracking multiple vulnerabilities at once — one report per vuln.
     - Re-reporting something you (or another agent) already filed.
+    - Known-CVE dependency / supply-chain findings that can't be
+      dynamically PoC'd — a vulnerable dependency version pinned in a
+      lockfile/manifest that matches a published advisory. File those
+      with ``create_dependency_report`` instead, never with this tool.
 
     Automatic LLM-based **deduplication** rejects reports that describe
     the same root cause on the same asset as an existing report. If you
     get a ``duplicate_of`` response, do NOT retry — move on to other
     areas.
 
-    **Customer-facing report rules** (the report is PDF-rendered for
-    delivery):
+    **Report output rules** (this content may be rendered into generated
+    reports):
 
     - No internal/system details: never mention paths like
       ``/workspace``, internal tools, agents, sandboxes, models, system
       prompts, internal errors / stack traces, or tester environment.
+      Never leak internal identifiers (proxy request IDs, internal
+      report IDs) into any field.
     - Tone: formal, objective, third-person, vendor-neutral, concise.
-    - Standard finding structure: Overview → Severity & CVSS →
-      Affected assets → Technical details → PoC (steps + code) →
-      Impact → Remediation → Evidence (in technical_analysis).
+      Avoid internal-guidance headings like "QUICK", "Approach", or
+      "Techniques" that read like an engineering runbook rather than a
+      client deliverable.
+    - **Use markdown in every text field**: ``**bold**`` for emphasis,
+      ``inline code`` for identifiers/values/parameters, and fenced
+      code blocks (```` ```language ````) for any code/payload/HTTP
+      excerpt. Never leave code bare/unformatted. When referencing a
+      file, annotate the fence, e.g.
+      ```` ```python title=app.py startLineNumber=42 endLineNumber=50 ````.
+    - Field discipline: ``poc_description`` is steps only — NO code (all
+      code goes in ``poc_script_code``); ``remediation_steps`` is prose
+      only — NO code/diffs (code fixes go in ``code_locations``).
     - Numbered steps allowed only in PoC and Remediation sections.
     - Avoid hedging language; be precise and non-vague.
+    - Follow a standard pentest report structure across the fields:
+      (1) overview (``description``), (2) severity & CVSS vector
+      (``cvss_breakdown``), (3) affected asset(s) (``target`` /
+      ``endpoint``), (4) technical details (``technical_analysis``),
+      (5) proof of concept (``poc_description`` + ``poc_script_code``),
+      (6) impact (``impact``), (7) evidence (``evidence``), and
+      (8) remediation (``remediation_steps``).
 
     **White-box requirement**: when source is available, you MUST
     populate ``code_locations``. See the ``code_locations`` arg below
@@ -407,13 +453,23 @@ async def create_vulnerability_report(
         title: Specific finding title (e.g.
             ``"SQL Injection in /api/users login parameter"``). Don't
             include the CVE number in the title.
-        description: How the vuln was discovered + what it is.
+        description: Concise, non-technical TL;DR of the vulnerability
+            (1-3 sentences) — it appears first in the report. Deep
+            technical detail and root-cause analysis belong in
+            ``technical_analysis``, not here.
         impact: What an attacker achieves; business risk; data at risk.
         target: Affected URL / domain / repository.
         technical_analysis: The mechanism and root cause.
-        poc_description: Step-by-step reproduction.
+        poc_description: Step-by-step reproduction (steps only, no code).
         poc_script_code: Working PoC (Python preferred).
-        remediation_steps: Specific, actionable fix.
+        remediation_steps: Specific, actionable fix (prose, no code).
+        evidence: Concrete proof the issue is real and exploitable —
+            request/response excerpts, observed behavior, tool output.
+            Use fenced code blocks; no internal identifiers/paths.
+        assumptions: Short note on the assumptions/prerequisites that
+            make this finding impactful or exploitable (e.g. "assumes an
+            authenticated low-privilege user").
+        fix_effort: One of ``trivial`` / ``low`` / ``medium`` / ``high``.
         cvss_breakdown: 8-metric object per the format above.
         endpoint: API path / Git path (e.g. ``/api/login``).
         method: HTTP method when relevant.
@@ -482,6 +538,47 @@ async def create_vulnerability_report(
             - Padding ``fix_before`` with surrounding context lines
               that aren't part of the fix.
             - Duplicating the same change across multiple locations.
+        fix_pr_body: Optional. When source is available and you have a
+            concrete fix, a markdown PR-description body proposing the
+            fix (summary + rationale). Prose/markdown only — the code
+            change itself belongs in ``code_locations``. Omit for
+            black-box findings.
+
+    Example (abbreviated — mirror this structure)::
+
+        title: "Reflected XSS in /search q parameter"
+        description:
+            The **`q`** parameter of `/search` reflects user input into
+            the HTML response without encoding, allowing script
+            injection.
+        technical_analysis:
+            The handler interpolates `q` directly into the page body:
+
+            ```python title=views.py startLineNumber=42 endLineNumber=44
+            html = f"<h2>Results for {q}</h2>"
+            return HttpResponse(html)
+            ```
+
+            No output encoding is applied, so `<script>` executes.
+        poc_description:
+            1. Navigate to `/search?q=<payload>`.
+            2. Observe the payload executes in the victim's browser.
+        poc_script_code:
+            ```
+            GET /search?q=<script>alert(document.domain)</script>
+            ```
+        evidence:
+            Response echoes the payload verbatim:
+
+            ```html
+            <h2>Results for <script>alert(document.domain)</script></h2>
+            ```
+        assumptions:
+            Assumes a victim can be induced to open a crafted link.
+        remediation_steps:
+            Context-encode all user input rendered into HTML; prefer the
+            template engine's auto-escaping over string interpolation.
+        fix_effort: "low"
     """
     inner = ctx.context if isinstance(ctx.context, dict) else {}
     raw_agent_id = inner.get("agent_id")
@@ -503,12 +600,327 @@ async def create_vulnerability_report(
         poc_description=poc_description,
         poc_script_code=poc_script_code,
         remediation_steps=remediation_steps,
+        evidence=evidence,
+        assumptions=assumptions,
+        fix_effort=fix_effort,
         cvss_breakdown=cvss_breakdown,
         endpoint=endpoint,
         method=method,
         cve=cve,
         cwe=cwe,
         code_locations=code_locations,
+        fix_pr_body=fix_pr_body,
+        agent_id=agent_id,
+        agent_name=agent_name,
+    )
+    return json.dumps(result, ensure_ascii=False, default=str)
+
+
+_DEP_SEVERITY_FROM_CVSS = {
+    (9.0, 10.0): "critical",
+    (7.0, 9.0): "high",
+    (4.0, 7.0): "medium",
+    (0.0, 4.0): "low",
+}
+
+
+def _dependency_severity(advisory_cvss: float | None) -> tuple[float, str]:
+    if advisory_cvss is None:
+        return 0.0, "info"
+    score = max(0.0, min(10.0, advisory_cvss))
+    for (lo, hi), label in _DEP_SEVERITY_FROM_CVSS.items():
+        if lo <= score < hi or (hi == 10.0 and score == 10.0):
+            return score, label
+    return score, "none"
+
+
+def _build_dependency_metadata(
+    *,
+    package_name: str,
+    installed_version: str,
+    package_ecosystem: str | None,
+    fixed_version: str | None,
+) -> dict[str, str]:
+    metadata = {
+        "package_name": package_name.strip(),
+        "installed_version": installed_version.strip(),
+    }
+    if package_ecosystem and package_ecosystem.strip():
+        metadata["package_ecosystem"] = package_ecosystem.strip()
+    if fixed_version and fixed_version.strip():
+        metadata["fixed_version"] = fixed_version.strip()
+    return metadata
+
+
+def _build_dependency_evidence(
+    *,
+    cve: str,
+    package_name: str,
+    installed_version: str,
+    fixed_version: str | None,
+) -> str:
+    evidence = (
+        f"**Advisory evidence:** `{cve}` applies to `{package_name}` "
+        f"at installed version `{installed_version}`."
+    )
+    if fixed_version and fixed_version.strip():
+        evidence += f" The advisory is fixed in `{fixed_version.strip()}`."
+    return evidence
+
+
+async def _do_create_dependency(  # noqa: PLR0912
+    *,
+    title: str,
+    description: str,
+    target: str,
+    cve: str,
+    package_name: str,
+    installed_version: str,
+    impact: str,
+    remediation_steps: str,
+    assumptions: str,
+    package_ecosystem: str | None,
+    fixed_version: str | None,
+    cwe: str | None,
+    advisory_cvss: float | None,
+    technical_analysis: str | None,
+    fix_effort: str,
+    agent_id: str | None = None,
+    agent_name: str | None = None,
+) -> dict[str, Any]:
+    errors: list[str] = []
+    required = {
+        "title": title,
+        "description": description,
+        "target": target,
+        "package_name": package_name,
+        "installed_version": installed_version,
+        "package_ecosystem": package_ecosystem,
+        "impact": impact,
+        "remediation_steps": remediation_steps,
+        "assumptions": assumptions,
+    }
+    for name, value in required.items():
+        if not str(value or "").strip():
+            errors.append(f"{name} cannot be empty")
+
+    parsed_cve = _extract_cve(cve or "")
+    cve_err = _validate_cve(parsed_cve)
+    if cve_err:
+        errors.append(cve_err)
+
+    if cwe:
+        cwe = _extract_cwe(cwe)
+        cwe_err = _validate_cwe(cwe)
+        if cwe_err:
+            errors.append(cwe_err)
+
+    fix_effort = (fix_effort or "").strip().lower()
+    if fix_effort not in _VALID_FIX_EFFORT:
+        errors.append(
+            f"Invalid fix_effort: {fix_effort!r}. Must be one of: {sorted(_VALID_FIX_EFFORT)}"
+        )
+
+    if advisory_cvss is None:
+        errors.append(
+            "advisory_cvss is required: read the published advisory base score "
+            "(0.0-10.0) off the advisory (trivy CVSS / NVD / GHSA). Severity is "
+            "derived solely from it — do not omit it or the finding cannot be rated."
+        )
+    elif not 0.0 <= advisory_cvss <= 10.0:
+        errors.append(f"advisory_cvss must be between 0.0 and 10.0, got {advisory_cvss}")
+
+    if errors:
+        return {"success": False, "error": "Validation failed", "errors": errors}
+
+    cvss_score, severity = _dependency_severity(advisory_cvss)
+    dependency_metadata = _build_dependency_metadata(
+        package_name=package_name,
+        installed_version=installed_version,
+        package_ecosystem=package_ecosystem,
+        fixed_version=fixed_version,
+    )
+    evidence = _build_dependency_evidence(
+        cve=parsed_cve,
+        package_name=package_name.strip(),
+        installed_version=installed_version.strip(),
+        fixed_version=fixed_version,
+    )
+
+    try:
+        from strix.report.state import get_global_report_state
+
+        report_state = get_global_report_state()
+        if report_state is None:
+            logger.warning("No global report state; dependency report not persisted")
+            return {
+                "success": True,
+                "message": f"Dependency finding '{title}' created (not persisted)",
+                "warning": "Report could not be persisted - report state unavailable",
+            }
+
+        from strix.report.dedupe import check_duplicate
+
+        existing = report_state.get_existing_vulnerabilities()
+        candidate = {
+            "title": title,
+            "description": description,
+            "target": target,
+            "cve": parsed_cve,
+            "dependency_metadata": dependency_metadata,
+            "technical_analysis": technical_analysis,
+        }
+        dedupe = await check_duplicate(candidate, existing)
+        if dedupe.get("is_duplicate"):
+            duplicate_id = dedupe.get("duplicate_id", "")
+            return {
+                "success": False,
+                "error": (
+                    f"Potential duplicate (id={duplicate_id[:8]}...) — "
+                    "do not re-report the same dependency finding"
+                ),
+                "duplicate_of": duplicate_id,
+                "confidence": dedupe.get("confidence", 0.0),
+                "reason": dedupe.get("reason", ""),
+            }
+
+        report_id = report_state.add_vulnerability_report(
+            title=title,
+            description=description,
+            severity=severity,
+            impact=impact,
+            target=target,
+            technical_analysis=technical_analysis,
+            remediation_steps=remediation_steps,
+            evidence=evidence,
+            assumptions=assumptions,
+            fix_effort=fix_effort,
+            cvss=cvss_score if advisory_cvss is not None else None,
+            cve=parsed_cve,
+            cwe=cwe,
+            finding_class="dependency_cve",
+            dependency_metadata=dependency_metadata,
+            agent_id=agent_id if isinstance(agent_id, str) else None,
+            agent_name=agent_name if isinstance(agent_name, str) else None,
+        )
+    except (ImportError, AttributeError) as e:
+        logger.exception("create_dependency_report persistence failed")
+        return {"success": False, "error": f"Failed to create dependency report: {e!s}"}
+    else:
+        logger.info(
+            "Dependency report created: id=%s cve=%s package=%s severity=%s",
+            report_id,
+            parsed_cve,
+            package_name,
+            severity,
+        )
+        return {
+            "success": True,
+            "message": f"Dependency finding '{title}' created successfully",
+            "report_id": report_id,
+            "severity": severity,
+            "cve": parsed_cve,
+        }
+
+
+@function_tool(timeout=180, strict_mode=False)
+async def create_dependency_report(
+    ctx: RunContextWrapper,
+    title: str,
+    description: str,
+    target: str,
+    cve: str,
+    package_name: str,
+    installed_version: str,
+    advisory_cvss: float,
+    impact: str,
+    remediation_steps: str,
+    assumptions: str,
+    package_ecosystem: str,
+    fixed_version: str | None = None,
+    cwe: str | None = None,
+    technical_analysis: str | None = None,
+    fix_effort: str = "low",
+) -> str:
+    """File a known-CVE dependency (SCA) finding — one report per CVE x package.
+
+    Use this instead of ``create_vulnerability_report`` when the finding
+    is a **known-CVE supply-chain issue**: a vulnerable third-party
+    package/version identified from a lockfile, manifest, or SBOM. Unlike
+    a dynamic finding, you do NOT need to trigger the vulnerability with a
+    live PoC — a verified advisory + the affected installed version is the
+    evidence.
+
+    **When to file**:
+
+    - A dependency is pinned to a version covered by a published CVE.
+    - You have verified the CVE ID and the installed version falls in the
+      affected range (use ``web_search`` if unsure).
+
+    **When NOT to file**:
+
+    - Dynamically-proven vulnerabilities → use
+      ``create_vulnerability_report`` (``finding_class`` dynamic).
+    - Outdated-but-not-vulnerable dependencies with no CVE.
+    - Re-reporting the same CVE/package already filed.
+
+    **Reachability**: do NOT silently downgrade or suppress a finding
+    because the vulnerable code path may be unreachable — instead state
+    reachability as an ``assumptions`` / confidence factor. Report the
+    finding; let the reader weigh exploitability.
+
+    **Formatting**: use markdown in text fields (``**bold**``, ``inline
+    code`` for package/version identifiers, fenced code blocks for
+    manifest excerpts). No internal paths/tooling/agent references.
+
+    Args:
+        title: e.g. ``"CVE-2024-1234 in lodash 4.17.20 (prototype pollution)"``.
+        description: What the CVE is and why the pinned version is affected.
+        target: Affected repository / project / manifest.
+        cve: ``CVE-YYYY-NNNNN`` — required and must be verified.
+        package_name: Affected package name (e.g. ``lodash``).
+        installed_version: The version currently pinned/installed.
+        impact: What the CVE enables; business risk in this context.
+        remediation_steps: How to fix (usually upgrade to a fixed version).
+        assumptions: Exploitability/reachability assumptions & confidence.
+        package_ecosystem: e.g. ``npm`` / ``pypi`` / ``maven`` / ``go``.
+        fixed_version: First non-vulnerable version, if known.
+        cwe: ``CWE-NNN`` (most specific) if certain, else omit.
+        advisory_cvss: **Required.** Published advisory base score
+            (0.0-10.0) — read it off the advisory (trivy CVSS / NVD / GHSA).
+            Severity is derived solely from this score, so it must be the
+            real published value; do not guess or omit it.
+        technical_analysis: Optional deeper mechanism/root-cause detail.
+        fix_effort: One of ``trivial`` / ``low`` / ``medium`` / ``high``
+            (dependency upgrades are usually ``trivial``/``low``).
+    """
+    inner = ctx.context if isinstance(ctx.context, dict) else {}
+    raw_agent_id = inner.get("agent_id")
+    agent_id = raw_agent_id if isinstance(raw_agent_id, str) else None
+    agent_name = None
+    coordinator = inner.get("coordinator")
+    if agent_id is not None and coordinator is not None:
+        names = getattr(coordinator, "names", {})
+        if isinstance(names, dict):
+            raw_agent_name = names.get(agent_id)
+            agent_name = raw_agent_name if isinstance(raw_agent_name, str) else None
+
+    result = await _do_create_dependency(
+        title=title,
+        description=description,
+        target=target,
+        cve=cve,
+        package_name=package_name,
+        installed_version=installed_version,
+        impact=impact,
+        remediation_steps=remediation_steps,
+        assumptions=assumptions,
+        package_ecosystem=package_ecosystem,
+        fixed_version=fixed_version,
+        cwe=cwe,
+        advisory_cvss=advisory_cvss,
+        technical_analysis=technical_analysis,
+        fix_effort=fix_effort,
         agent_id=agent_id,
         agent_name=agent_name,
     )

strix/tools/shell/README.md

@@ -5,6 +5,23 @@ invocation the agent makes (nmap, ffuf, agent-browser, python3, …) goes
 through `exec_command`. `write_stdin` streams input to a still-running
 process started by an earlier `exec_command` (for interactive prompts).
 
+## `write_stdin` requires a TTY-backed process
+
+`exec_command` runs each command in a fresh **non-interactive** shell (plain
+pipes, no TTY) by default. `write_stdin` only works against a process that is
+still running **and** was started with a PTY. The canonical sequence is:
+
+```text
+exec_command(cmd="python3", tty=true)      # start a PTY-backed process
+write_stdin(session_id=<id>, chars="print(1)\n")
+```
+
+Calling `write_stdin` on a command started with the default `tty=false`, or on
+a process that has already exited, fails with
+`stdin is not available for this process. Start the command with 'tty=true' in
+'exec_command' before using 'write_stdin'.` Use `tty=true` for REPLs,
+`ssh`/`nc`/`ftp`, `msfconsole`, or to deliver a Ctrl-C to a long-running job.
+
 - **Implementation:** `agents.sandbox.capabilities.tools.shell_tool.ShellTool`
   (in the upstream `agents` SDK)
 - **Wired in:** `strix/agents/factory.py` — added per-run via the SDK

tests/test_agent_factory_shell.py

@@ -0,0 +1,51 @@
+"""Tests for the shell tool adapters in the agent factory."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, cast
+
+import pytest
+from agents.tool import FunctionTool
+
+from strix.agents import factory
+
+
+def _capturing_exec_tool(captured: dict[str, str]) -> FunctionTool:
+    async def invoke(_ctx: Any, raw_input: str) -> str:
+        captured["raw_input"] = raw_input
+        return "ok"
+
+    return FunctionTool(
+        name="exec_command",
+        description="test tool",
+        params_json_schema={"type": "object", "properties": {}},
+        on_invoke_tool=invoke,
+    )
+
+
+@pytest.mark.asyncio
+async def test_wrap_exec_command_defaults_shell_to_bash() -> None:
+    captured: dict[str, str] = {}
+    wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured))
+
+    result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "source /tmp/env"}))
+
+    assert result == "ok"
+    assert json.loads(captured["raw_input"]) == {
+        "cmd": "source /tmp/env",
+        "shell": "bash",
+    }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("shell", ["/bin/zsh", ""])
+async def test_wrap_exec_command_preserves_explicit_shell(shell: str) -> None:
+    captured: dict[str, str] = {}
+    wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured))
+
+    await wrapped.on_invoke_tool(
+        cast("Any", None), json.dumps({"cmd": "echo test", "shell": shell})
+    )
+
+    assert json.loads(captured["raw_input"])["shell"] == shell

tests/test_agent_tool_registration.py

@@ -0,0 +1,88 @@
+"""Tests for scan-agent tool registration in factory."""
+
+from __future__ import annotations
+
+import pytest
+from agents.tool import FunctionTool
+
+from strix.agents import factory
+
+
+def _tool(name: str) -> FunctionTool:
+    return FunctionTool(
+        name=name,
+        description="test tool",
+        params_json_schema={"type": "object", "properties": {}, "additionalProperties": False},
+        on_invoke_tool=lambda _ctx, _inp: "ok",
+    )
+
+
+@pytest.fixture(autouse=True)
+def _reset_registry() -> object:
+    saved = list(factory._EXTRA_TOOLS)
+    factory._EXTRA_TOOLS.clear()
+    try:
+        yield
+    finally:
+        factory._EXTRA_TOOLS[:] = saved
+
+
+def test_register_agent_tools_is_deduped() -> None:
+    tool = _tool("dup")
+    factory.register_agent_tools(tool)
+    factory.register_agent_tools(tool)
+    assert factory.registered_agent_tools() == (tool,)
+
+
+def test_registered_tools_appear_before_lifecycle_tool() -> None:
+    tool = _tool("extra")
+    factory.register_agent_tools(tool)
+
+    root = factory.build_strix_agent(is_root=True)
+    child = factory.build_strix_agent(is_root=False)
+
+    root_names = [t.name for t in root.tools]
+    child_names = [t.name for t in child.tools]
+
+    assert root_names[-2:] == ["extra", "finish_scan"]
+    assert child_names[-2:] == ["extra", "agent_finish"]
+
+
+def test_per_call_extra_tools_stack_with_registry() -> None:
+    factory.register_agent_tools(_tool("registered"))
+
+    agent = factory.build_strix_agent(is_root=True, extra_tools=[_tool("per_call")])
+    names = [t.name for t in agent.tools]
+
+    assert "registered" in names
+    assert "per_call" in names
+    assert names[-1] == "finish_scan"
+
+
+def test_register_agent_tools_rejects_duplicate_names() -> None:
+    factory.register_agent_tools(_tool("same_name"))
+
+    with pytest.raises(ValueError, match="same_name"):
+        factory.register_agent_tools(_tool("same_name"))
+
+
+def test_per_call_extra_tools_reject_duplicate_registered_names() -> None:
+    factory.register_agent_tools(_tool("same_name"))
+
+    with pytest.raises(ValueError, match="same_name"):
+        factory.build_strix_agent(is_root=True, extra_tools=[_tool("same_name")])
+
+
+def test_instructions_override_is_used_verbatim() -> None:
+    custom = "You are a scan agent. Follow the provided scope."
+
+    agent = factory.build_strix_agent(is_root=True, instructions_override=custom)
+
+    assert agent.instructions == custom
+
+
+def test_no_override_renders_builtin_prompt() -> None:
+    agent = factory.build_strix_agent(is_root=True)
+
+    assert isinstance(agent.instructions, str)
+    assert agent.instructions != ""

tests/test_cli_target_list.py

@@ -0,0 +1,90 @@
+"""Tests for CLI target-list argument parsing."""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, Any
+
+import pytest
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+cli_main: Any = importlib.import_module("strix.interface.main")
+
+
+def _stub_settings(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        cli_main,
+        "load_settings",
+        lambda: SimpleNamespace(runtime=SimpleNamespace(max_local_copy_mb=1024)),
+    )
+
+
+def test_parse_arguments_accepts_target_list_file(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    target_list = tmp_path / "targets.txt"
+    target_list.write_text(
+        "https://test1.com/\n"
+        "\n"
+        "http://test2.com:5789/\n",
+        encoding="utf-8",
+    )
+    _stub_settings(monkeypatch)
+    monkeypatch.setattr(sys, "argv", ["strix", "--target-list", str(target_list), "-n"])
+
+    args = cli_main.parse_arguments()
+
+    assert [target["original"] for target in args.targets_info] == [
+        "https://test1.com/",
+        "http://test2.com:5789/",
+    ]
+    assert [target["type"] for target in args.targets_info] == [
+        "web_application",
+        "web_application",
+    ]
+
+
+def test_parse_arguments_combines_target_and_target_list(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    target_list = tmp_path / "targets.txt"
+    target_list.write_text("http://test2.com:5789/\n", encoding="utf-8")
+    _stub_settings(monkeypatch)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["strix", "-t", "https://test1.com/", "--target-list", str(target_list)],
+    )
+
+    args = cli_main.parse_arguments()
+
+    assert [target["original"] for target in args.targets_info] == [
+        "https://test1.com/",
+        "http://test2.com:5789/",
+    ]
+
+
+def test_parse_arguments_rejects_resume_with_target_list(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    target_list = tmp_path / "targets.txt"
+    target_list.write_text("https://test1.com/\n", encoding="utf-8")
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["strix", "--resume", "old-run", "--target-list", str(target_list)],
+    )
+
+    with pytest.raises(SystemExit):
+        cli_main.parse_arguments()
+
+    assert (
+        "Cannot combine --resume with --target/--target-list/--mount"
+        in capsys.readouterr().err
+    )

tests/test_config_loader.py

@@ -26,6 +26,7 @@ _LLM_ENV_KEYS = [
     "LITELLM_BASE_URL",
     "OLLAMA_API_BASE",
     "STRIX_REASONING_EFFORT",
+    "STRIX_FORCE_REQUIRED_TOOL_CHOICE",
     "LLM_TIMEOUT",
     "PERPLEXITY_API_KEY",
     # RuntimeSettings

tests/test_docker_client_delete.py

@@ -0,0 +1,86 @@
+"""StrixDockerSandboxClient.delete() best-effort teardown.
+
+delete() kills the sandbox container before delegating to the SDK's delete().
+The kill is meant to be best-effort, but the ``contextlib.suppress`` around it
+must cover the case where the docker daemon socket is already gone: then
+``containers.get()`` -> ``inspect_container`` raises requests'
+``ConnectionError``, which is a *sibling* of ``docker.errors.APIError`` under
+``requests.RequestException`` (not a subclass), so an APIError-only suppress
+would let it escape and surface a traceback on every teardown.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from agents.sandbox.sandboxes.docker import DockerSandboxClient
+from docker import errors as docker_errors
+from requests.exceptions import ConnectionError as RequestsConnectionError
+
+from strix.runtime.docker_client import StrixDockerSandboxClient
+
+
+def _client_with_kill_error(exc: Exception) -> StrixDockerSandboxClient:
+    """A StrixDockerSandboxClient whose containers.get(...).kill() raises ``exc``."""
+    client = StrixDockerSandboxClient.__new__(StrixDockerSandboxClient)
+    docker_client = MagicMock()
+    docker_client.containers.get.side_effect = exc
+    client.docker_client = docker_client
+    return client
+
+
+def _session() -> object:
+    # delete() reads session._inner.state.container_id
+    return SimpleNamespace(_inner=SimpleNamespace(state=SimpleNamespace(container_id="abc123")))
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [
+        RequestsConnectionError("Connection aborted", FileNotFoundError(2, "No such file")),
+        docker_errors.NotFound("gone"),
+        docker_errors.APIError("unhappy"),
+    ],
+)
+@pytest.mark.asyncio
+async def test_delete_swallows_best_effort_kill_errors(exc):
+    """A torn-down socket (ConnectionError) or a gone/unhappy container
+    (NotFound/APIError) during the kill must not propagate; delete() still
+    delegates to the SDK's delete()."""
+    client = _client_with_kill_error(exc)
+    session = _session()
+
+    with patch.object(
+        DockerSandboxClient, "delete", new=AsyncMock(return_value=session)
+    ) as super_delete:
+        result = await client.delete(session)
+
+    assert result is session
+    super_delete.assert_awaited_once()  # teardown proceeded despite the kill error
+
+
+@pytest.mark.asyncio
+async def test_delete_does_not_swallow_unrelated_errors():
+    """A programming error (e.g. ValueError) is not part of best-effort kill and
+    must still propagate."""
+    client = _client_with_kill_error(ValueError("boom"))
+    with pytest.raises(ValueError):
+        await client.delete(_session())
+
+
+@pytest.mark.asyncio
+async def test_delete_noop_without_container_id():
+    """No container_id -> no kill attempt, just delegate."""
+    client = StrixDockerSandboxClient.__new__(StrixDockerSandboxClient)
+    client.docker_client = MagicMock()
+    session = SimpleNamespace(_inner=SimpleNamespace(state=SimpleNamespace(container_id=None)))
+
+    with patch.object(
+        DockerSandboxClient, "delete", new=AsyncMock(return_value=session)
+    ) as super_delete:
+        await client.delete(session)
+
+    client.docker_client.containers.get.assert_not_called()
+    super_delete.assert_awaited_once()

tests/test_inputs.py

@@ -7,7 +7,7 @@ from typing import Any
 
 import pytest
 
-from strix.core.inputs import build_root_task, child_initial_input
+from strix.core.inputs import build_root_task, child_initial_input, make_model_settings
 
 
 def _child_kwargs(parent_history: list[Any]) -> dict[str, Any]:
@@ -112,3 +112,46 @@ def test_build_root_task_diff_scope() -> None:
     assert "Scope Constraints:" in task
     assert "3 changed file(s)" in task
     assert "2 deleted file(s)" in task
+
+
+@pytest.mark.parametrize("model_name", ["openai/o3", "gpt-4o"])
+def test_make_model_settings_forces_required_tool_choice_for_openai_models(
+    model_name: str,
+) -> None:
+    settings = make_model_settings(
+        "none",
+        model_name=model_name,
+        force_required_tool_choice=True,
+    )
+
+    assert settings.tool_choice == "required"
+
+
+def test_make_model_settings_skips_required_tool_choice_for_non_openai_models() -> None:
+    settings = make_model_settings(
+        "none",
+        model_name="anthropic/claude-3-7-sonnet-latest",
+        force_required_tool_choice=True,
+    )
+
+    assert settings.tool_choice is None
+
+
+def test_make_model_settings_forces_required_for_routed_openai_model() -> None:
+    settings = make_model_settings(
+        None,
+        model_name="litellm/openai/gpt-4o",
+        force_required_tool_choice=True,
+    )
+
+    assert settings.tool_choice == "required"
+
+
+def test_make_model_settings_forces_required_for_anyllm_routed_openai_model() -> None:
+    settings = make_model_settings(
+        None,
+        model_name="any-llm/openai/gpt-4o",
+        force_required_tool_choice=True,
+    )
+
+    assert settings.tool_choice == "required"

tests/test_local_dir_staging.py

@@ -0,0 +1,108 @@
+"""Tests for symlink-safe LocalDir staging."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from strix.runtime.local_dir_staging import stage_symlink_safe_dir, tree_has_symlink
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _make_repo(tmp_path: Path) -> Path:
+    repo = tmp_path / "repo"
+    (repo / "pkg").mkdir(parents=True)
+    (repo / "pkg" / "mod.py").write_text("x = 1\n")
+    (repo / "README.md").write_text("readme\n")
+    return repo
+
+
+def test_tree_without_symlinks_used_as_is(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+
+    upload_path, staged = stage_symlink_safe_dir(repo)
+
+    assert staged is None
+    assert upload_path == repo.resolve()
+    assert not tree_has_symlink(repo)
+
+
+def test_in_tree_file_symlink_is_dereferenced(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    (repo / "link.py").symlink_to(repo / "pkg" / "mod.py")
+
+    upload_path, staged = stage_symlink_safe_dir(repo)
+
+    assert staged is not None
+    assert upload_path == staged
+    assert not (staged / "link.py").is_symlink()
+    assert (staged / "link.py").read_text() == "x = 1\n"
+    assert (staged / "pkg" / "mod.py").read_text() == "x = 1\n"
+    assert not tree_has_symlink(staged)
+
+
+def test_in_tree_relative_dir_symlink_is_dereferenced(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    (repo / "pkg_alias").symlink_to("pkg")
+
+    _upload, staged = stage_symlink_safe_dir(repo)
+
+    assert staged is not None
+    assert (staged / "pkg_alias" / "mod.py").read_text() == "x = 1\n"
+    assert not tree_has_symlink(staged)
+
+
+def test_out_of_tree_symlink_is_dropped(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    outside = tmp_path / "outside.txt"
+    outside.write_text("secret\n")
+    (repo / "escape.txt").symlink_to(outside)
+    (repo / "abs_escape").symlink_to("/etc")
+
+    _upload, staged = stage_symlink_safe_dir(repo)
+
+    assert staged is not None
+    assert not (staged / "escape.txt").exists()
+    assert not (staged / "abs_escape").exists()
+    assert (staged / "README.md").exists()
+
+
+def test_dangling_symlink_is_dropped(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    (repo / "dangling").symlink_to(repo / "does-not-exist")
+
+    _upload, staged = stage_symlink_safe_dir(repo)
+
+    assert staged is not None
+    assert not (staged / "dangling").exists()
+    assert not (staged / "dangling").is_symlink()
+
+
+def test_cyclic_symlink_terminates(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    (repo / "self").symlink_to(repo)
+    (repo / "pkg" / "up").symlink_to("..")
+
+    _upload, staged = stage_symlink_safe_dir(repo)
+
+    assert staged is not None
+    assert (staged / "README.md").exists()
+    assert not tree_has_symlink(staged)
+
+
+def test_nested_symlinks_inside_linked_dir(tmp_path: Path) -> None:
+    repo = _make_repo(tmp_path)
+    shared = repo / "shared"
+    shared.mkdir()
+    (shared / "conf.json").write_text("{}\n")
+    (shared / "escape").symlink_to("/etc/passwd")
+    (repo / "pkg" / "shared_link").symlink_to(shared)
+
+    _upload, staged = stage_symlink_safe_dir(repo)
+
+    assert staged is not None
+    assert (staged / "pkg" / "shared_link" / "conf.json").read_text() == "{}\n"
+    assert not (staged / "pkg" / "shared_link" / "escape").exists()
+    assert not (staged / "shared" / "escape").exists()

tests/test_local_sources.py

@@ -19,6 +19,7 @@ from strix.interface.utils import (
     dedupe_local_targets,
     directory_size_bytes,
     find_oversized_local_targets,
+    read_target_list_file,
 )
 
 
@@ -157,6 +158,66 @@ def test_build_mount_targets_info_rejects_empty_path(empty: str) -> None:
         build_mount_targets_info([empty])
 
 
+def test_read_target_list_file_strips_blank_lines(tmp_path: Path) -> None:
+    target_list = tmp_path / "targets.txt"
+    target_list.write_text(
+        "\n"
+        " https://test1.com/ \n"
+        "\n"
+        "http://test2.com:5789/\n"
+        "  \n",
+        encoding="utf-8",
+    )
+
+    assert read_target_list_file(str(target_list)) == [
+        "https://test1.com/",
+        "http://test2.com:5789/",
+    ]
+
+
+def test_read_target_list_file_ignores_comment_lines(tmp_path: Path) -> None:
+    target_list = tmp_path / "targets.txt"
+    target_list.write_text(
+        "# production targets\n"
+        "https://test1.com/\n"
+        "  # staging targets\n"
+        "http://test2.com:5789/\n",
+        encoding="utf-8",
+    )
+
+    assert read_target_list_file(str(target_list)) == [
+        "https://test1.com/",
+        "http://test2.com:5789/",
+    ]
+
+
+def test_read_target_list_file_rejects_empty_file(tmp_path: Path) -> None:
+    target_list = tmp_path / "targets.txt"
+    target_list.write_text(" \n# no targets yet\n\n", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="is empty"):
+        read_target_list_file(str(target_list))
+
+
+def test_read_target_list_file_rejects_missing_path(tmp_path: Path) -> None:
+    with pytest.raises(ValueError, match="not an existing file"):
+        read_target_list_file(str(tmp_path / "missing.txt"))
+
+
+def test_read_target_list_file_rejects_non_utf8_file(tmp_path: Path) -> None:
+    target_list = tmp_path / "targets.txt"
+    target_list.write_bytes(b"https://test1.com/\xff\n")
+
+    with pytest.raises(ValueError, match="must be valid UTF-8 text"):
+        read_target_list_file(str(target_list))
+
+
+@pytest.mark.parametrize("empty", ["", "   "])
+def test_read_target_list_file_rejects_empty_path(empty: str) -> None:
+    with pytest.raises(ValueError, match="must not be empty"):
+        read_target_list_file(empty)
+
+
 def test_dedupe_keeps_distinct_targets_in_order() -> None:
     targets = [
         _local_target("/a"),

tests/test_models.py

@@ -0,0 +1,66 @@
+"""Tests for LLM model recommendation helpers."""
+
+from __future__ import annotations
+
+import pytest
+
+from strix.config.models import RECOMMENDED_MODEL_NAMES, is_recommended_or_frontier_model
+
+
+@pytest.mark.parametrize("model_name", RECOMMENDED_MODEL_NAMES)
+def test_recommended_models_are_accepted(model_name: str) -> None:
+    assert is_recommended_or_frontier_model(model_name)
+
+
+def test_recommended_models_are_matched_case_insensitively() -> None:
+    assert is_recommended_or_frontier_model("Vertex_AI/Gemini-3-Pro-Preview")
+
+
+@pytest.mark.parametrize(
+    "model_name",
+    [
+        "gpt-5.5",
+        "litellm/openai/gpt-5.4-pro",
+        "azure_ai/gpt-5.5-pro",
+        "bedrock_mantle/openai.gpt-5.5",
+        "anthropic/claude-opus-4-8",
+        "anthropic.claude-opus-4-8",
+        "anthropic/claude-opus-4-7",
+        "anthropic/claude-fable-5",
+        "anthropic/claude-sonnet-5",
+        "vertex_ai/claude-sonnet-5@default",
+        "vertex_ai/claude-sonnet-4-6@default",
+        "any-llm/anthropic/claude-sonnet-4-6",
+        "vertex_ai/gemini-3.1-pro-preview",
+        "openrouter/google/gemini-3.1-pro-preview",
+        "deepseek/deepseek-v4-pro",
+        "deepseek/deepseek-r1-0528",
+        "deepseek/deepseek-reasoner",
+        "dashscope/qwen3-max-2026-01-23",
+        "qwen3.7-max",
+        "moonshot/kimi-k2.6",
+        "kimi-k2.7-code",
+    ],
+)
+def test_frontier_model_families_are_accepted(model_name: str) -> None:
+    assert is_recommended_or_frontier_model(model_name)
+
+
+@pytest.mark.parametrize(
+    "model_name",
+    [
+        "",
+        "openai/gpt-4.1",
+        "anthropic/claude-3-5-sonnet-latest",
+        "ollama/llama3.1",
+        "deepseek/deepseek-chat",
+        "custom-ollama/gpt-5-mini-local",
+        "custom-provider/claude-opus-4-local",
+        "xai/grok-4.5",
+        "openrouter/x-ai/grok-4",
+        "mistral/mistral-medium-3-5",
+        "mistral/magistral-medium-latest",
+    ],
+)
+def test_non_frontier_models_are_rejected(model_name: str) -> None:
+    assert not is_recommended_or_frontier_model(model_name)

tests/test_optional_deps.py

@@ -0,0 +1,26 @@
+"""Tests for the optional-dependency extras declared in pyproject.toml."""
+
+from __future__ import annotations
+
+import tomllib
+from pathlib import Path
+
+
+PYPROJECT = Path(__file__).resolve().parent.parent / "pyproject.toml"
+
+
+def _optional_dependencies() -> dict[str, list[str]]:
+    data = tomllib.loads(PYPROJECT.read_text(encoding="utf-8"))
+    return data["project"]["optional-dependencies"]
+
+
+def test_vertex_extra_pins_google_auth() -> None:
+    extras = _optional_dependencies()
+    assert "vertex" in extras
+    assert any(req.startswith("google-auth") for req in extras["vertex"])
+
+
+def test_bedrock_extra_pins_boto3() -> None:
+    extras = _optional_dependencies()
+    assert "bedrock" in extras
+    assert any(req.startswith("boto3") for req in extras["bedrock"])

tests/test_provider_hints.py

@@ -0,0 +1,74 @@
+"""Tests for the provider import-error hint helper in interface/main.py."""
+
+from __future__ import annotations
+
+from strix.interface.main import _provider_import_hint
+
+
+VERTEX_MODEL = "vertex_ai/gemini-3-pro-preview"
+BEDROCK_MODEL = "bedrock/anthropic.claude-4-5-sonnet"
+VERTEX_EXTRA_NAME = "vertex"
+BEDROCK_EXTRA_NAME = "bedrock"
+INSTALL_EXTRA_COMMAND_FRAGMENT = 'pipx install "strix-agent['
+WRAPPED_VERTEX_GOOGLE_ERROR = "litellm.APIConnectionError: No module named 'google'"
+WRAPPED_BEDROCK_BOTO3_ERROR = "litellm.APIConnectionError: No module named 'boto3'"
+
+
+def test_bedrock_boto3_hint() -> None:
+    exc = ModuleNotFoundError("No module named 'boto3'")
+    hint = _provider_import_hint(exc, BEDROCK_MODEL)
+    assert hint is not None
+    assert INSTALL_EXTRA_COMMAND_FRAGMENT in hint
+    assert BEDROCK_EXTRA_NAME in hint
+
+
+def test_vertex_google_hint() -> None:
+    exc = ImportError("No module named 'google'")
+    hint = _provider_import_hint(exc, VERTEX_MODEL)
+    assert hint is not None
+    assert INSTALL_EXTRA_COMMAND_FRAGMENT in hint
+    assert VERTEX_EXTRA_NAME in hint
+
+
+def test_vertex_google_hint_for_litellm_wrapped_connection_error() -> None:
+    exc = ConnectionError(WRAPPED_VERTEX_GOOGLE_ERROR)
+    hint = _provider_import_hint(exc, VERTEX_MODEL)
+    assert hint is not None
+    assert INSTALL_EXTRA_COMMAND_FRAGMENT in hint
+    assert VERTEX_EXTRA_NAME in hint
+
+
+def test_bedrock_boto3_hint_for_litellm_wrapped_connection_error() -> None:
+    exc = ConnectionError(WRAPPED_BEDROCK_BOTO3_ERROR)
+    hint = _provider_import_hint(exc, BEDROCK_MODEL)
+    assert hint is not None
+    assert INSTALL_EXTRA_COMMAND_FRAGMENT in hint
+    assert BEDROCK_EXTRA_NAME in hint
+
+
+def test_vertex_google_submodule_hint() -> None:
+    exc = ModuleNotFoundError("No module named 'google.auth'")
+    hint = _provider_import_hint(exc, VERTEX_MODEL)
+    assert hint is not None
+    assert INSTALL_EXTRA_COMMAND_FRAGMENT in hint
+    assert VERTEX_EXTRA_NAME in hint
+
+
+def test_vertex_google_hint_for_deeply_chained_error() -> None:
+    root = ModuleNotFoundError("No module named 'google.auth'")
+    middle = RuntimeError("provider init failed")
+    middle.__cause__ = root
+    exc = ConnectionError("litellm.APIConnectionError: request failed")
+    exc.__cause__ = middle
+    hint = _provider_import_hint(exc, VERTEX_MODEL)
+    assert hint is not None
+    assert VERTEX_EXTRA_NAME in hint
+
+
+def test_non_import_error_returns_none() -> None:
+    assert _provider_import_hint(ConnectionError("boom"), "bedrock/whatever") is None
+
+
+def test_unrelated_provider_returns_none() -> None:
+    exc = ImportError("No module named 'something'")
+    assert _provider_import_hint(exc, "openai/gpt-4") is None

tests/test_proxy_client.py

@@ -0,0 +1,321 @@
+"""Tests for the shared Caido client lifecycle and proxy error handling.
+
+Covers the concurrency/reconnect guarantees of ``caido_api.call_with_client``
+(the sandbox-imported path) and ``caido_api.SharedCaidoClient`` (the host-side
+holder), plus the actionable HTTPQL errors in ``proxy.tools``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import TYPE_CHECKING, Any, cast
+
+import pytest
+
+from strix.tools.proxy import caido_api, tools
+from strix.tools.proxy.caido_api import SharedCaidoClient
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+class _FakeClient:
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.closed = False
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+@pytest.fixture(autouse=True)
+def _clear_cache() -> Iterator[None]:
+    caido_api._CLIENT_CACHE.clear()
+    yield
+    caido_api._CLIENT_CACHE.clear()
+
+
+async def test_call_with_client_reuses_cached_client(monkeypatch: pytest.MonkeyPatch) -> None:
+    cached = _FakeClient("cached")
+    caido_api._CLIENT_CACHE["default"] = cast("Any", cached)
+
+    async def _new() -> Any:
+        raise AssertionError("_new_client must not run when a client is cached")
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    seen: dict[str, Any] = {}
+
+    async def fn(client: Any) -> str:
+        seen["client"] = client
+        return "ok"
+
+    assert await caido_api.call_with_client(fn) == "ok"
+    assert seen["client"] is cached
+
+
+async def test_call_with_client_creates_and_caches_when_empty(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    created = _FakeClient("fresh")
+
+    async def _new() -> Any:
+        return created
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    seen: dict[str, Any] = {}
+
+    async def fn(client: Any) -> str:
+        seen["client"] = client
+        return "ok"
+
+    assert await caido_api.call_with_client(fn) == "ok"
+    assert seen["client"] is created
+    assert caido_api._CLIENT_CACHE["default"] is created
+
+
+async def test_failed_init_does_not_poison_cache(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def _new() -> Any:
+        raise ConnectionRefusedError("caido not up yet")
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    async def fn(_client: Any) -> str:
+        return "unreachable"
+
+    with pytest.raises(ConnectionRefusedError):
+        await caido_api.call_with_client(fn)
+    assert "default" not in caido_api._CLIENT_CACHE
+
+
+async def test_call_with_client_reconnects_and_closes_dead_transport(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    dead = _FakeClient("dead")
+    fresh = _FakeClient("fresh")
+    caido_api._CLIENT_CACHE["default"] = cast("Any", dead)
+
+    new_calls = {"n": 0}
+
+    async def _new() -> Any:
+        new_calls["n"] += 1
+        return fresh
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    attempts: list[Any] = []
+
+    async def fn(client: Any) -> str:
+        attempts.append(client)
+        if len(attempts) == 1:
+            raise RuntimeError("Transport is already connected")
+        return "ok"
+
+    assert await caido_api.call_with_client(fn) == "ok"
+    assert attempts == [dead, fresh]
+    assert new_calls["n"] == 1
+    assert caido_api._CLIENT_CACHE["default"] is fresh
+    assert dead.closed is True  # stale transport is not leaked
+
+
+async def test_call_with_client_non_idempotent_rebuilds_but_reraises(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    dead = _FakeClient("dead")
+    fresh = _FakeClient("fresh")
+    caido_api._CLIENT_CACHE["default"] = cast("Any", dead)
+
+    async def _new() -> Any:
+        return fresh
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    calls = {"n": 0}
+
+    async def fn(_client: Any) -> str:
+        calls["n"] += 1
+        raise RuntimeError("Server disconnected")
+
+    # A mutation must not be auto-retried (it may already have applied), but the
+    # dead client is still healed so later calls succeed.
+    with pytest.raises(RuntimeError, match="Server disconnected"):
+        await caido_api.call_with_client(fn, idempotent=False)
+    assert calls["n"] == 1
+    assert caido_api._CLIENT_CACHE["default"] is fresh
+    assert dead.closed is True
+
+
+async def test_call_with_client_does_not_retry_application_errors(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    cached = _FakeClient("cached")
+    caido_api._CLIENT_CACHE["default"] = cast("Any", cached)
+
+    async def _new() -> Any:
+        raise AssertionError("deterministic errors must not trigger a reconnect")
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    calls = {"n": 0}
+
+    async def fn(_client: Any) -> str:
+        calls["n"] += 1
+        raise ValueError("Invalid HTTPQL filter")
+
+    with pytest.raises(ValueError, match="Invalid HTTPQL"):
+        await caido_api.call_with_client(fn)
+    assert calls["n"] == 1
+    assert caido_api._CLIENT_CACHE["default"] is cached
+
+
+async def test_call_with_client_serializes_concurrent_calls(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    caido_api._CLIENT_CACHE["default"] = cast("Any", _FakeClient("shared"))
+
+    async def _new() -> Any:
+        raise AssertionError("no reconnect expected")
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    state = {"active": 0, "max": 0}
+
+    async def fn(_client: Any) -> str:
+        state["active"] += 1
+        state["max"] = max(state["max"], state["active"])
+        await asyncio.sleep(0.01)
+        state["active"] -= 1
+        return "ok"
+
+    await asyncio.gather(*(caido_api.call_with_client(fn) for _ in range(6)))
+    assert state["max"] == 1
+
+
+async def test_shared_client_reconnects_and_closes_dead_transport() -> None:
+    dead = _FakeClient("dead")
+    fresh = _FakeClient("fresh")
+
+    async def _reconnect() -> Any:
+        return fresh
+
+    holder = SharedCaidoClient(cast("Any", dead), _reconnect)
+
+    attempts: list[Any] = []
+
+    async def fn(client: Any) -> str:
+        attempts.append(client)
+        if len(attempts) == 1:
+            raise RuntimeError("Connector is closed")
+        return "ok"
+
+    assert await holder.call(fn) == "ok"
+    assert attempts == [dead, fresh]
+    assert dead.closed is True
+
+
+async def test_shared_client_non_idempotent_rebuilds_but_reraises() -> None:
+    dead = _FakeClient("dead")
+    fresh = _FakeClient("fresh")
+
+    async def _reconnect() -> Any:
+        return fresh
+
+    holder = SharedCaidoClient(cast("Any", dead), _reconnect)
+
+    calls = {"n": 0}
+
+    async def fn(_client: Any) -> str:
+        calls["n"] += 1
+        raise RuntimeError("Server disconnected")
+
+    with pytest.raises(RuntimeError, match="Server disconnected"):
+        await holder.call(fn, idempotent=False)
+    assert calls["n"] == 1
+    assert dead.closed is True
+    # The healthy client remains for the next call.
+    assert await holder.call(lambda _c: _ok()) == "ok"
+
+
+async def _ok() -> str:
+    return "ok"
+
+
+async def test_shared_client_serializes_concurrent_calls() -> None:
+    async def _reconnect() -> Any:
+        raise AssertionError("no reconnect expected")
+
+    holder = SharedCaidoClient(cast("Any", _FakeClient("shared")), _reconnect)
+
+    state = {"active": 0, "max": 0}
+
+    async def fn(_client: Any) -> str:
+        state["active"] += 1
+        state["max"] = max(state["max"], state["active"])
+        await asyncio.sleep(0.01)
+        state["active"] -= 1
+        return "ok"
+
+    await asyncio.gather(*(holder.call(fn) for _ in range(6)))
+    assert state["max"] == 1
+
+
+async def test_shared_client_passes_through_application_errors() -> None:
+    async def _reconnect() -> Any:
+        raise AssertionError("deterministic errors must not trigger a reconnect")
+
+    holder = SharedCaidoClient(cast("Any", _FakeClient("c")), _reconnect)
+
+    async def fn(_client: Any) -> str:
+        raise ValueError("Invalid HTTPQL filter")
+
+    with pytest.raises(ValueError, match="Invalid HTTPQL"):
+        await holder.call(fn)
+
+
+def test_is_connection_error_matches_markers_and_causes() -> None:
+    assert caido_api._is_connection_error(RuntimeError("Transport is already connected"))
+    assert caido_api._is_connection_error(RuntimeError("Connector is closed"))
+    assert caido_api._is_connection_error(RuntimeError("Server disconnected"))
+    assert not caido_api._is_connection_error(ValueError("Invalid HTTPQL filter"))
+
+    nested = RuntimeError("wrapper")
+    nested.__cause__ = RuntimeError("connection reset by peer")
+    assert caido_api._is_connection_error(nested)
+
+
+class _Ctx:
+    def __init__(self, context: Any) -> None:
+        self.context = context
+
+
+def test_ctx_proxy_returns_holder_when_present() -> None:
+    async def _reconnect() -> Any:
+        raise AssertionError("unused")
+
+    holder = SharedCaidoClient(cast("Any", _FakeClient("c")), _reconnect)
+    got = tools._ctx_proxy(cast("Any", _Ctx({"caido_client": holder})))
+    assert got is holder
+
+
+def test_ctx_proxy_returns_none_without_holder() -> None:
+    assert tools._ctx_proxy(cast("Any", _Ctx({}))) is None
+    assert tools._ctx_proxy(cast("Any", _Ctx(None))) is None
+    assert tools._ctx_proxy(cast("Any", _Ctx({"caido_client": object()}))) is None
+
+
+def test_is_httpql_error_detection() -> None:
+    assert tools._is_httpql_error(RuntimeError("HTTPQL parse error at column 4"))
+    assert tools._is_httpql_error(RuntimeError("failed to parse filter"))
+    assert not tools._is_httpql_error(RuntimeError("Transport is already connected"))
+
+
+def test_httpql_error_preserves_message_and_query() -> None:
+    exc = RuntimeError("HTTPQL parse error: unexpected token at column 12")
+    payload = json.loads(tools._httpql_error(exc, 'resp.code.eq:"200"'))
+    assert payload["success"] is False
+    assert "unexpected token at column 12" in payload["error"]
+    assert payload["httpql_filter"] == 'resp.code.eq:"200"'
+    assert "AND / OR" in payload["hint"]

tests/test_report_writer.py

@@ -79,6 +79,40 @@ def test_render_vulnerability_md_includes_core_sections() -> None:
     assert "**Endpoint:** /api/login" in md
 
 
+def test_render_vulnerability_md_includes_dependency_fields() -> None:
+    md = render_vulnerability_md(
+        _sample_report(
+            title="CVE-2021-23337 in lodash 4.17.20",
+            severity="high",
+            target="repo/package.json",
+            endpoint=None,
+            method=None,
+            cve="CVE-2021-23337",
+            cwe="CWE-94",
+            cvss=7.2,
+            fix_effort="trivial",
+            finding_class="dependency_cve",
+            evidence="**Advisory evidence:** `CVE-2021-23337` applies to `lodash`.",
+            assumptions="Assumes lodash ships in deployed builds.",
+            dependency_metadata={
+                "package_name": "lodash",
+                "package_ecosystem": "npm",
+                "installed_version": "4.17.20",
+                "fixed_version": "4.17.21",
+            },
+            remediation_steps="Upgrade to 4.17.21.",
+        ),
+    )
+    assert "**Package:** lodash" in md
+    assert "**Ecosystem:** npm" in md
+    assert "**Installed Version:** 4.17.20" in md
+    assert "**Fixed Version:** 4.17.21" in md
+    assert "**CWE:** CWE-94" in md
+    assert "**Fix Effort:** Trivial" in md
+    assert "## Evidence" in md
+    assert "## Assumptions" in md
+
+
 def test_write_vulnerabilities_creates_markdown_csv_and_json(tmp_path: Path) -> None:
     reports = [
         _sample_report(id="vuln-0001", severity="medium", timestamp="2026-07-02 11:00:00 UTC"),

tests/test_reporting_fields.py

@@ -0,0 +1,564 @@
+"""Tests for restored report fields, SCA tool, and report formatting guidance."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from strix.report.dedupe import (
+    _check_dependency_duplicate,
+    _prepare_report_for_comparison,
+    check_duplicate,
+)
+from strix.report.state import ReportState, set_global_report_state
+from strix.tools.finish.tool import finish_scan
+from strix.tools.reporting.tool import (
+    _do_create,
+    _do_create_dependency,
+    create_dependency_report,
+    create_vulnerability_report,
+)
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+_CVSS = {
+    "attack_vector": "N",
+    "attack_complexity": "L",
+    "privileges_required": "N",
+    "user_interaction": "N",
+    "scope": "U",
+    "confidentiality": "H",
+    "integrity": "H",
+    "availability": "H",
+}
+
+
+@pytest.fixture
+def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState:
+    monkeypatch.chdir(tmp_path)
+    state = ReportState(run_name="test-run")
+    set_global_report_state(state)
+    return state
+
+
+async def test_create_report_persists_new_fields(report_state: ReportState) -> None:
+    result = await _do_create(
+        title="Reflected XSS in search",
+        description="q reflects unencoded input.",
+        impact="Session theft.",
+        target="https://app.example.com",
+        technical_analysis="Input interpolated into HTML.",
+        poc_description="1. open /search?q=<payload>",
+        poc_script_code="GET /search?q=<script>alert(1)</script>",
+        remediation_steps="Context-encode output.",
+        evidence="Response echoes the payload verbatim.",
+        assumptions="Assumes a victim opens a crafted link.",
+        fix_effort="LOW",
+        cvss_breakdown=_CVSS,
+        endpoint="/search",
+        method="GET",
+        cve=None,
+        cwe="CWE-79",
+        code_locations=None,
+        fix_pr_body="## Fix\nEncode output.",
+    )
+    assert result["success"] is True
+    report = report_state.vulnerability_reports[0]
+    assert report["evidence"] == "Response echoes the payload verbatim."
+    assert report["assumptions"] == "Assumes a victim opens a crafted link."
+    assert report["fix_effort"] == "low"
+    assert report["fix_pr_body"] == "## Fix\nEncode output."
+    assert report["finding_class"] == "dynamic"
+
+
+async def test_create_report_requires_evidence_and_assumptions(
+    report_state: ReportState,
+) -> None:
+    result = await _do_create(
+        title="X",
+        description="d",
+        impact="i",
+        target="t",
+        technical_analysis="ta",
+        poc_description="p",
+        poc_script_code="c",
+        remediation_steps="r",
+        evidence="   ",
+        assumptions="",
+        fix_effort="low",
+        cvss_breakdown=_CVSS,
+        endpoint=None,
+        method=None,
+        cve=None,
+        cwe=None,
+        code_locations=None,
+    )
+    assert result["success"] is False
+    joined = " ".join(result["errors"])
+    assert "Evidence" in joined
+    assert "Assumptions" in joined
+    assert not report_state.vulnerability_reports
+
+
+async def test_create_report_rejects_invalid_fix_effort(report_state: ReportState) -> None:
+    result = await _do_create(
+        title="X",
+        description="d",
+        impact="i",
+        target="t",
+        technical_analysis="ta",
+        poc_description="p",
+        poc_script_code="c",
+        remediation_steps="r",
+        evidence="e",
+        assumptions="a",
+        fix_effort="enormous",
+        cvss_breakdown=_CVSS,
+        endpoint=None,
+        method=None,
+        cve=None,
+        cwe=None,
+        code_locations=None,
+    )
+    assert result["success"] is False
+    assert any("fix_effort" in e for e in result["errors"])
+    assert not report_state.vulnerability_reports
+
+
+async def test_dependency_report_sets_class_and_metadata(report_state: ReportState) -> None:
+    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",
+        fixed_version="4.17.21",
+        cwe="CWE-94",
+        advisory_cvss=7.2,
+        technical_analysis=None,
+        fix_effort="trivial",
+    )
+    assert result["success"] is True
+    report = report_state.vulnerability_reports[0]
+    assert report["finding_class"] == "dependency_cve"
+    assert report["cve"] == "CVE-2021-23337"
+    assert report["severity"] == "high"
+    assert report["evidence"] == (
+        "**Advisory evidence:** `CVE-2021-23337` applies to `lodash` "
+        "at installed version `4.17.20`. The advisory is fixed in `4.17.21`."
+    )
+    assert report["dependency_metadata"] == {
+        "package_name": "lodash",
+        "installed_version": "4.17.20",
+        "package_ecosystem": "npm",
+        "fixed_version": "4.17.21",
+    }
+
+
+async def test_dependency_report_with_zero_cvss_remains_low_severity(
+    report_state: ReportState,
+) -> None:
+    result = await _do_create_dependency(
+        title="CVE-2024-0001 in sample 1.0.0",
+        description="Published advisory affects the pinned version.",
+        target="repo/package.json",
+        cve="CVE-2024-0001",
+        package_name="sample",
+        installed_version="1.0.0",
+        impact="Low-impact dependency advisory.",
+        remediation_steps="Upgrade to 1.0.1.",
+        assumptions="Assumes the package is included in deployed builds.",
+        package_ecosystem="npm",
+        fixed_version="1.0.1",
+        cwe=None,
+        advisory_cvss=0.0,
+        technical_analysis=None,
+        fix_effort="low",
+    )
+
+    assert result["success"] is True
+    assert result["severity"] == "low"
+    report = report_state.vulnerability_reports[0]
+    assert report["severity"] == "low"
+    assert report["cvss"] == 0.0
+
+
+async def test_dependency_report_requires_advisory_cvss(report_state: ReportState) -> None:
+    result = await _do_create_dependency(
+        title="CVE-2024-0001 in sample 1.0.0",
+        description="Published advisory affects the pinned version.",
+        target="repo/package.json",
+        cve="CVE-2024-0001",
+        package_name="sample",
+        installed_version="1.0.0",
+        impact="Some impact.",
+        remediation_steps="Upgrade to 1.0.1.",
+        assumptions="Assumes the package ships in deployed builds.",
+        package_ecosystem="npm",
+        fixed_version="1.0.1",
+        cwe=None,
+        advisory_cvss=None,
+        technical_analysis=None,
+        fix_effort="low",
+    )
+
+    assert result["success"] is False
+    assert any("advisory_cvss is required" in e for e in result["errors"])
+    assert not report_state.vulnerability_reports
+
+
+async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
+    report_state: ReportState,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured: dict[str, object] = {}
+
+    async def fake_check_duplicate(
+        candidate: dict[str, object],
+        existing: list[dict[str, object]],
+    ) -> dict[str, object]:
+        captured["candidate"] = candidate
+        captured["existing"] = existing
+        return {"is_duplicate": False}
+
+    monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate)
+    report_state.vulnerability_reports.append(
+        {
+            "id": "vuln-0001",
+            "title": "CVE-2024-0001 in other 1.0.0",
+            "severity": "low",
+            "timestamp": "2026-01-01 00:00:00 UTC",
+            "description": "Existing dependency finding.",
+            "target": "repo/package.json",
+            "cve": "CVE-2024-0001",
+            "dependency_metadata": {
+                "package_name": "other",
+                "installed_version": "1.0.0",
+                "package_ecosystem": "npm",
+            },
+        }
+    )
+
+    result = await _do_create_dependency(
+        title="CVE-2024-0001 in sample 1.0.0",
+        description="Published advisory affects the pinned version.",
+        target="repo/package.json",
+        cve="CVE-2024-0001",
+        package_name="sample",
+        installed_version="1.0.0",
+        impact="Low-impact dependency advisory.",
+        remediation_steps="Upgrade to 1.0.1.",
+        assumptions="Assumes the package is included in deployed builds.",
+        package_ecosystem="npm",
+        fixed_version="1.0.1",
+        cwe=None,
+        advisory_cvss=0.0,
+        technical_analysis=None,
+        fix_effort="low",
+    )
+
+    assert result["success"] is True
+    assert captured["candidate"] == {
+        "title": "CVE-2024-0001 in sample 1.0.0",
+        "description": "Published advisory affects the pinned version.",
+        "target": "repo/package.json",
+        "cve": "CVE-2024-0001",
+        "dependency_metadata": {
+            "package_name": "sample",
+            "installed_version": "1.0.0",
+            "package_ecosystem": "npm",
+            "fixed_version": "1.0.1",
+        },
+        "technical_analysis": None,
+    }
+
+
+async def test_dependency_report_rejects_bad_cve(report_state: ReportState) -> None:
+    result = await _do_create_dependency(
+        title="bad",
+        description="d",
+        target="t",
+        cve="not-a-cve",
+        package_name="pkg",
+        installed_version="1.0.0",
+        impact="i",
+        remediation_steps="r",
+        assumptions="a",
+        package_ecosystem="npm",
+        fixed_version=None,
+        cwe=None,
+        advisory_cvss=None,
+        technical_analysis=None,
+        fix_effort="low",
+    )
+    assert result["success"] is False
+    assert not report_state.vulnerability_reports
+
+
+async def test_dependency_report_requires_ecosystem(report_state: ReportState) -> None:
+    result = await _do_create_dependency(
+        title="CVE-2024-0001 in sample 1.0.0",
+        description="Published advisory affects the pinned version.",
+        target="repo/package.json",
+        cve="CVE-2024-0001",
+        package_name="sample",
+        installed_version="1.0.0",
+        impact="Low-impact dependency advisory.",
+        remediation_steps="Upgrade to 1.0.1.",
+        assumptions="Assumes the package is included in deployed builds.",
+        package_ecosystem="",
+        fixed_version="1.0.1",
+        cwe=None,
+        advisory_cvss=0.0,
+        technical_analysis=None,
+        fix_effort="low",
+    )
+
+    assert result["success"] is False
+    assert any("package_ecosystem" in error for error in result["errors"])
+    assert not report_state.vulnerability_reports
+
+
+def test_dedupe_comparison_preserves_cve_identity() -> None:
+    cleaned = _prepare_report_for_comparison(
+        {
+            "title": "CVE-2021-23337 in lodash",
+            "description": "Pinned vulnerable dependency.",
+            "target": "repo/package.json",
+            "cve": "CVE-2021-23337",
+            "dependency_metadata": {"package_name": "lodash"},
+        }
+    )
+
+    assert cleaned["cve"] == "CVE-2021-23337"
+    assert cleaned["dependency_metadata"] == {"package_name": "lodash"}
+
+
+async def test_dependency_dedupe_uses_cve_package_identity() -> None:
+    existing = [
+        {
+            "id": "vuln-0001",
+            "title": "CVE-2024-0001 in other",
+            "cve": "CVE-2024-0001",
+            "dependency_metadata": {
+                "package_name": "other",
+                "installed_version": "1.0.0",
+                "package_ecosystem": "npm",
+            },
+        }
+    ]
+    candidate = {
+        "title": "CVE-2024-0001 in sample",
+        "description": "Similar advisory prose.",
+        "target": "repo/package.json",
+        "cve": "CVE-2024-0001",
+        "dependency_metadata": {
+            "package_name": "sample",
+            "installed_version": "1.0.0",
+            "package_ecosystem": "npm",
+        },
+    }
+
+    result = await check_duplicate(candidate, existing)
+
+    assert result["is_duplicate"] is False
+    assert result["confidence"] == 1.0
+
+
+async def test_dependency_dedupe_rejects_same_cve_package_identity() -> None:
+    existing = [
+        {
+            "id": "vuln-0001",
+            "title": "CVE-2024-0001 in sample",
+            "cve": "CVE-2024-0001",
+            "dependency_metadata": {
+                "package_name": "sample",
+                "installed_version": "1.0.0",
+                "package_ecosystem": "npm",
+            },
+        }
+    ]
+    candidate = {
+        "title": "CVE-2024-0001 in sample with different prose",
+        "description": "Different prose for the same dependency identity.",
+        "target": "repo/package.json",
+        "cve": "CVE-2024-0001",
+        "dependency_metadata": {
+            "package_name": "sample",
+            "installed_version": "1.0.1",
+            "package_ecosystem": "npm",
+        },
+    }
+
+    result = await check_duplicate(candidate, existing)
+
+    assert result["is_duplicate"] is True
+    assert result["duplicate_id"] == "vuln-0001"
+    assert result["confidence"] == 1.0
+
+
+async def test_dependency_dedupe_detects_legacy_same_cve_package() -> None:
+    existing = [
+        {
+            "id": "vuln-0001",
+            "title": "CVE-2024-0001 in npm sample package",
+            "description": "Legacy dependency finding without structured metadata.",
+            "cve": "CVE-2024-0001",
+        }
+    ]
+    candidate = {
+        "title": "CVE-2024-0001 in sample",
+        "description": "Different prose for the same dependency identity.",
+        "target": "repo/package.json",
+        "cve": "CVE-2024-0001",
+        "dependency_metadata": {
+            "package_name": "sample",
+            "installed_version": "1.0.1",
+            "package_ecosystem": "npm",
+        },
+    }
+
+    result = await check_duplicate(candidate, existing)
+
+    assert result["is_duplicate"] is True
+    assert result["duplicate_id"] == "vuln-0001"
+    assert result["confidence"] == 1.0
+
+
+def test_dependency_dedupe_defers_unclear_legacy_same_cve() -> None:
+    existing = [
+        {
+            "id": "vuln-0001",
+            "title": "CVE-2024-0001 dependency finding",
+            "description": "Legacy dependency finding without package identity.",
+            "cve": "CVE-2024-0001",
+        }
+    ]
+    candidate = {
+        "title": "CVE-2024-0001 in sample",
+        "description": "Candidate dependency finding.",
+        "target": "repo/package.json",
+        "cve": "CVE-2024-0001",
+        "dependency_metadata": {
+            "package_name": "sample",
+            "installed_version": "1.0.1",
+            "package_ecosystem": "npm",
+        },
+    }
+
+    assert _check_dependency_duplicate(candidate, existing) is None
+
+
+def test_dependency_dedupe_defers_legacy_package_substring_match() -> None:
+    existing = [
+        {
+            "id": "vuln-0001",
+            "title": "CVE-2024-0001 in sample-package",
+            "description": "Legacy dependency finding for a different package.",
+            "cve": "CVE-2024-0001",
+        }
+    ]
+    candidate = {
+        "title": "CVE-2024-0001 in sample",
+        "description": "Candidate dependency finding.",
+        "target": "repo/package.json",
+        "cve": "CVE-2024-0001",
+        "dependency_metadata": {
+            "package_name": "sample",
+            "installed_version": "1.0.1",
+            "package_ecosystem": "npm",
+        },
+    }
+
+    assert _check_dependency_duplicate(candidate, existing) is None
+
+
+def test_dependency_dedupe_defers_legacy_ecosystem_mismatch() -> None:
+    existing = [
+        {
+            "id": "vuln-0001",
+            "title": "CVE-2024-0001 in npm sample",
+            "description": "Legacy dependency finding for a different ecosystem.",
+            "cve": "CVE-2024-0001",
+        }
+    ]
+    candidate = {
+        "title": "CVE-2024-0001 in sample",
+        "description": "Candidate dependency finding.",
+        "target": "repo/requirements.txt",
+        "cve": "CVE-2024-0001",
+        "dependency_metadata": {
+            "package_name": "sample",
+            "installed_version": "1.0.1",
+            "package_ecosystem": "pypi",
+        },
+    }
+
+    assert _check_dependency_duplicate(candidate, existing) is None
+
+
+def test_dependency_dedupe_matches_structured_missing_ecosystem() -> None:
+    existing = [
+        {
+            "id": "vuln-0001",
+            "title": "CVE-2024-0001 in sample",
+            "cve": "CVE-2024-0001",
+            "dependency_metadata": {
+                "package_name": "sample",
+                "installed_version": "1.0.0",
+            },
+        }
+    ]
+    candidate = {
+        "title": "CVE-2024-0001 in sample",
+        "description": "Candidate dependency finding.",
+        "target": "repo/package.json",
+        "cve": "CVE-2024-0001",
+        "dependency_metadata": {
+            "package_name": "sample",
+            "installed_version": "1.0.1",
+            "package_ecosystem": "npm",
+        },
+    }
+
+    result = _check_dependency_duplicate(candidate, existing)
+
+    assert result is not None
+    assert result["is_duplicate"] is True
+    assert result["duplicate_id"] == "vuln-0001"
+
+
+def test_tool_descriptions_include_formatting_guidance() -> None:
+    vuln_desc = create_vulnerability_report.description
+    assert "markdown" in vuln_desc.lower()
+    assert "fenced code" in vuln_desc.lower()
+
+    finish_desc = finish_scan.description
+    assert "markdown" in finish_desc.lower()
+    assert "# Executive Summary" in finish_desc
+
+    dep_desc = create_dependency_report.description
+    assert "cve" in dep_desc.lower()
+    assert "reachab" in dep_desc.lower()
+
+
+def test_vuln_tool_exposes_new_params() -> None:
+    props = create_vulnerability_report.params_json_schema["properties"]
+    for field in ("evidence", "assumptions", "fix_effort", "fix_pr_body"):
+        assert field in props
+
+    dep_props = create_dependency_report.params_json_schema["properties"]
+    for field in ("package_name", "installed_version", "cve", "advisory_cvss"):
+        assert field in dep_props
+    dep_required = create_dependency_report.params_json_schema["required"]
+    assert "package_ecosystem" in dep_required
+    assert "advisory_cvss" in dep_required

tests/test_runner_rate_limit.py

@@ -33,7 +33,12 @@ async def test_persistent_rate_limit_stops_gracefully(
     monkeypatch.setattr(runner, "set_scan_id", lambda _scan_id: None)
 
     settings = types.SimpleNamespace(
-        llm=types.SimpleNamespace(model="openai/gpt-4o", reasoning_effort="high")
+        llm=types.SimpleNamespace(
+            model="openai/gpt-4o",
+            reasoning_effort="high",
+            force_required_tool_choice=False,
+        ),
+        runtime=types.SimpleNamespace(max_context_images=3),
     )
     monkeypatch.setattr(runner, "load_settings", lambda: settings)
     monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
@@ -50,8 +55,8 @@ async def test_persistent_rate_limit_stops_gracefully(
     async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
         return None
 
-    monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)
-    monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)
+    monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)  # type: ignore[attr-defined]
+    monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)  # type: ignore[attr-defined]
 
     monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
     monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: "")

tests/test_runner_root_prompt.py

@@ -0,0 +1,174 @@
+"""Tests for root scan prompt options in run_strix_scan.
+
+Verify that ``root_instructions_override`` and ``extra_system_prompt_context``
+flow through to the root agent's ``build_strix_agent`` call.
+"""
+
+from __future__ import annotations
+
+import types
+from typing import Any
+
+import httpx
+import pytest
+from openai import RateLimitError
+
+import strix.tools.notes.tools as notes_tools
+import strix.tools.todo.tools as todo_tools
+from strix.core import runner
+from strix.core.agents import AgentCoordinator
+
+
+def _make_rate_limit_error() -> RateLimitError:
+    request = httpx.Request("POST", "https://api.openai.com/v1/responses")
+    response = httpx.Response(status_code=429, request=request)
+    return RateLimitError("rate limited", response=response, body=None)
+
+
+def _patch_engine_scaffold(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Any,
+    scope_context: dict[str, Any],
+) -> dict[str, Any]:
+    """Stub out everything around build_strix_agent and stop at run_agent_loop.
+
+    Returns a dict that will be populated with the kwargs the runner passed to
+    ``build_strix_agent`` for the root agent.
+    """
+    monkeypatch.setattr(runner, "run_dir_for", lambda _scan_id: tmp_path)
+    monkeypatch.setattr(runner, "runtime_state_dir", lambda _run_dir: tmp_path)
+    monkeypatch.setattr(runner, "setup_scan_logging", lambda _run_dir: lambda: None)
+    monkeypatch.setattr(runner, "set_scan_id", lambda _scan_id: None)
+
+    settings = types.SimpleNamespace(
+        llm=types.SimpleNamespace(
+            model="openai/gpt-4o",
+            reasoning_effort="high",
+            force_required_tool_choice=False,
+        )
+    )
+    monkeypatch.setattr(runner, "load_settings", lambda: settings)
+    monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
+    monkeypatch.setattr(
+        runner,
+        "uses_chat_completions_tool_schema",
+        lambda _model, _settings: False,
+    )
+
+    monkeypatch.setattr(todo_tools, "hydrate_todos_from_disk", lambda _state_dir: None)
+    monkeypatch.setattr(notes_tools, "hydrate_notes_from_disk", lambda _state_dir: None)
+
+    async def _create_or_reuse(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
+        return {"client": object(), "session": object(), "caido_client": None}
+
+    async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
+        return None
+
+    monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)
+    monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)
+
+    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: object())
+
+    captured: dict[str, Any] = {}
+
+    def _build_strix_agent(**kwargs: Any) -> object:
+        if kwargs.get("is_root") and "kwargs" not in captured:
+            captured["kwargs"] = kwargs
+        return object()
+
+    monkeypatch.setattr(runner, "build_strix_agent", _build_strix_agent)
+    monkeypatch.setattr(runner, "make_child_factory", lambda **_kwargs: lambda **_k: object())
+    monkeypatch.setattr(runner, "open_agent_session", lambda _root_id, _db: object())
+
+    async def _raise_rate_limit(*_args: Any, **_kwargs: Any) -> None:
+        raise _make_rate_limit_error()
+
+    monkeypatch.setattr(runner, "run_agent_loop", _raise_rate_limit)
+    return captured
+
+
+@pytest.mark.asyncio
+async def test_root_prompt_options_flow_into_root_agent(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Any,
+) -> None:
+    scope_context = {
+        "scope_source": "system_scan_config",
+        "authorization_source": "strix_platform_verified_targets",
+        "authorized_targets": [
+            {
+                "type": "web_application",
+                "value": "https://example.com",
+                "workspace_path": "",
+            },
+        ],
+        "user_instructions_do_not_expand_scope": True,
+    }
+    captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
+
+    await runner.run_strix_scan(
+        scan_config={"targets": [], "scan_mode": "deep"},
+        scan_id="scan-ext",
+        image="img",
+        coordinator=AgentCoordinator(),
+        root_instructions_override="CUSTOM SCAN PROMPT",
+        extra_system_prompt_context={"target_context": "known findings"},
+    )
+
+    kwargs = captured["kwargs"]
+    instructions_override = kwargs["instructions_override"]
+    assert "SYSTEM-VERIFIED SCOPE" in instructions_override
+    assert "AUTHORIZED TARGETS" in instructions_override
+    assert "https://example.com" in instructions_override
+    assert "CUSTOM SCAN PROMPT" in instructions_override
+    assert (
+        "cannot expand, replace, or weaken authorized target constraints"
+        in instructions_override
+    )
+    assert kwargs["system_prompt_context"] == {
+        **scope_context,
+        "target_context": "known findings",
+    }
+
+
+@pytest.mark.asyncio
+async def test_extra_system_prompt_context_cannot_override_scope_context(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Any,
+) -> None:
+    scope_context = {"authorized_targets": [{"type": "web_application"}]}
+    captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
+
+    with pytest.raises(ValueError, match="authorized_targets"):
+        await runner.run_strix_scan(
+            scan_config={"targets": [], "scan_mode": "deep"},
+            scan_id="scan-conflict",
+            image="img",
+            coordinator=AgentCoordinator(),
+            extra_system_prompt_context={"authorized_targets": []},
+        )
+
+    assert "kwargs" not in captured
+
+
+@pytest.mark.asyncio
+async def test_root_prompt_options_default_to_none(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Any,
+) -> None:
+    """Without the new args, behavior is unchanged: no override, scope context as-is."""
+    scope_context = {"scope": "built-in"}
+    captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
+
+    await runner.run_strix_scan(
+        scan_config={"targets": [], "scan_mode": "deep"},
+        scan_id="scan-default",
+        image="img",
+        coordinator=AgentCoordinator(),
+    )
+
+    kwargs = captured["kwargs"]
+    assert kwargs["instructions_override"] is None
+    assert kwargs["system_prompt_context"] == {"scope": "built-in"}

tests/test_session_entries.py

@@ -18,15 +18,18 @@ def _source(subdir: str, path: str, *, mount: bool = False) -> dict[str, Any]:
 
 
 def test_copied_source_becomes_localdir_entry(tmp_path: Path) -> None:
-    entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path))])
+    entries, bind_mounts, staged_dirs = build_session_entries([_source("repo", str(tmp_path))])
 
     assert bind_mounts == []
+    assert staged_dirs == []
     assert isinstance(entries["repo"], LocalDir)
     assert entries["repo"].src == tmp_path.resolve()
 
 
 def test_mounted_source_becomes_bind_mount(tmp_path: Path) -> None:
-    entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path), mount=True)])
+    entries, bind_mounts, _staged = build_session_entries(
+        [_source("repo", str(tmp_path), mount=True)]
+    )
 
     assert entries == {}
     assert bind_mounts == [
@@ -44,7 +47,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
     copied.mkdir()
     mounted.mkdir()
 
-    entries, bind_mounts = build_session_entries(
+    entries, bind_mounts, _staged = build_session_entries(
         [
             _source("copied", str(copied)),
             _source("mounted", str(mounted), mount=True),
@@ -57,7 +60,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
 
 
 def test_incomplete_sources_are_skipped() -> None:
-    entries, bind_mounts = build_session_entries(
+    entries, bind_mounts, staged_dirs = build_session_entries(
         [
             {"source_path": "", "workspace_subdir": "x"},
             {"source_path": "/p", "workspace_subdir": ""},
@@ -65,3 +68,20 @@ def test_incomplete_sources_are_skipped() -> None:
     )
     assert entries == {}
     assert bind_mounts == []
+    assert staged_dirs == []
+
+
+def test_symlink_tree_is_staged(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "real.txt").write_text("content")
+    (repo / "link.txt").symlink_to(repo / "real.txt")
+
+    entries, _mounts, staged_dirs = build_session_entries([_source("repo", str(repo))])
+
+    assert len(staged_dirs) == 1
+    entry = entries["repo"]
+    assert isinstance(entry, LocalDir)
+    assert entry.src == staged_dirs[0]
+    assert not (staged_dirs[0] / "link.txt").is_symlink()
+    assert (staged_dirs[0] / "link.txt").read_text() == "content"

tests/test_skill_dir_extension.py

@@ -0,0 +1,120 @@
+from pathlib import Path
+
+import pytest
+
+import strix.skills as skills_mod
+from strix.skills import (
+    get_all_skill_names,
+    get_available_skills,
+    load_skills,
+    register_skill_dir,
+    registered_skill_dirs,
+    skill_search_dirs,
+    validate_requested_skills,
+)
+
+
+@pytest.fixture(autouse=True)
+def _clear_extra_dirs() -> None:
+    original = list(skills_mod._EXTRA_SKILL_DIRS)
+    skills_mod._EXTRA_SKILL_DIRS.clear()
+    try:
+        yield
+    finally:
+        skills_mod._EXTRA_SKILL_DIRS[:] = original
+
+
+def _write_skill(root: Path, category: str, name: str, body: str) -> None:
+    category_dir = root / category
+    category_dir.mkdir(parents=True, exist_ok=True)
+    (category_dir / f"{name}.md").write_text(body, encoding="utf-8")
+
+
+def _write_root_skill(root: Path, name: str, body: str) -> None:
+    root.mkdir(parents=True, exist_ok=True)
+    (root / f"{name}.md").write_text(body, encoding="utf-8")
+
+
+def test_no_registration_leaves_builtin_only() -> None:
+    assert registered_skill_dirs() == ()
+    builtin = skills_mod.get_strix_resource_path("skills")
+    assert skill_search_dirs() == (builtin,)
+    assert {"nmap", "subfinder"}.issubset(get_available_skills()["tooling"])
+
+
+def test_register_is_idempotent_and_ordered(tmp_path: Path) -> None:
+    a = tmp_path / "a"
+    b = tmp_path / "b"
+    a.mkdir()
+    b.mkdir()
+
+    register_skill_dir(a)
+    register_skill_dir(b)
+    register_skill_dir(a)
+
+    # Most recently registered wins → highest precedence first.
+    assert registered_skill_dirs() == (b, a)
+
+
+def test_registered_dir_adds_new_skill(tmp_path: Path) -> None:
+    _write_skill(tmp_path, "extra", "widget", "widget body")
+    register_skill_dir(tmp_path)
+
+    assert "widget" in get_all_skill_names()
+    assert get_available_skills()["extra"] == ["widget"]
+    assert load_skills(["widget"]) == {"widget": "widget body"}
+
+
+def test_registered_root_skill_is_discoverable_and_valid(tmp_path: Path) -> None:
+    _write_root_skill(tmp_path, "widget", "widget body")
+    register_skill_dir(tmp_path)
+
+    assert "widget" in get_all_skill_names()
+    assert get_available_skills()["root"] == ["widget"]
+    assert validate_requested_skills(["widget"]) is None
+    assert validate_requested_skills(["root/widget"]) is None
+    assert load_skills(["widget"]) == {"widget": "widget body"}
+    assert load_skills(["root/widget"]) == {"widget": "widget body"}
+
+
+def test_ambiguous_bare_skill_requires_qualified_name(tmp_path: Path) -> None:
+    _write_skill(tmp_path, "alpha", "widget", "alpha body")
+    _write_skill(tmp_path, "beta", "widget", "beta body")
+    register_skill_dir(tmp_path)
+
+    assert "widget" in get_all_skill_names()
+    assert get_available_skills()["alpha"] == ["widget"]
+    assert get_available_skills()["beta"] == ["widget"]
+    assert validate_requested_skills(["alpha/widget"]) is None
+    assert validate_requested_skills(["beta/widget"]) is None
+
+    error = validate_requested_skills(["widget"])
+    assert error is not None
+    assert "Ambiguous skill name" in error
+    assert "alpha/widget" in error
+    assert "beta/widget" in error
+
+    assert load_skills(["widget"]) == {}
+    assert load_skills(["alpha/widget"]) == {"widget": "alpha body"}
+    assert load_skills(["beta/widget"]) == {"widget": "beta body"}
+
+
+def test_registered_dir_overrides_builtin_skill(tmp_path: Path) -> None:
+    _write_skill(tmp_path, "coordination", "root_agent", "overridden root agent")
+    register_skill_dir(tmp_path)
+
+    loaded = load_skills(["coordination/root_agent"])
+    assert loaded["root_agent"] == "overridden root agent"
+
+
+def test_builtin_skill_still_loads_when_not_overridden(tmp_path: Path) -> None:
+    _write_skill(tmp_path, "extra", "widget", "widget body")
+    register_skill_dir(tmp_path)
+
+    # A packaged skill the registered dir does not shadow still resolves.
+    assert load_skills(["scan_modes/deep"]).get("deep")
+
+
+def test_missing_skill_is_skipped(tmp_path: Path) -> None:
+    register_skill_dir(tmp_path)
+    assert load_skills(["does_not_exist"]) == {}

tests/test_state_repo_context.py

@@ -41,6 +41,15 @@ def test_repository_context_uri_only_without_clone() -> None:
     }
 
 
+def test_repository_context_none_for_multiple_repository_targets() -> None:
+    state = ReportState(run_name="t")
+    state.run_record["targets_info"] = [
+        {"type": "repository", "details": {"target_repo": "https://github.com/acme/widget"}},
+        {"type": "repository", "details": {"target_repo": "https://github.com/acme/api"}},
+    ]
+    assert state._sarif_repository_context() is None
+
+
 def test_repository_context_derives_commit_and_branch_from_clone(tmp_path: Path) -> None:
     repo = tmp_path / "widget"
     repo.mkdir()