Compare commits

..
Author SHA1 Message Date
theo-at-strix 37b8ee1bd4 fix(skills): strengthen DoubleClickjacking validation 2026-11-05 15:30:05 +02:00
theo-at-strix aae7c97461 feat(skills): improve DoubleClickjacking detection and coverage 2026-11-05 02:46:15 +02:00
Samir Hanna Verza 09536cf148 fix(report): validate deduplication decisions 2026-11-04 21:57:34 +02:00
alex s cb581bf93d Feat/str 1646 opengrep (#1625)
* feat(sandbox): replace Semgrep with Opengrep and bundle offline rules

* fix(sandbox): keep opengrep-rules LICENSE and drop proprietary-licensed rules

* feat(sandbox): add Strix rule for GitHub-owned actions not pinned to a commit SHA

* chore: bump sandbox image to 1.4.0 (opengrep)

* fix(sandbox): resolve the gitleaks release tag without the GitHub API

* fix(sandbox): install a pinned, sha256-verified trivy release
2026-11-04 10:26:08 -05:00
Ahmed Allam 6b7ae2ebe4 fix(STR-1654): read manual LLM prices from settings and apply them to STRIX_LLM only 2026-10-29 04:30:48 +03:00
Alex Schapiro 92b30bcad5 fix(STR-1654): price LLM usage from model config 2026-10-29 04:30:48 +03:00
alex s dfdbe4b2ab fix(image): patch brotli decoder (CVE-2026-111497) in sandbox, bump tag 1.3.0 -> 1.3.1 (#1617)
kali-rolling still ships libbrotli 1.2.0-4; node, curl and python3 in the
sandbox all load the shared libbrotlidec, so a malicious Content-Encoding: br
response would hit the vulnerable decoder. Install Debian's fixed 1.2.0-4.1
packages over Kali's and fail the build if the version is not applied.
2026-10-28 19:58:04 -04:00
Ahmed Allam 98e3df1425 feat(llm): send app attribution headers to Vercel AI Gateway 2026-10-28 06:04:41 +03:00
Thibault Jaigu ffa35e22d3 docs(config): describe the current STRIX_API_TYPE default
Since 99c0711 the API route follows the model's catalog entry, not
LLM_API_BASE, so the doc saying a custom base URL defaults to chat
completions was out of date.
2026-10-27 22:24:15 +03:00
Ian ef5d810047 refactor(runner): default replace_scan_prompt to False in _compose_root_instructions_override 2026-10-26 22:56:57 +03:00
ian-at-strix d6f4bc5ced feat(runner): let root_instructions_override replace the root's scan prompt (#1592) 2026-10-26 15:49:51 -04:00
Ahmed Allam 4ac5dc65e4 docs: show Vercel AI Gateway and OpenRouter in the README setup 2026-10-26 19:44:33 +03:00
Ahmed Allam 4bf353fb7b docs: use the vercel_ai_gateway provider for the Vercel AI Gateway setup 2026-10-26 18:58:20 +03:00
Ahmed Allam f121ab03fb docs: update the recommended models in the repo markdown 2026-10-26 18:48:14 +03:00
Ahmed Allam 2c9e9f3444 docs: recommend Vercel AI Gateway alongside OpenRouter and update model ids
Quickstart, providers overview and docs index show GLM-5.3 through Vercel AI Gateway and OpenRouter side by side, Vercel first. Model ids across the provider pages move to the current releases, including DeepSeek V4.1 Flash 1017, and the OpenAI, Anthropic and Gemini pages note the Daybreak, CVP and Fairwind access programs. Docs only, README unchanged.

Closes STR-1686
2026-10-26 18:48:14 +03:00
Ahmed Allam 200a501f15 feat(runtime): pass host devices from STRIX_DOCKER_SANDBOX_DEVICES through to the sandbox container 2026-10-25 01:59:21 +03:00
Ahmed Allam dc40345e26 test(agents): check the diff skill's own coverage lines, not the word 2026-10-24 21:02:27 +03:00
Ahmed Allam fa3539cb75 fix(skills): the diff skill drops its coverage ledger instructions when the coverage tools are off 2026-10-24 21:02:27 +03:00
Ahmed Allam aaf1f8c823 fix(agents): finish_scan drops its coverage reconciliation step when the coverage tools are off 2026-10-24 21:02:27 +03:00
Ian bac386e509 refactor(skills): put the coverage conditionals on their own lines 2026-10-24 21:02:27 +03:00
Ian 898bbb4741 feat(agents): drop the coverage sections from skills and tool descriptions when the tools are off 2026-10-24 21:02:27 +03:00
Ian 5418c767a2 feat(agents): let callers turn off the coverage ledger tools 2026-10-24 21:02:27 +03:00
Ian fff22045ae refactor(prompts): put the parallel-tool-calls conditionals on their own lines 2026-10-24 20:32:27 +03:00
Ian dca9482c82 feat(core): opt-in parallel tool calls via STRIX_PARALLEL_TOOL_CALLS 2026-10-24 20:32:27 +03:00
Ian f46bed6740 refactor(hooks): name the soft turn-limit warning after the soft limit 2026-10-24 20:19:38 +03:00
Ian b8d586a39b fix(agents): an agent with several limited skills gets the largest turn limit 2026-10-24 20:19:38 +03:00
Ian b17c0275a9 feat(agents): soft per-skill turn limits for sub-agents
run_strix_scan(subagent_turn_limits={skill: turns}) gives each sub-agent the
smallest limit among its skills. The agent is warned 10 turns before the limit
and reminded every 5 turns past it, but never stopped.
2026-10-24 20:19:38 +03:00
alex s 14cc948c36 Make the root agent follow up open coverage rows before finishing (#1573) 2026-10-23 10:48:00 -04:00
alex s 594398876f Bump aiohttp, graphql-core, vite and trufflehog past known CVEs (#1569)
- aiohttp 3.14.3 -> 3.14.5 and graphql-core 3.2.11 -> 3.2.13 in uv.lock
- viewer frontend: vite 6.4.3 -> 6.4.4
- sandbox image: trufflehog 3.95.9 -> 3.99.2 (updates its embedded go-git,
  go-billy, amqp091-go and mongo-driver)
2026-10-22 23:28:10 -04:00
theo-at-strixandalex s 0f81f40765 feat(skills): expand XSS coverage for HTML injection and normalization (#1559)
* feat(skills): expand XSS coverage for HTML injection and normalization

* Update xss.md

---------

Co-authored-by: alex s <[email protected]>
2026-10-22 01:09:33 -04:00
alex s a3c31236bd build(deps): enforce a 5-day release cooldown for uv and npm resolution (#1556) 2026-10-21 11:30:08 -04:00
alex s fde37a149f fix(reporting): keep HTTP exchange evidence when the proxy is slow to verify ids (#1551)
Verifying http_exchange_ids fetched every request with its full raw
request and response bodies, serially, and timed out on busy proxy
projects; the ids were then dropped and the finding lost its evidence.

Check existence with an id-only GraphQL query, retry once, and when the
proxy still does not answer keep the ids with a warning instead of
discarding them.
2026-10-20 23:36:38 -04:00
ian-at-strix 3a3855fc4f feat(agents): let callers choose the report tool (#1550)
* feat(agents): let callers remove base tools from scan agents

* feat(agents): let callers choose the report tool instead of removing tools

* refactor(agents): name the report tool via the prompt template, not string replacement
2026-10-20 20:21:31 -04:00
Ian aa2036f229 fix(llm): keep request_log.install(); record tool call timings; both only with STRIX_RUN_RECORD_TIMINGS 2026-10-20 18:26:50 +03:00
Ian f4a7413963 feat(llm): run.json records every provider request's timing, provider and tokens in llm_requests 2026-10-20 18:26:50 +03:00
Ahmed Allam a0f8ea2cd3 fix(core): a parked agent keeps its pause reason and drains queued messages on resume
BudgetPausedError carries whether the operator pause caused it, so a resume that lands between the hook's refusal and the loop's handler still parks and continues instead of exiting under the stop policy. After the resume the agent drains its mailbox into the session before retrying the call, so a message queued while it was held is not skipped.
2026-10-18 14:58:16 +03:00
Ahmed Allam 6bd1cead1a fix(core): an operator pause holds under every budget policy and only resume_budget lifts it 2026-10-18 14:58:16 +03:00
Ahmed Allam ee22e2b8ed fix(agents): the running-process marker is matched only in the result header, never in command output 2026-10-18 02:48:44 +03:00
Ahmed Allam 2b98deb3d2 fix(agents): the non-TTY note goes into the Output section so renderers keep stripping the metadata block 2026-10-18 02:48:44 +03:00
Ahmed Allam 5e9e69a07e fix(agents): exec_command states the 30 s yield cap and that a running non-TTY process cannot be polled 2026-10-18 02:48:44 +03:00
Szymon NastalyandSzymon Nastaly 972f5097c2 feat(llm): extra request-body fields via LLM_EXTRA_BODY (#1495)
* feat(llm): extra request-body fields via LLM_EXTRA_BODY

Some provider parameters, such as OpenAI's service_tier and
safety_identifier, are read only from the request body, so
LLM_EXTRA_HEADERS can't set them. LLM_EXTRA_BODY merges a JSON object
into the body of every request to the configured model, on both routing
paths. A dedicated dedupe model gets its own DEDUPE_LLM_EXTRA_BODY
instead, as with headers.

* fix(config): decode saved JSON-object settings when loading

persist_current() saves env values verbatim, so LLM_EXTRA_HEADERS,
LLM_EXTRA_BODY and their DEDUPE_ variants land in cli-config.json as JSON
text. The loader passed that text unchanged to the dict fields, so a later
launch without the env var failed validation. Decode saved strings for
dict fields, as pydantic-settings does for env vars.

---------

Co-authored-by: Szymon Nastaly <[email protected]>
2026-10-15 16:37:51 -04:00
alex s ca20626f84 fix(models): replay malformed tool-call arguments as valid JSON (#1327)
* fix(models): replay malformed tool-call arguments as valid JSON

A model that emits a tool call with truncated or otherwise invalid JSON
arguments fails that one call, but the raw string is recorded in the
session. Strict OpenAI-compatible servers (vLLM, SGLang) validate every
assistant tool call in the request, so each later turn is rejected with
"Assistant tool call function.arguments must be valid JSON" and the agent
never recovers. The turn guard now rewrites such arguments on the way out
to a JSON object carrying the original text, alongside the existing
duplicate call-id repair.

* fix(agents): tell the model to re-issue tool calls whose arguments are not valid JSON

A streamed turn that ends mid-tool-call leaves the client with a JSON prefix
as the arguments. Instead of the SDK's generic parse-error result, the tool
wrapper now short-circuits with a clear 'not executed, re-issue with complete
arguments' message so the model recovers in one turn.

* fix(models): treat NaN/Infinity tool-call arguments as malformed

* test(cost): pin model resolution in the provider-prefixed cost test

* fix(models): treat over-nested tool-call arguments as malformed
2026-10-14 18:17:20 -04:00
ian-at-strix e21f9ce4cc test(resume-picker): put the pty in cbreak so the key-reader test can't hang on macOS (#1527) 2026-10-14 15:55:22 -04:00
yoni-at-strix 87f2d783d6 fix(reporting): drop the static-analysis disclaimer from dependency evidence (#1520) 2026-10-14 10:59:11 -04:00
Ahmed Allam 956a26715b feat(runtime): accept an awaitable extra_files in create_or_reuse so the sandbox create does not wait for it 2026-10-14 03:32:29 +03:00
Ahmed Allam 0f8db1fca8 fix(runtime): keep the default path blocking on the proxy endpoint and discard a sandbox created after a cancelled deferred bring-up 2026-10-14 00:48:53 +03:00
Ahmed Allam 4bf7541599 feat(runtime): defer the sandbox bring-up behind a DeferredSandboxSession when create_or_reuse is called with defer=True 2026-10-14 00:48:53 +03:00
Ian d8c00a8f4e fix(proxy): log all Caido SDK errors as warnings 2026-10-12 22:18:42 +03:00
Ian c5854e7be2 fix(proxy): log Caido timeouts as warnings 2026-10-12 22:18:42 +03:00
Ian 442b3c3df5 fix(tui): give the targets box only rows the agents panel would get, and always count hidden targets 2026-10-12 20:17:29 +03:00
Ian 710fe7b3b6 feat(tui): move the run's targets into their own sidebar box under the viewer 2026-10-12 20:17:29 +03:00
Ian b68cdb8edc feat(tui): list the run's targets in the sidebar stats panel 2026-10-12 20:17:29 +03:00
yoni-at-strix 429f6218f5 feat(reporting): accept code_locations on dependency reports (#1366) 2026-10-12 08:54:36 -07:00
yoni-at-strix f7a32ae62d docs(readme): add update instructions (#1501) 2026-10-10 20:31:50 -04:00
ian-at-strix 62b496430d feat(agents): let callers choose the top-level agent's finish tool (#1496) 2026-10-08 19:11:50 -04:00
alex s f5a900416b fix(mcp): load catalogs before agents discover connections (#1492)
* fix(mcp): load catalogs before agents discover connections

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

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

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

* feat(llm): retry 4xx model errors

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

* refactor: retry any 4xx or 5xx status

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

No files matched your search

+2
View File
@@ -31,6 +31,8 @@ jobs:
assert not any(name.startswith("strix/interface/tui/cmd/") for name in names)
PY
- name: Install into a clean environment and run the CLI
env:
UV_EXCLUDE_NEWER: 5 days
run: |
uv venv --python 3.12 /tmp/strix-smoke
uv pip install --python /tmp/strix-smoke/bin/python dist/*.whl
+1 -1
View File
@@ -28,7 +28,7 @@ Target-specific workflows built on the same engine:
- **Open-source CLI (self-hosted):** free, fully local, BYO LLM key, needs Docker. Best for local dev loops, air-gapped/offline, and full control.
```bash
curl -sSL https://strix.ai/install | bash # install
export STRIX_LLM="openrouter/z-ai/glm-5.3" # any LiteLLM model id
export STRIX_LLM="openrouter/z-ai/glm-5.3" # any LiteLLM model id, or vercel_ai_gateway/zai/glm-5.3
export LLM_API_KEY="<key>"
strix -n -t ./ --scan-mode quick --max-budget 10 # headless scan; always use -n
```
+1 -1
View File
@@ -31,7 +31,7 @@ Thank you for your interest in contributing to Strix! This guide will help you g
3. **Configure your LLM provider**
```bash
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="openrouter/z-ai/glm-5.3" # or vercel_ai_gateway/zai/glm-5.3
export LLM_API_KEY="your-api-key"
```
+23 -7
View File
@@ -84,7 +84,8 @@ Strix are autonomous AI penetration testing agents that act just like real hacke
curl -sSL https://strix.ai/install | bash
# Configure your AI provider
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="vercel_ai_gateway/zai/glm-5.3" # Vercel AI Gateway
# export STRIX_LLM="openrouter/z-ai/glm-5.3" # OpenRouter
export LLM_API_KEY="your-api-key"
# Run your first security assessment
@@ -94,6 +95,18 @@ strix --target ./app-directory
> [!NOTE]
> First run automatically pulls the sandbox Docker image. Results are saved to `strix_runs/<run-name>`
### Updating
```bash
# Self-update the standalone binary to the latest release
strix --update
# ...or re-run the installer
curl -sSL https://strix.ai/install | bash
```
Installed with pipx, uv, or pip? `strix --update` prints the matching upgrade command instead (`pipx upgrade strix-agent`, `uv tool upgrade strix-agent`, `pip install --upgrade strix-agent`). Strix also shows a notice with the right command when a new version is available.
---
## Ways to Run Strix
@@ -284,7 +297,8 @@ jobs:
### Configuration
```bash
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="vercel_ai_gateway/zai/glm-5.3" # Vercel AI Gateway
# export STRIX_LLM="openrouter/z-ai/glm-5.3" # OpenRouter
export LLM_API_KEY="your-api-key"
# Optional
@@ -340,12 +354,14 @@ Each server's tools are namespaced by `name`, for example `github_list_issues`.
**Recommended models for best results:**
- [Z.ai GLM-5.3 on OpenRouter](https://openrouter.ai/z-ai/glm-5.3) - `openrouter/z-ai/glm-5.3` (the default pick)
- [OpenAI GPT-5.4](https://openai.com/api/) - `openai/gpt-5.4`
- [Anthropic Claude Sonnet 4.6](https://claude.com/platform/api) - `anthropic/claude-sonnet-4-6`
- [Google Gemini 3 Pro Preview](https://cloud.google.com/vertex-ai) - `vertex_ai/gemini-3-pro-preview`
- [Z.ai GLM-5.3 on Vercel AI Gateway](https://vercel.com/ai-gateway/models/glm-5.3) - `vercel_ai_gateway/zai/glm-5.3` (recommended)
- [Z.ai GLM-5.3 on OpenRouter](https://openrouter.ai/z-ai/glm-5.3) - `openrouter/z-ai/glm-5.3` (recommended)
- [DeepSeek V4.1 Flash on OpenRouter](https://openrouter.ai/deepseek/deepseek-v4.1-flash-1017) - `openrouter/deepseek/deepseek-v4.1-flash-1017`
- [OpenAI GPT-6.1 Sol](https://openai.com/api/) - `openai/gpt-6.1-sol` (needs [Daybreak](https://openai.com/daybreak/) access)
- [Anthropic Claude Opus 5.5](https://claude.com/platform/api) - `anthropic/claude-opus-5-5` (needs [CVP](https://support.claude.com/en/articles/14604842-cyber-verification-program) access)
- [Google Gemini 4 Argon](https://cloud.google.com/vertex-ai) - `vertex_ai/gemini-4-argon` (needs [Fairwind](https://deepmind.google/fairwind-program/) access)
- [DeepSeek V4 Pro](https://platform.deepseek.com) - `deepseek/deepseek-v4-pro`
- [Moonshot Kimi K3](https://platform.kimi.ai) - `moonshot/kimi-k3`
- [Moonshot Kimi K3.1 Pro](https://platform.kimi.ai) - `moonshot/kimi-k3.1-pro`
See the [LLM Providers documentation](https://docs.strix.ai/llm-providers/overview) for all supported providers including Vertex AI, Bedrock, Azure, and local models.
+69 -5
View File
@@ -176,7 +176,7 @@ USER root
# overwrite a root-owned binary under /usr/local/bin, which otherwise fails with
# "cannot move binary" and aborts the scan. Pin the initial version for
# reproducible builds; self-update then pulls fresh detectors at runtime.
ARG TRUFFLEHOG_VERSION=3.95.9
ARG TRUFFLEHOG_VERSION=3.99.2
RUN curl -sSfL https://raw.githubusercontent.com/trufflesecurity/trufflehog/main/scripts/install.sh | sh -s -- -b /home/pentester/.local/bin "v${TRUFFLEHOG_VERSION}" && \
chown -R pentester:pentester /home/pentester/.local
RUN set -eux; \
@@ -186,20 +186,84 @@ RUN set -eux; \
aarch64|arm64) GITLEAKS_ARCH="arm64" ;; \
*) echo "Unsupported architecture: $ARCH" >&2; exit 1 ;; \
esac; \
TAG="$(curl -fsSL https://api.github.com/repos/gitleaks/gitleaks/releases/latest | jq -r .tag_name)"; \
TAG="$(curl -fsSLI -o /dev/null -w '%{url_effective}' https://github.com/gitleaks/gitleaks/releases/latest | sed 's#.*/##')"; \
curl -fsSL "https://github.com/gitleaks/gitleaks/releases/download/${TAG}/gitleaks_${TAG#v}_linux_${GITLEAKS_ARCH}.tar.gz" -o /tmp/gitleaks.tgz; \
tar -xzf /tmp/gitleaks.tgz -C /tmp; \
install -m 0755 /tmp/gitleaks /usr/local/bin/gitleaks; \
rm -f /tmp/gitleaks /tmp/gitleaks.tgz
RUN curl -sfL https://raw.githubusercontent.com/aquasecurity/trivy/main/contrib/install.sh | sh -s -- -b /usr/local/bin
ARG TRIVY_VERSION=0.76.1
ARG TRIVY_SHA256_X86=c7cbe3137fb8a48a17574bede3dd6beb16742d1e9772929940fc90082c39d9c0
ARG TRIVY_SHA256_AARCH64=e174dff2a6dba5e3ae41f8d0537a6324405bd08132497340304c6b20e0f16a1f
RUN set -eux; \
case "$(uname -m)" in \
x86_64) TRIVY_ARCH="64bit"; TRIVY_SHA256="${TRIVY_SHA256_X86}" ;; \
aarch64|arm64) TRIVY_ARCH="ARM64"; TRIVY_SHA256="${TRIVY_SHA256_AARCH64}" ;; \
*) echo "Unsupported architecture: $(uname -m)" >&2; exit 1 ;; \
esac; \
curl -fsSL "https://github.com/aquasecurity/trivy/releases/download/v${TRIVY_VERSION}/trivy_${TRIVY_VERSION}_Linux-${TRIVY_ARCH}.tar.gz" -o /tmp/trivy.tgz; \
echo "${TRIVY_SHA256} /tmp/trivy.tgz" | sha256sum -c -; \
tar -xzf /tmp/trivy.tgz -C /tmp trivy; \
install -m 0755 /tmp/trivy /usr/local/bin/trivy; \
rm -f /tmp/trivy /tmp/trivy.tgz; \
trivy --version
RUN apt-get install -y wapiti
# CVE-2026-111497: out-of-bounds write in the brotli decoder, fixed in brotli
# 1.2.1 / Debian 1.2.0-4.1. kali-rolling still ships 1.2.0-4, and node, curl and
# python3 in this image all load the shared libbrotlidec, so every HTTP client here
# would decode a malicious `Content-Encoding: br` response with the vulnerable
# code. Install the fixed Debian packages until Kali picks them up.
ARG BROTLI_DEB_VERSION=1.2.0-4.1
RUN set -eux; \
ARCH="$(dpkg --print-architecture)"; \
for pkg in libbrotli1 python3-brotli; do \
curl -fsSL "https://deb.debian.org/debian/pool/main/b/brotli/${pkg}_${BROTLI_DEB_VERSION}_${ARCH}.deb" -o "/tmp/${pkg}.deb"; \
done; \
dpkg -i /tmp/libbrotli1.deb /tmp/python3-brotli.deb; \
rm -f /tmp/libbrotli1.deb /tmp/python3-brotli.deb; \
test "$(dpkg-query -W -f='${Version}' libbrotli1)" = "${BROTLI_DEB_VERSION}"; \
node -e 'const z=require("zlib");z.brotliDecompressSync(z.brotliCompressSync(Buffer.from("ok")))'
ARG OPENGREP_VERSION=1.31.0
ARG OPENGREP_SHA256_X86=dbbe3c2058dbe600c449a8ba22ea79a1643a6660658400090d0323a8d0d31147
ARG OPENGREP_SHA256_AARCH64=2b713a29a00237f85408b83002fc111db5b94c7fa8d263f0a93184a960655d9d
RUN set -eux; \
case "$(uname -m)" in \
x86_64) OPENGREP_ARCH="x86"; OPENGREP_SHA256="${OPENGREP_SHA256_X86}" ;; \
aarch64|arm64) OPENGREP_ARCH="aarch64"; OPENGREP_SHA256="${OPENGREP_SHA256_AARCH64}" ;; \
*) echo "Unsupported architecture: $(uname -m)" >&2; exit 1 ;; \
esac; \
curl -fsSL "https://github.com/opengrep/opengrep/releases/download/v${OPENGREP_VERSION}/opengrep_manylinux_${OPENGREP_ARCH}" -o /tmp/opengrep; \
echo "${OPENGREP_SHA256} /tmp/opengrep" | sha256sum -c -; \
install -m 0755 /tmp/opengrep /usr/local/bin/opengrep; \
rm -f /tmp/opengrep; \
opengrep --version
# Opengrep ships no rule registry of its own; bake the community ruleset so
# scans run offline. Keep only rule YAML and LICENSE; drop tests, fixtures,
# tooling, and rules marked with a proprietary license.
ARG OPENGREP_RULES_REF=f1d2b562b414783763fd02a6ed2736eaed622efa
RUN set -eux; \
mkdir -p /opt/opengrep-rules; \
curl -fsSL "https://codeload.github.com/opengrep/opengrep-rules/tar.gz/${OPENGREP_RULES_REF}" \
| tar -xz --strip-components=1 -C /opt/opengrep-rules; \
cd /opt/opengrep-rules; \
rm -rf .github scripts stats; \
find . -maxdepth 1 -type f ! -name LICENSE -delete; \
find . -mindepth 2 -type f \( ! \( -name '*.yaml' -o -name '*.yml' \) -o -name '*.test.yaml' -o -name '*.test.yml' -o -name '.*' \) -delete; \
grep -rlZ 'license: proprietary' . | xargs -0 -r rm -f; \
find . -type d -empty -delete; \
chmod -R a+rX /opt/opengrep-rules
# Strix-authored rules that close gaps in the community ruleset.
COPY containers/opengrep-rules/ /opt/opengrep-rules/strix/
RUN chmod -R a+rX /opt/opengrep-rules/strix
USER pentester
RUN pipx install semgrep && \
pipx install bandit
RUN pipx install bandit
RUN npm install -g jshint
@@ -0,0 +1,37 @@
rules:
- id: github-owned-action-not-pinned-to-commit-sha
patterns:
- pattern-inside: "{steps: ...}"
- pattern: |
uses: "$USES"
- metavariable-regex:
metavariable: $USES
regex: '^(actions|github)/'
- metavariable-pattern:
metavariable: $USES
language: generic
patterns:
- pattern-not-regex: "@[0-9a-f]{40}$"
message: >-
GitHub-owned action `$USES` is referenced by a mutable tag or branch
instead of a full-length commit SHA. Tags can be moved, so a compromised
or force-pushed release changes what this workflow runs. Pin to the
commit SHA of the release you reviewed and let a bot keep it current.
languages:
- yaml
severity: WARNING
metadata:
cwe:
- "CWE-1357: Reliance on Insufficiently Trustworthy Component"
- "CWE-353: Missing Support for Integrity Check"
owasp: A06:2021 - Vulnerable and Outdated Components
references:
- https://docs.github.com/en/actions/security-guides/security-hardening-for-github-actions#using-third-party-actions
category: security
technology:
- github-actions
subcategory:
- vuln
likelihood: LOW
impact: LOW
confidence: HIGH
+45 -4
View File
@@ -8,7 +8,7 @@ Configure Strix using environment variables or a config file.
## LLM Configuration
<ParamField path="STRIX_LLM" type="string" required>
Model name in LiteLLM format (e.g., `openrouter/z-ai/glm-5.3`, `openai/gpt-5.4`).
Model name in LiteLLM format (e.g., `openrouter/z-ai/glm-5.3`, `openai/gpt-6.1-sol`).
</ParamField>
<ParamField path="LLM_API_KEY" type="string">
@@ -19,10 +19,29 @@ Configure Strix using environment variables or a config file.
Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`.
</ParamField>
<ParamField path="STRIX_LLM_INPUT_COST_PER_MTOK" type="number">
Price per 1M input tokens in USD for `STRIX_LLM`. Use it for gateway or self-hosted model ids that LiteLLM can't price.
</ParamField>
<ParamField path="STRIX_LLM_OUTPUT_COST_PER_MTOK" type="number">
Price per 1M output tokens in USD for `STRIX_LLM`.
</ParamField>
<ParamField path="STRIX_LLM_CACHED_INPUT_COST_PER_MTOK" type="number">
Price per 1M cached input tokens in USD for `STRIX_LLM`. If unset, cached tokens use the input price.
</ParamField>
<Note>
Setting any of these prices replaces the provider-reported and LiteLLM cost for `STRIX_LLM` in reports and `--max-budget`. An unset input or output price counts as `0`, so set both. A separate `STRIX_DEDUPE_MODEL` keeps its LiteLLM price.
</Note>
<ParamField path="STRIX_API_TYPE" type="string">
Select the OpenAI API path for the model: `responses` or `chat_completions`.
By default, a custom `LLM_API_BASE` uses chat completions. Set this variable
when your gateway requires the other API. Also accepts `STRIX_FORCE_API`.
By default the model decides: Responses when LiteLLM's model catalog lists
`/v1/responses` for it (e.g. `gpt-5`), chat completions otherwise. A custom
`LLM_API_BASE` does not change this, so set `STRIX_API_TYPE=chat_completions`
when your gateway only serves chat completions. Applies to bare and `openai/`
model names. Also accepts `STRIX_FORCE_API`.
</ParamField>
<ParamField path="LLM_EXTRA_HEADERS" type="string">
@@ -33,6 +52,18 @@ Configure Strix using environment variables or a config file.
the LiteLLM and native OpenAI routing paths.
</ParamField>
<ParamField path="LLM_EXTRA_BODY" type="string">
Extra fields merged into the JSON body of every request to the main model, as a JSON object
(e.g. `{"service_tier":"flex","safety_identifier":"9f86d081..."}`). Use it
for provider parameters Strix doesn't expose, such as OpenAI's `service_tier`
or `safety_identifier`. Applies to both the LiteLLM and native OpenAI routing
paths. Fields are not validated, so the provider or gateway must accept them;
unknown fields are usually rejected. Extra fields can override fields Strix
sends and break requests, so don't set request-shaping fields such as `model`, `messages`, `input`,
`tools` or `stream`, and use `STRIX_REASONING_EFFORT` rather than
`reasoning_effort`.
</ParamField>
<ParamField path="LLM_TIMEOUT" default="300" type="integer">
Request timeout in seconds for LLM calls.
</ParamField>
@@ -88,6 +119,12 @@ affecting the agents that do the actual testing.
inherits `LLM_EXTRA_HEADERS`; set this when its endpoint needs custom headers.
</ParamField>
<ParamField path="DEDUPE_LLM_EXTRA_BODY" type="string">
Optional JSON object of extra request-body fields sent on every
deduplication-model request. A dedicated dedupe model never inherits
`LLM_EXTRA_BODY`; set this when its endpoint needs them too.
</ParamField>
<ParamField path="STRIX_DEDUPE_REASONING_EFFORT" type="string">
Reasoning effort for the deduplication model. Defaults to the model's own
baseline when unset.
@@ -123,6 +160,10 @@ affecting the agents that do the actual testing.
Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL).
</ParamField>
<ParamField path="STRIX_RUN_RECORD_TIMINGS" default="false" type="boolean">
Record every LLM request's and tool call's timing in `run.json`, under `llm_requests` and `tool_calls`. Adds about 0.5 MB to a typical scan's `run.json`.
</ParamField>
<ParamField path="TRACELOOP_BASE_URL" type="string">
OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only.
</ParamField>
@@ -145,7 +186,7 @@ When remote vars are set, Strix dual-writes telemetry to both local JSONL and th
## Docker Configuration
<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.3.0" type="string">
<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.4.0" type="string">
Docker image to use for the sandbox container.
</ParamField>
+1 -1
View File
@@ -35,8 +35,8 @@
"llm-providers/overview",
"llm-providers/openai",
"llm-providers/anthropic",
"llm-providers/openrouter",
"llm-providers/vercel-ai-gateway",
"llm-providers/openrouter",
"llm-providers/vertex",
"llm-providers/bedrock",
"llm-providers/azure",
+15 -1
View File
@@ -73,7 +73,20 @@ Strix uses a graph of specialized agents for comprehensive security testing:
## Quick Example
```bash
<CodeGroup>
```bash Vercel AI Gateway
# Install
curl -sSL https://strix.ai/install | bash
# Configure
export STRIX_LLM="vercel_ai_gateway/zai/glm-5.3"
export LLM_API_KEY="your-api-key"
# Scan
strix --target ./your-app
```
```bash OpenRouter
# Install
curl -sSL https://strix.ai/install | bash
@@ -84,6 +97,7 @@ export LLM_API_KEY="your-api-key"
# Scan
strix --target ./your-app
```
</CodeGroup>
## Community
+7 -3
View File
@@ -6,7 +6,7 @@ description: "Configure Strix with Claude models"
## Setup
```bash
export STRIX_LLM="anthropic/claude-sonnet-4-6"
export STRIX_LLM="anthropic/claude-sonnet-5-5"
export LLM_API_KEY="sk-ant-..."
```
@@ -14,8 +14,12 @@ export LLM_API_KEY="sk-ant-..."
| Model | Description |
|-------|-------------|
| `anthropic/claude-sonnet-4-6` | Best balance of intelligence and speed |
| `anthropic/claude-opus-4-6` | Maximum capability for deep analysis |
| `anthropic/claude-sonnet-5-5` | Best balance of intelligence and speed |
| `anthropic/claude-opus-5-5` | Maximum capability for deep analysis |
<Note>
Security testing with Claude needs the [Cyber Verification Program](https://support.claude.com/en/articles/14604842-cyber-verification-program) (CVP). Without it the safety classifiers block most offensive tasks. Red Team Access covers authorized penetration testing.
</Note>
## Get API Key
+7 -3
View File
@@ -6,7 +6,7 @@ description: "Configure Strix with OpenAI models via Azure"
## Setup
```bash
export STRIX_LLM="azure/your-gpt5-deployment"
export STRIX_LLM="azure/your-gpt6-deployment"
export AZURE_API_KEY="your-azure-api-key"
export AZURE_API_BASE="https://your-resource.openai.azure.com"
export AZURE_API_VERSION="2025-11-01-preview"
@@ -24,7 +24,7 @@ export AZURE_API_VERSION="2025-11-01-preview"
## Example
```bash
export STRIX_LLM="azure/gpt-5.4-deployment"
export STRIX_LLM="azure/gpt-6.1-sol-deployment"
export AZURE_API_KEY="abc123..."
export AZURE_API_BASE="https://mycompany.openai.azure.com"
export AZURE_API_VERSION="2025-11-01-preview"
@@ -33,5 +33,9 @@ export AZURE_API_VERSION="2025-11-01-preview"
## Prerequisites
1. Create an Azure OpenAI resource
2. Deploy a model (e.g., GPT-5.4)
2. Deploy a model such as GPT-6.1 Sol
3. Get the endpoint URL and API key from the Azure portal
<Note>
Security testing with OpenAI models needs [Daybreak](https://openai.com/daybreak/) access, also on Azure.
</Note>
+8 -5
View File
@@ -6,7 +6,7 @@ description: "Configure Strix with models via AWS Bedrock"
## Setup
```bash
export STRIX_LLM="bedrock/anthropic.claude-4-5-sonnet-20251022-v1:0"
export STRIX_LLM="bedrock/us.anthropic.claude-sonnet-5-5"
```
No API key required—uses AWS credentials from environment.
@@ -36,10 +36,13 @@ Automatically uses instance role credentials.
| Model | Description |
|-------|-------------|
| `bedrock/anthropic.claude-4-5-sonnet-20251022-v1:0` | Claude 4.5 Sonnet |
| `bedrock/anthropic.claude-4-5-opus-20251022-v1:0` | Claude 4.5 Opus |
| `bedrock/anthropic.claude-4-5-haiku-20251022-v1:0` | Claude 4.5 Haiku |
| `bedrock/amazon.titan-text-premier-v2:0` | Amazon Titan Premier v2 |
| `bedrock/us.anthropic.claude-sonnet-5-5` | Claude Sonnet 5.5 |
| `bedrock/us.anthropic.claude-opus-5-5` | Claude Opus 5.5 |
| `bedrock/us.anthropic.claude-haiku-5-5` | Claude Haiku 5.5 |
<Note>
Security testing with Claude needs the [Cyber Verification Program](https://support.claude.com/en/articles/14604842-cyber-verification-program) (CVP), also on Bedrock.
</Note>
## Prerequisites
+15
View File
@@ -68,6 +68,21 @@ export LLM_API_KEY="your-bearer-token" # sent as Authorization: Bearer
export LLM_EXTRA_HEADERS='{"X-Feature-Key":"value","X-Tenant":"acme"}'
```
### Extra request-body fields
Some provider parameters go in the request body rather than a header, and
Strix has no dedicated setting for them. Set them with `LLM_EXTRA_BODY` as a
JSON object — they are merged into the body of every request to the model. For example, to
use OpenAI's cheaper flex processing and attribute requests to a user with a
safety identifier (a hash, never a raw email):
```bash
export LLM_EXTRA_BODY='{"service_tier":"flex","safety_identifier":"9f86d081884c7d65"}'
```
Fields aren't validated, so use only fields your provider accepts, and don't
set fields Strix already sends, such as `model`, `messages` or `stream`.
For endpoints behind a private CA, point Strix at your certificate bundle with
the standard `SSL_CERT_FILE=/path/to/ca-bundle.pem` — never disable TLS
verification against a real endpoint.
+5 -4
View File
@@ -8,7 +8,7 @@ description: "Configure Strix with Novita AI models"
## Setup
```bash
export STRIX_LLM="openai/moonshotai/kimi-k2.5"
export STRIX_LLM="openai/zai-org/glm-5.3"
export LLM_API_KEY="your-novita-api-key"
export LLM_API_BASE="https://api.novita.ai/openai"
```
@@ -18,11 +18,12 @@ export LLM_API_BASE="https://api.novita.ai/openai"
| Model | Configuration |
|-------|---------------|
| GLM-5.3 | `openai/zai-org/glm-5.3` |
| Kimi K3.1 Pro | `openai/moonshotai/kimi-k3.1-pro` |
| Kimi K3 | `openai/moonshotai/kimi-k3` |
| DeepSeek V4.1 Flash | `openai/deepseek/deepseek-v4.1-flash-1017` |
| DeepSeek V4 Pro | `openai/deepseek/deepseek-v4-pro` |
| Kimi K2.5 | `openai/moonshotai/kimi-k2.5` |
| GLM-5 | `openai/zai-org/glm-5` |
| MiniMax M2.5 | `openai/minimax/minimax-m2.5` |
| MiniMax M3.1 Flash | `openai/minimax/minimax-m3.1-flash` |
| GLM-5.3 Flash | `openai/zai-org/glm-5.3-flash` |
## Get API Key
+13 -3
View File
@@ -6,13 +6,23 @@ description: "Configure Strix with OpenAI models"
## Setup
```bash
export STRIX_LLM="openai/gpt-5.4"
export STRIX_LLM="openai/gpt-6.1-sol"
export LLM_API_KEY="sk-..."
```
## Available Models
See [OpenAI Models Documentation](https://platform.openai.com/docs/models) for the full list of available models.
| Model | Description |
|-------|-------------|
| `openai/gpt-6.1-sol` | Best balance of capability and cost for agentic security work |
| `openai/gpt-6-astra` | Flagship, maximum capability |
| `openai/gpt-6-luna` | Fast and cheap |
See [OpenAI Models Documentation](https://platform.openai.com/docs/models) for the full list.
<Note>
Security testing with OpenAI models needs [Daybreak](https://openai.com/daybreak/) (Trusted Access for Cyber). Without it the models refuse most offensive tasks. [Request access](https://openai.com/form/enterprise-trusted-access-for-cyber/).
</Note>
## Get API Key
@@ -25,7 +35,7 @@ See [OpenAI Models Documentation](https://platform.openai.com/docs/models) for t
For OpenAI-compatible APIs:
```bash
export STRIX_LLM="openai/gpt-5.4"
export STRIX_LLM="openai/gpt-6.1-sol"
export LLM_API_KEY="your-key"
export LLM_API_BASE="https://your-proxy.com/v1"
```
+12 -6
View File
@@ -18,13 +18,19 @@ Access any model on OpenRouter using the format `openrouter/<provider>/<model>`:
| Model | Configuration |
|-------|---------------|
| GLM-5.3 (default) | `openrouter/z-ai/glm-5.3` |
| GPT-5.4 | `openrouter/openai/gpt-5.4` |
| Claude Sonnet 4.6 | `openrouter/anthropic/claude-sonnet-4.6` |
| Gemini 3 Pro | `openrouter/google/gemini-3-pro-preview` |
| GLM-5.3 (recommended) | `openrouter/z-ai/glm-5.3` |
| GPT-6.1 Sol | `openrouter/openai/gpt-6.1-sol` |
| GPT-6 Astra | `openrouter/openai/gpt-6-astra` |
| Claude Opus 5.5 | `openrouter/anthropic/claude-opus-5.5` |
| Claude Sonnet 5.5 | `openrouter/anthropic/claude-sonnet-5.5` |
| Gemini 4 Argon | `openrouter/google/gemini-4-argon` |
| Gemini 3.8 Flash | `openrouter/google/gemini-3.8-flash` |
| DeepSeek V4.1 Flash | `openrouter/deepseek/deepseek-v4.1-flash-1017` |
| DeepSeek V4 Pro | `openrouter/deepseek/deepseek-v4-pro` |
| Kimi K3 | `openrouter/moonshotai/kimi-k3` |
| GLM-4.7 | `openrouter/z-ai/glm-4.7` |
| Kimi K3.1 Pro | `openrouter/moonshotai/kimi-k3.1-pro` |
| GLM-5.3 Flash | `openrouter/z-ai/glm-5.3-flash` |
OpenAI, Anthropic and Gemini models keep their provider safeguards on OpenRouter. Security testing with them needs [Daybreak](https://openai.com/daybreak/), [CVP](https://support.claude.com/en/articles/14604842-cyber-verification-program) or [Fairwind](https://deepmind.google/fairwind-program/) access.
## Get API Key
+28 -14
View File
@@ -9,19 +9,33 @@ Strix uses [LiteLLM](https://docs.litellm.ai/docs/providers) for model compatibi
Set your model and API key:
| Model | Provider | Configuration |
| -------------------- | ----------------- | -------------------------------- |
| GLM-5.3 (default) | Z.ai (OpenRouter) | `openrouter/z-ai/glm-5.3` |
| GPT-5.4 | OpenAI | `openai/gpt-5.4` |
| Claude Sonnet 4.6 | Anthropic | `anthropic/claude-sonnet-4-6` |
| Gemini 3 Pro | Google Vertex | `vertex_ai/gemini-3-pro-preview` |
| DeepSeek V4 Pro | DeepSeek | `deepseek/deepseek-v4-pro` |
| Kimi K3 | Moonshot | `moonshot/kimi-k3` |
| Model | Provider | Configuration |
| --------------------- | ------------------------- | -------------------------------- |
| GLM-5.3 (recommended) | Vercel AI Gateway | `vercel_ai_gateway/zai/glm-5.3` |
| GLM-5.3 (recommended) | OpenRouter | `openrouter/z-ai/glm-5.3` |
| GPT-6.1 Sol | OpenAI | `openai/gpt-6.1-sol` |
| Claude Opus 5.5 | Anthropic | `anthropic/claude-opus-5-5` |
| Claude Sonnet 5.5 | Anthropic | `anthropic/claude-sonnet-5-5` |
| Gemini 4 Argon | Google Vertex | `vertex_ai/gemini-4-argon` |
| DeepSeek V4.1 Flash | DeepSeek | `deepseek/deepseek-flash` |
| DeepSeek V4 Pro | DeepSeek | `deepseek/deepseek-v4-pro` |
| Kimi K3.1 Pro | Moonshot | `moonshot/kimi-k3.1-pro` |
```bash
<Note>
Frontier OpenAI, Anthropic and Gemini models need a cyber access program for security testing: [Daybreak](https://openai.com/daybreak/) for OpenAI, the [Cyber Verification Program](https://support.claude.com/en/articles/14604842-cyber-verification-program) for Anthropic, and the [Fairwind Program](https://deepmind.google/fairwind-program/) for Gemini.
</Note>
<CodeGroup>
```bash Vercel AI Gateway
export STRIX_LLM="vercel_ai_gateway/zai/glm-5.3"
export LLM_API_KEY="your-api-key"
```
```bash OpenRouter
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export LLM_API_KEY="your-api-key"
```
</CodeGroup>
## Local Models
@@ -47,7 +61,7 @@ See the [Local Models guide](/llm-providers/local) for setup instructions and re
Access 100+ models through a single API.
</Card>
<Card title="Vercel AI Gateway" href="/llm-providers/vercel-ai-gateway">
Access models from multiple providers through one endpoint.
Access 100+ models through one OpenAI-compatible endpoint.
</Card>
<Card title="Google Vertex AI" href="/llm-providers/vertex">
Gemini 3 models via Google Cloud.
@@ -69,9 +83,9 @@ Use LiteLLM's `provider/model-name` format:
```
openrouter/z-ai/glm-5.3
openai/gpt-5.4
anthropic/claude-sonnet-4-6
vertex_ai/gemini-3-pro-preview
bedrock/anthropic.claude-4-5-sonnet-20251022-v1:0
openai/gpt-6.1-sol
anthropic/claude-sonnet-5-5
vertex_ai/gemini-4-argon
bedrock/us.anthropic.claude-sonnet-5-5
ollama/llama4
```
+26 -10
View File
@@ -1,32 +1,47 @@
---
title: "Vercel AI Gateway"
description: "Configure Strix with models via Vercel AI Gateway"
description: "Configure Strix with models through Vercel AI Gateway"
---
[Vercel AI Gateway](https://vercel.com/docs/ai-gateway) provides an OpenAI-compatible API for models from multiple providers.
[Vercel AI Gateway](https://vercel.com/docs/ai-gateway) provides access to 100+ models from multiple providers through one OpenAI-compatible API.
## Setup
Create an [AI Gateway API key](https://vercel.com/docs/ai-gateway/authentication-and-byok), then configure Strix:
```bash
export STRIX_LLM="openai/anthropic/claude-opus-5"
export STRIX_LLM="vercel_ai_gateway/zai/glm-5.3"
export LLM_API_KEY="your-ai-gateway-api-key"
export LLM_API_BASE="https://ai-gateway.vercel.sh/v1"
```
The first `openai/` segment tells Strix to use its OpenAI-compatible client. The remaining value is the [AI Gateway model ID](https://vercel.com/docs/ai-gateway/models-and-providers).
The `vercel_ai_gateway/` prefix selects the Gateway. The remaining value is the [AI Gateway model ID](https://vercel.com/docs/ai-gateway/models-and-providers).
## Available Models
Use any language model returned by the AI Gateway models endpoint:
Access any model on the Gateway using the format `vercel_ai_gateway/<provider>/<model>`:
| Model | Configuration |
|-------|---------------|
| GLM-5.3 (recommended) | `vercel_ai_gateway/zai/glm-5.3` |
| GPT-6.1 Sol | `vercel_ai_gateway/openai/gpt-6.1-sol` |
| GPT-6 Astra | `vercel_ai_gateway/openai/gpt-6-astra` |
| Claude Opus 5.5 | `vercel_ai_gateway/anthropic/claude-opus-5.5` |
| Claude Sonnet 5.5 | `vercel_ai_gateway/anthropic/claude-sonnet-5.5` |
| Gemini 4 Argon | `vercel_ai_gateway/google/gemini-4-argon` |
| Gemini 3.8 Flash | `vercel_ai_gateway/google/gemini-3.8-flash` |
| DeepSeek V4.1 Flash | `vercel_ai_gateway/deepseek/deepseek-v4.1-flash-1017` |
| DeepSeek V4 Pro | `vercel_ai_gateway/deepseek/deepseek-v4-pro` |
| Kimi K3.1 Pro | `vercel_ai_gateway/moonshotai/kimi-k3.1-pro` |
| GLM-5.3 Flash | `vercel_ai_gateway/zai/glm-5.3-flash` |
OpenAI, Anthropic and Gemini models keep their provider safeguards on the Gateway. Security testing with them needs [Daybreak](https://openai.com/daybreak/), [CVP](https://support.claude.com/en/articles/14604842-cyber-verification-program) or [Fairwind](https://deepmind.google/fairwind-program/) access.
The full list is at the [AI Gateway models page](https://vercel.com/ai-gateway/models) or the models endpoint:
```text
https://ai-gateway.vercel.sh/v1/models
```
Prefix its model ID with `openai/` when setting `STRIX_LLM`. For example, the Gateway model ID `anthropic/claude-opus-5` becomes `openai/anthropic/claude-opus-5` in Strix.
## Get API Key
1. Open the [AI Gateway API key settings](https://vercel.com/docs/ai-gateway/authentication-and-byok)
@@ -35,5 +50,6 @@ Prefix its model ID with `openai/` when setting `STRIX_LLM`. For example, the Ga
## Benefits
- Access models from multiple providers through one endpoint
- Track Gateway usage and cost in Vercel
- **Single API** — Access models from OpenAI, Anthropic, Google, Z.ai, and more
- **Bring your own keys** — Route through your own provider accounts when you want
- **Cost tracking** — Monitor Gateway usage and cost in Vercel
+7 -3
View File
@@ -6,7 +6,7 @@ description: "Configure Strix with Gemini models via Google Cloud"
## Setup
```bash
export STRIX_LLM="vertex_ai/gemini-3-pro-preview"
export STRIX_LLM="vertex_ai/gemini-4-argon"
```
No API key required—uses Google Cloud Application Default Credentials.
@@ -29,8 +29,12 @@ export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"
| Model | Description |
|-------|-------------|
| `vertex_ai/gemini-3-pro-preview` | Best overall performance for security testing |
| `vertex_ai/gemini-3-flash-preview` | Faster and cheaper |
| `vertex_ai/gemini-4-argon` | Best overall performance for security testing |
| `vertex_ai/gemini-3.8-flash` | Faster and cheaper |
<Note>
Security testing with Gemini needs the [Fairwind Program](https://deepmind.google/fairwind-program/). Without it the cyber guardrails block most offensive tasks.
</Note>
## Project Configuration
+11 -4
View File
@@ -25,15 +25,22 @@ description: "Install Strix and run your first security scan"
## Configuration
Set your LLM provider:
Set your LLM provider. We recommend GLM-5.3 through [Vercel AI Gateway](/llm-providers/vercel-ai-gateway) or [OpenRouter](/llm-providers/openrouter):
```bash
export STRIX_LLM="openrouter/z-ai/glm-5.3"
<CodeGroup>
```bash Vercel AI Gateway
export STRIX_LLM="vercel_ai_gateway/zai/glm-5.3"
export LLM_API_KEY="your-api-key"
```
```bash OpenRouter
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export LLM_API_KEY="your-api-key"
```
</CodeGroup>
<Tip>
For best results, use `openrouter/z-ai/glm-5.3` (the default pick), `openai/gpt-5.4`, `anthropic/claude-opus-4-6`, or `openai/gpt-5.2`.
For best results, use `zai/glm-5.3`, `deepseek/deepseek-v4.1-flash-1017`, `anthropic/claude-opus-5-5` (needs [CVP](https://support.claude.com/en/articles/14604842-cyber-verification-program) access), or `openai/gpt-6.1-sol` (needs [Daybreak](https://openai.com/daybreak/) access).
</Tip>
## Run Your First Scan
+1 -1
View File
@@ -49,7 +49,7 @@ Strix runs inside a Kali Linux-based Docker container with a comprehensive set o
| Tool | Description |
| ------------------------------------------------------- | --------------------------------------------- |
| [Semgrep](https://github.com/semgrep/semgrep) | Fast SAST and custom rule matching |
| [Opengrep](https://github.com/opengrep/opengrep) | Fast SAST and custom rule matching |
| [ast-grep](https://ast-grep.github.io) | Structural AST/CST-aware code search (`sg`) |
| [Tree-sitter](https://tree-sitter.github.io/tree-sitter/) | Syntax tree parsing and symbol extraction (Java/JS/TS/Python/Go/Bash/JSON/YAML grammars pre-configured) |
| [Bandit](https://bandit.readthedocs.io) | Python security linter |
+1 -1
View File
@@ -33,7 +33,7 @@ sqlmap -u "https://example.com/page?id=1"
```bash
# Fast SAST triage
semgrep --config auto ./src
opengrep scan --config /opt/opengrep-rules ./src
# Structural AST search
sg scan ./src
+1 -1
View File
@@ -46,7 +46,7 @@ Thorough penetration testing. Best for:
**Duration**: 1-4 hours depending on target complexity
**White-box behavior**: Runs broad source-aware triage (`semgrep`, AST structural search, secrets, supply-chain checks) and then systematically validates top candidates dynamically.
**White-box behavior**: Runs broad source-aware triage (`opengrep`, AST structural search, secrets, supply-chain checks) and then systematically validates top candidates dynamically.
<Note>
Deep mode is the default. It explores edge cases, chained vulnerabilities, and complex attack paths.
+6
View File
@@ -73,6 +73,12 @@ dev = [
[tool.pytest.ini_options]
asyncio_mode = "auto"
[tool.uv]
# Supply-chain cooldown: resolve only distributions that have been on the index for
# at least 5 days, so a poisoned release (litellm 1.107.0, 2026-10-21) is never
# picked up by `uv lock` / `uv add` before it can be caught and yanked.
exclude-newer = "5 days"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
+1 -1
View File
@@ -4,7 +4,7 @@ set -euo pipefail
APP=strix
REPO="usestrix/strix"
STRIX_IMAGE="ghcr.io/usestrix/strix-sandbox:1.3.0"
STRIX_IMAGE="ghcr.io/usestrix/strix-sandbox:1.4.0"
MUTED='\033[0;2m'
RED='\033[0;31m'
@@ -63,7 +63,7 @@ jobs:
fi
```
Then tell the user to add two repository secrets: `STRIX_LLM` (model id, for example `openai/gpt-5.4`) and `LLM_API_KEY` (the provider key). Do not create these values yourself.
Then tell the user to add two repository secrets: `STRIX_LLM` (model id, for example `openrouter/z-ai/glm-5.3`) and `LLM_API_KEY` (the provider key). Do not create these values yourself.
Notes:
- In CI/headless runs Strix automatically scopes to the PR's changed files (`--scope-mode auto`). If diff resolution fails, keep `fetch-depth: 0` or set `--diff-base` to the PR's actual base branch — use `origin/${{ github.base_ref }}` in GitHub Actions rather than a hard-coded `origin/main`, since repos use different default branches.
@@ -47,7 +47,7 @@ If unsure and the user has (or will create) an app.strix.ai account, prefer **Cl
```
3. **LLM configured** — two environment variables:
```bash
export STRIX_LLM="openai/gpt-5.4" # any LiteLLM model id (openai/..., anthropic/..., openrouter/...)
export STRIX_LLM="openrouter/z-ai/glm-5.3" # any LiteLLM model id (openai/..., anthropic/..., openrouter/...)
export LLM_API_KEY="<provider api key>"
```
Ask the user for these if unset. Never hardcode or commit keys.
+95 -4
View File
@@ -13,11 +13,13 @@ from agents.agent import ToolsToFinalOutputResult
from agents.sandbox import SandboxAgent
from agents.sandbox.capabilities import Filesystem, Shell
from agents.sandbox.errors import InvalidManifestPathError
from agents.sandbox.session.pty_types import PTY_YIELD_TIME_MS_MAX
from agents.tool import CustomTool, FunctionTool, Tool
from pydantic import ValidationError
from strix.agents.prompt import render_system_prompt
from strix.config import load_settings
from strix.config.tool_call_arguments import describe_malformed_arguments
from strix.tools.agents_graph.tools import (
agent_finish,
create_agent,
@@ -267,6 +269,10 @@ def _with_coerced_arguments(tool: FunctionTool) -> FunctionTool:
nullish = tool.name.startswith(_QUERY_TOOL_PREFIXES)
async def invoke(ctx: Any, raw_input: str) -> Any:
malformed = describe_malformed_arguments(tool.name, raw_input)
if malformed is not None:
logger.debug("Tool %s got malformed arguments; asking the model to re-issue", tool.name)
return malformed
return await invoke_tool(ctx, _coerce_arguments(raw_input, schema, nullish=nullish))
tool.on_invoke_tool = invoke
@@ -375,11 +381,15 @@ def _configure_filesystem_tools(
)
def _make_filesystem_configurator(*, chat_completions: bool, strict_schemas: bool) -> Any:
def _make_filesystem_configurator(
*, chat_completions: bool, strict_schemas: bool, supports_images: bool = True
) -> Any:
def configure(toolset: Any) -> None:
_configure_filesystem_tools(
toolset, chat_completions=chat_completions, strict_schemas=strict_schemas
)
if not supports_images:
toolset.view_image.is_enabled = False
return configure
@@ -432,8 +442,47 @@ def _apply_shell_output_cap(parsed: dict[str, Any]) -> None:
)
_EXEC_YIELD_CAP_S = PTY_YIELD_TIME_MS_MAX // 1000
_EXEC_COMMAND_DESCRIPTION_NOTE = (
f"The wait for output is capped at {_EXEC_YIELD_CAP_S} s whatever yield_time_ms says. "
"A non-TTY command still running at the yield cannot be read later (write_stdin fails "
"on it and its output is lost): run anything longer in the background with its output "
"redirected to a file (`cmd > /tmp/out.log 2>&1 &`, then read the log), or start it with "
'tty=true and poll with write_stdin(session_id=<id>, chars="").'
)
_RUNNING_NON_TTY_NOTICE = (
"Note: this non-TTY process cannot be polled and its later output is lost. Rerun it in "
"the background (`cmd > /tmp/out.log 2>&1 &`, then read the log) or with tty=true."
)
_PROCESS_RUNNING_RE = re.compile(r"^Process running with session ID \d+$", re.MULTILINE)
_OUTPUT_HEADER_RE = re.compile(r"^Output:\n", re.MULTILINE)
def _annotate_running_non_tty(parsed: dict[str, Any], result: Any) -> Any:
if parsed.get("tty") or not isinstance(result, str):
return result
header = _OUTPUT_HEADER_RE.search(result)
head_end = len(result) if header is None else header.start()
if _PROCESS_RUNNING_RE.search(result, 0, head_end) is None:
return result
notice = _RUNNING_NON_TTY_NOTICE
requested = parsed.get("yield_time_ms")
if isinstance(requested, int | float) and requested > PTY_YIELD_TIME_MS_MAX:
notice = (
f"Note: yield_time_ms={int(requested)} was capped at {PTY_YIELD_TIME_MS_MAX} "
f"({_EXEC_YIELD_CAP_S} s). " + notice.removeprefix("Note: ")
)
if header is None:
return f"{result}\n{notice}"
return f"{result[: header.end()]}{notice}\n{result[header.end() :]}"
def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
invoke_tool = tool.on_invoke_tool
tool.description = f"{tool.description.rstrip()} {_EXEC_COMMAND_DESCRIPTION_NOTE}"
async def invoke(ctx: Any, raw_input: str) -> Any:
try:
@@ -446,7 +495,7 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
_apply_shell_output_cap(parsed)
raw_input = json.dumps(parsed)
try:
return await invoke_tool(ctx, raw_input)
result = await invoke_tool(ctx, raw_input)
except ValidationError as exc:
return _format_validation_error(tool.name, exc)
except InvalidManifestPathError as exc:
@@ -456,6 +505,9 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
"(or omitted to use the turn's cwd). "
f"Got: {rel!r}."
)
if isinstance(parsed, dict):
return _annotate_running_non_tty(parsed, result)
return result
tool.on_invoke_tool = invoke
return tool
@@ -567,6 +619,17 @@ def _finish_tool_use_behavior(
return ToolsToFinalOutputResult(is_final_output=False, final_output=None)
_COVERAGE_TOOLS: tuple[Tool, ...] = (record_coverage, update_coverage, list_coverage)
_FINISH_SCAN_COVERAGE_STEP = re.compile(
r"\n5\. \*\*Coverage reconciliation\.\*\*.*?(?=\n\n)", re.DOTALL
)
def _without_coverage_step(tool: FunctionTool) -> FunctionTool:
description = _FINISH_SCAN_COVERAGE_STEP.sub("", tool.description)
return dataclasses.replace(tool, description=description)
_BASE_TOOLS: tuple[Tool, ...] = (
think,
load_skill,
@@ -665,11 +728,16 @@ def build_strix_agent(
is_whitebox: bool = False,
is_diff_scoped: bool = False,
interactive: bool = False,
parallel_tool_calls: bool = False,
chat_completions_tools: bool = False,
strict_tool_schemas: bool = True,
system_prompt_context: dict[str, Any] | None = None,
extra_tools: Sequence[Tool] | None = None,
instructions_override: str | None = None,
supports_images: bool = True,
finish_tool: Tool = finish_scan,
report_tool: Tool = create_vulnerability_report,
coverage_tools: bool = True,
) -> SandboxAgent[Any]:
"""Build a SandboxAgent for either root or child use.
@@ -682,6 +750,11 @@ def build_strix_agent(
registered via ``register_agent_tools``.
instructions_override: Use this verbatim as the system prompt instead
of rendering the built-in scan prompt.
finish_tool: The tool that ends the run, given to the root agent.
report_tool: The tool every agent files findings with, in place of
``create_vulnerability_report``.
coverage_tools: Give the agent the coverage ledger tools and the
prompt rules for them.
"""
if instructions_override is not None:
instructions = instructions_override
@@ -693,17 +766,26 @@ def build_strix_agent(
is_root=is_root,
is_diff_scoped=is_diff_scoped,
interactive=interactive,
parallel_tool_calls=parallel_tool_calls,
system_prompt_context=system_prompt_context,
supports_images=supports_images,
report_tool=report_tool.name,
coverage_tools=coverage_tools,
)
agent_tools = [*_EXTRA_TOOLS, *(extra_tools or [])]
if interactive:
# Yielding to the user is only meaningful when one is attached.
agent_tools.append(wait_for_user)
base_tools = [report_tool if t is create_vulnerability_report else t for t in _BASE_TOOLS]
if not coverage_tools:
base_tools = [t for t in base_tools if t not in _COVERAGE_TOOLS]
if finish_tool is finish_scan:
finish_tool = _without_coverage_step(finish_scan)
if is_root:
tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_scan]
tools: list[Tool] = [*base_tools, *agent_tools, finish_tool]
else:
tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
tools = [*base_tools, *agent_tools, agent_finish]
_ensure_unique_tool_names(tools)
tools = [
_with_bounded_result(_with_strictness(_with_coerced_arguments(tool), strict_tool_schemas))
@@ -733,6 +815,7 @@ def build_strix_agent(
configure_tools=_make_filesystem_configurator(
chat_completions=chat_completions_tools,
strict_schemas=strict_tool_schemas,
supports_images=supports_images,
),
),
Shell(
@@ -751,9 +834,13 @@ def make_child_factory(
is_whitebox: bool = False,
is_diff_scoped: bool = False,
interactive: bool = False,
parallel_tool_calls: bool = False,
chat_completions_tools: bool = False,
strict_tool_schemas: bool = True,
system_prompt_context: dict[str, Any] | None = None,
supports_images: bool = True,
report_tool: Tool = create_vulnerability_report,
coverage_tools: bool = True,
) -> Any:
"""Return the runner-owned builder used by ``spawn_child_agent``.
@@ -771,9 +858,13 @@ def make_child_factory(
is_whitebox=is_whitebox,
is_diff_scoped=is_diff_scoped,
interactive=interactive,
parallel_tool_calls=parallel_tool_calls,
chat_completions_tools=chat_completions_tools,
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=system_prompt_context,
supports_images=supports_images,
report_tool=report_tool,
coverage_tools=coverage_tools,
)
return _factory
+10 -1
View File
@@ -83,8 +83,12 @@ def render_system_prompt(
is_root: bool = False,
is_diff_scoped: bool = False,
interactive: bool = False,
parallel_tool_calls: bool = False,
system_prompt_context: dict[str, Any] | None = None,
include_scope: bool = True,
supports_images: bool = True,
report_tool: str = "create_vulnerability_report",
coverage_tools: bool = True,
) -> str:
"""Render the system prompt. Returns empty string on template failure.
@@ -119,7 +123,9 @@ def render_system_prompt(
is_root=is_root,
is_diff_scoped=is_diff_scoped,
)
skill_content = load_skills(skills_to_load)
skill_content = load_skills(
skills_to_load, supports_images=supports_images, coverage_tools=coverage_tools
)
env.globals["get_skill"] = lambda name: skill_content.get(name, "")
# Skills every agent of this kind loads come first, so siblings share them
@@ -129,9 +135,12 @@ def render_system_prompt(
requested_skill_names=[name for name in skill_content if name not in shared],
available_skills=get_available_skills(),
interactive=interactive,
parallel_tool_calls=parallel_tool_calls,
is_root=is_root,
system_prompt_context=system_prompt_context or {},
include_scope=include_scope,
report_tool=report_tool,
coverage_tools=coverage_tools,
**skill_content,
)
except Exception:
+1 -1
View File
@@ -21,7 +21,7 @@ MCP CONNECTIONS (available this run):
{% if system_prompt_context.mcp_connections %}
- Connected this run (search one to find relevant tools):
{% for connection in system_prompt_context.mcp_connections %}
- {{ connection.name }} ({{ connection.tool_count }} tools){% if connection.purpose %}: {{ connection.purpose }}{% endif %}
- {{ connection.name }}{% if connection.state == "unavailable" %} (unavailable right now){% elif connection.tool_count is not none %} ({{ connection.tool_count }} tools){% endif %}{% if connection.purpose %}: {{ connection.purpose }}{% endif %}
{% endfor %}
{% endif %}
- Reach for a connection whenever the target itself cannot give you information a connection could: its database schema and access policies, real deployment or infrastructure configuration, known issues or prior findings, or server logs. In those cases call list_mcps early to see what is available, and prefer a connection's authoritative data over inferring from the target's responses. Do not wait to be told a connection exists.
+23 -12
View File
@@ -44,7 +44,11 @@ INTERACTIVE BEHAVIOR:
- If you end a turn on plain text and the nudge arrives, your words already reached the user. Do not repeat them: call wait_for_user.
- You can narrate while you work — plain text is shown to the user as you go. Narrating is free; wait_for_user is specifically the act of WAITING for the user, so do not call it just to give a status update.
- Respond naturally when the user asks questions or gives instructions.
{%- if parallel_tool_calls %}
- While actively working on a task, every turn should carry a tool call (independent calls can go in the same turn) — use think to plan, the appropriate tool to act, and wait_for_user only when you genuinely need the user.
{%- else %}
- While actively working on a task, every turn should carry exactly one tool call — use think to plan, the appropriate tool to act, and wait_for_user only when you genuinely need the user.
{%- endif %}
- Never loop through think or other tools just to prepare, polish, confirm, or announce an answer. Once you know the answer, write it and call wait_for_user.
{% else %}
AUTONOMOUS BEHAVIOR:
@@ -55,6 +59,9 @@ AUTONOMOUS BEHAVIOR:
- There is no user attached to this run, so there is nobody to ask and nothing to yield to. If there is nothing left to execute: do NOT send filler/repetitive text — either call wait_for_agents (only if you are genuinely expecting another agent to message you) or finish your work (subagents: agent_finish; root: finish_scan)
- While the agent loop is running, almost every output MUST be a tool call. Do NOT send plain text messages; act via tools. If waiting on another agent, use wait_for_agents; when done, use agent_finish (subagents) or finish_scan (root)
- A text-only turn does nothing: it neither ends the run nor yields — it just wastes a turn and forces a retry. The lifecycle tools (``finish_scan`` for root, ``agent_finish`` for subagents) are the ONLY way to terminate, and the report flows through them. If you find yourself wanting to say "Done!" or "Scan complete" without a tool call, call the lifecycle tool instead.
{%- if parallel_tool_calls %}
- Read-only calls can go together in one turn: when several file reads or searches don't depend on each other's output, issue them as separate tool calls in the same turn rather than one per turn. Anything else — writing files, running commands that change state, sending requests, creating or messaging agents, reporting, or finishing — goes alone in its own turn.
{%- endif %}
{% endif %}
</communication_rules>
@@ -108,18 +115,18 @@ BLACK-BOX TESTING (domain/subdomain only):
WHITE-BOX TESTING (code provided):
- MUST perform BOTH static AND dynamic analysis
- Static: Use source-aware triage first to map risk quickly (`semgrep`, `ast-grep`, Tree-sitter tooling, `gitleaks`, `trufflehog`, `trivy fs`). Then review code for vulnerabilities
- Static: Use source-aware triage first to map risk quickly (`opengrep`, `ast-grep`, Tree-sitter tooling, `gitleaks`, `trufflehog`, `trivy fs`). Then review code for vulnerabilities
- Static coverage floor: execute at least one structural AST mapping pass (`sg` and/or Tree-sitter) per repository and keep artifact output
- Static coverage target per repository: run one `semgrep` pass, one secrets pass (`gitleaks` and/or `trufflehog`), one `trivy fs` pass, and one AST-structural pass (`sg` and/or Tree-sitter)
- Static coverage target per repository: run one `opengrep` pass, one secrets pass (`gitleaks` and/or `trufflehog`), one `trivy fs` pass, and one AST-structural pass (`sg` and/or Tree-sitter)
- Keep AST artifacts bounded and high-signal: scope to relevant paths/hypotheses, avoid whole-repo generic function dumps
- AST target selection rule: build `sg-targets.txt` from `semgrep.json` scope first (`paths.scanned`, fallback to unique `results[].path`), then run `xargs ... sg run` against that file list. Only use path-heuristic fallback if semgrep scope is unavailable.
- AST target selection rule: build `sg-targets.txt` from `opengrep.json` scope first (`paths.scanned`, fallback to unique `results[].path`), then run `xargs ... sg run` against that file list. Only use path-heuristic fallback if opengrep scope is unavailable.
- Dynamic: Run the application and test live to validate exploitability
- NEVER rely solely on static code analysis when dynamic validation is possible
- Begin with fast source triage and dynamic run preparation in parallel; use static findings to prioritize live testing.
- Local execution, unit/integration testing, patch verification, and HTTP requests against locally started in-scope services are normal authorized white-box validation
- If dynamically running the code proves impossible after exhaustive attempts, pivot to comprehensive static analysis.
- Try to infer how to run the code based on its structure and content.
- Derive the code fix as PART OF reporting, not as a separate later pass: create_vulnerability_report already requires the concrete patch inline (`code_locations` with verbatim `fix_before`/`fix_after` and `fix_pr_body`), so the reporting agent that analyzes the root cause is the one that produces the fix. Do NOT spawn a downstream agent afterwards to re-derive/re-apply the same patch.
- Derive the code fix as PART OF reporting, not as a separate later pass: {{ report_tool }} already requires the concrete patch inline (`code_locations` with verbatim `fix_before`/`fix_after` and `fix_pr_body`), so the reporting agent that analyzes the root cause is the one that produces the fix. Do NOT spawn a downstream agent afterwards to re-derive/re-apply the same patch.
- If you also apply and verify the patch in the repo (edit the file, re-test that the vulnerability is gone), do it in the same agent/turn while the analysis is fresh — right before or as part of filing the report — never as a second re-analysis pass.
COMBINED MODE (code + deployed target present):
@@ -181,7 +188,7 @@ EFFICIENCY TACTICS:
- 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`
- Prefer established fuzzers/scanners where applicable: ffuf, sqlmap, nuclei, wapiti, arjun, httpx, katana, semgrep, bandit, trufflehog, nmap. Use scripts mainly to coordinate or validate around them, not to replace them without reason
- Prefer established fuzzers/scanners where applicable: ffuf, sqlmap, nuclei, wapiti, arjun, httpx, katana, opengrep, bandit, trufflehog, nmap. Use scripts mainly to coordinate or validate around them, not to replace them without reason
- For trial-heavy vectors (SQLi, XSS, XXE, SSRF, RCE, auth/JWT, deserialization), DO NOT iterate payloads manually in the browser. Always spray payloads via Python scripts through `exec_command` or terminal tools.
- When using established fuzzers/scanners, use the proxy for inspection where helpful
- Generate/adapt large payload corpora: combine encodings (URL, unicode, base64), comment styles, wrappers, time-based/differential probes. Expand with wordlists/templates
@@ -202,12 +209,14 @@ VALIDATION REQUIREMENTS:
- Keep going until you find something that matters
- CLOSURE DISCIPLINE: every candidate you open ends in exactly one explicit state — `confirmed` (working PoC, or a complete source→control→sink→impact trace that is reachable), `ruled_out` (you can name the SPECIFIC control, at a location, that runs on every attacker-reachable path before the sink), or `open_proof_gap` (plausible, unconfirmed, and you could NOT name such a control). "I moved on" is not a closure state. Silently dropping an uncertain candidate is mislabelling an `open_proof_gap` as `ruled_out` and is how real bugs get missed.
- Missing information is NOT proof of safety: no caller found, can't tell if deployed/exposed, couldn't stand up the service, build failed — each is an `open_proof_gap`, never a reason to mark a candidate clean. Difficulty is a reason to defer, not to suppress.
- COVERAGE: record every surface you assess with `record_coverage` (surface + risk area + outcome + evidence), including the ones that came back clean — a report that only lists findings cannot say what was reviewed and cleared. Use the `needs_follow_up` outcome for anything left in an `open_proof_gap` state, and carry the same items up in `agent_finish(open_items=[...])`. The ledger is shared and mutable: when you resolve a surface another agent left open — or find that a closed one is not — move that entry with `update_coverage` instead of recording a second one for the same surface. The root agent reconciles all of it via `list_coverage` before `finish_scan`.
{% if coverage_tools -%}
- COVERAGE: record every surface you assess with `record_coverage` (surface + risk area + outcome + evidence), including the ones that came back clean — a report that only lists findings cannot say what was reviewed and cleared. Use the `needs_follow_up` outcome for anything left in an `open_proof_gap` state, and carry the same items up in `agent_finish(open_items=[...])`. The ledger is shared and mutable: when you resolve a surface another agent left open — or find that a closed one is not — move that entry with `update_coverage` instead of recording a second one for the same surface. Before `finish_scan`, the root agent calls `list_coverage(outcome="needs_follow_up")` and follows up every row it returns: it assigns the outstanding work to a subagent that moves the existing entry with `update_coverage` to its real outcome, and a row stays `needs_follow_up` only when it is genuinely blocked, with the blocker stated in its evidence.
{% endif -%}
- THREAT MODEL: before you start testing, call `get_threat_model` on the target you were pointed at — it is the scan's shared answer to who the attacker is, where the trust boundaries sit, and what counts as critical here. It is scoped to this scan and nothing carries over from an earlier run, so `found: false` means no agent on this run has derived one yet. Read it instead of re-deriving trust boundaries yourself; where your testing disproves it — a boundary it calls trusted turns out to be attacker-reachable, a role it did not know about, a host or endpoint it never listed — record that with `amend_threat_model` so the agents after you inherit the correction. Amending is not optional politeness: a model nobody corrects turns the first agent's guesses into everyone's assumptions.
- Before filing any report, run the counterevidence pass: argue the strongest case AGAINST the finding, record what you found in the `counterevidence` field, set `confidence` honestly (a static-only trace you couldn't execute is at best `medium`), and state what evidence would change the severity. See the counterevidence and severity-calibration knowledge above.
- 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
- Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via create_vulnerability_report (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.)
- 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. If your evidence proves more than the finding it matched (a working exploit where that one had only a static trace, a chain that raises the impact), revise that finding with update_vulnerability_report using the duplicate_of id — never re-file it.
- A vulnerability is ONLY considered reported when a reporting agent uses {{ report_tool }} (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
- Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via {{ report_tool }} (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.)
- DEDUPLICATION: The {{ report_tool }} 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. If your evidence proves more than the finding it matched (a working exploit where that one had only a static trace, a chain that raises the impact), revise that finding with update_vulnerability_report using the duplicate_of id — never re-file it.
- HTTP EVIDENCE: a finding you validated through the proxy is not fully filed until `http_exchange_ids` carries the proxy request ids of the exchanges that prove it — the request that triggers the vulnerability plus the baseline/control request it differs from (an unauthenticated success next to the authenticated one, the payload response next to the benign one). Copy the ids exactly as `list_requests`/`view_request` show them, never invent or guess one, and never omit the field to bypass validation. Leave it out only when there is no captured HTTP exchange at all (static-only code findings, dependency CVEs). If you filed before the proving exchanges existed, attach them afterwards with update_vulnerability_report. Without the ids, the finding ships as prose nobody can replay.
- REVISING A FINDING: use update_vulnerability_report (report id + the fields you want to replace + update_reason) when you learn something a finding already on file does not carry — you built the PoC after filing it, a chain raised its impact, further testing weakened it, or its counterevidence/remediation/code locations were wrong. Editing a finding needs no duplicate verdict, and it is always better than filing a second report for the same issue. Read the finding first with get_report, and pass only the fields that change.
- WITHDRAWING A FINDING: when re-testing proves a finding you filed is not a vulnerability at all — the exploit only worked because of your own test setup (a mixed-up session, a self-inflicted state change, a misread response), the behaviour is intended, or the control is in fact enforced — delete it with delete_vulnerability_report (report id + delete_reason). Never leave a disproved finding on file as a zero-CVSS, "retracted", or "false positive" report, and never rewrite its text into a note saying it was wrong: a report that exists is read as a finding. Delete is for a finding that does not exist; a finding that is real but weaker than filed is revised, not deleted.
@@ -220,14 +229,16 @@ Every one of these tools writes to state the rest of the scan reads. Reaching fo
- TODOS — `create_todo` / `list_todos` / `update_todo` / `mark_todo_done` / `mark_todo_pending` / `delete_todo`: your own working checklist for a multi-step task. Create todos when your task has several distinct steps so nothing is dropped across a long run; mark them done as you finish. This is private working memory — use `notes` for anything another agent needs.
- NOTES — `create_note` / `list_notes` / `get_note` / `update_note` / `delete_note`: the scan's shared scratchpad, visible to every agent. Write a note for a durable cross-agent fact that is not a finding and not coverage — a working credential set, a discovered endpoint inventory, an enumerated tenant list, a rate-limit quirk the next agent needs. `update_note` to keep a living inventory current; `delete_note` only for something now wrong or superseded. Check `list_notes`/`get_note` before recon work so you build on what is already mapped instead of redoing it.
- THREAT MODEL — `get_threat_model` / `amend_threat_model` / `save_threat_model`: covered above. `save_threat_model` REPLACES the whole document and clears amendments, so it is for establishing the baseline or folding amendments in (normally root) — to correct part of an existing model, `amend_threat_model` instead.
{% if coverage_tools -%}
- COVERAGE — `record_coverage` / `update_coverage` / `list_coverage`: covered above. One row per surface+risk; correct an existing row with `update_coverage`, never a second `record_coverage`.
{% endif -%}
- RESEARCH — `web_search`: pull fresh, target-specific external knowledge — latest bypasses, WAF evasions, DB-/framework-specific syntax, CVE and advisory detail — before falling back to memorized payloads, and refresh payload corpora mid-spray.
- SPAWN WORK — `create_agent`: delegate a focused subtask to a specialist child (see the multi-agent rules below for when to spawn and how to scope it). Give it the target to model against and what is already known.
- TRACK CHILDREN — `view_agent_graph`: your live map of every agent and its status. Call it before spawning (to confirm no existing agent already covers the scope) and before finishing (to confirm no child is still running).
- STEER CHILDREN — `send_message_to_agent`: send a running child new information, a course correction, or a request to wrap up, without killing it. Use it to answer a child's question or narrow its scope mid-run.
- BLOCK ON CHILDREN — `wait_for_agents`: block until named children report back when your next move genuinely depends on their results. If you can keep making progress in parallel, keep working instead of waiting.
- CANCEL CHILDREN — `stop_agent`: gracefully cancel a child whose work is redundant, misdirected, or no longer needed. Prefer `send_message_to_agent` to redirect a child that is merely off-track; reserve `stop_agent` for work that should not continue at all.
- FINISH — subagents call `agent_finish` (with `open_items=[...]` for anything left unresolved); the root agent calls `finish_scan` exactly once, only after every child is wrapped up and coverage is reconciled. `agent_finish`/`finish_scan` are handoffs, not reporting channels — a vulnerability is reported only via `create_vulnerability_report`/`create_dependency_report`.
- FINISH — subagents call `agent_finish` (with `open_items=[...]` for anything left unresolved); the root agent calls `finish_scan` exactly once, only after every child is wrapped up and coverage is reconciled. `agent_finish`/`finish_scan` are handoffs, not reporting channels — a vulnerability is reported only via `{{ report_tool }}`/`create_dependency_report`.
</execution_guidelines>
<vulnerability_focus>
@@ -380,7 +391,7 @@ CRITICAL RULES:
- **REALISTIC OUTCOMES** - Some tests find nothing, some validations fail
- **ONE AGENT = ONE TASK** - Don't let agents do multiple unrelated jobs
- **SPAWN REACTIVELY** - Create new agents based on what you discover
- **ONLY REPORTING AGENTS** can use create_vulnerability_report tool
- **ONLY REPORTING AGENTS** can use {{ report_tool }} tool
- **AGENT SPECIALIZATION MANDATORY** - Each agent must be highly specialized; prefer 1–3 skills, up to 5 for complex contexts
- **NO GENERIC AGENTS** - Avoid creating broad, multi-purpose agents that dilute focus
@@ -447,7 +458,7 @@ JAVASCRIPT ANALYSIS:
- js-beautify - JS beautifier/deobfuscator
CODE ANALYSIS:
- semgrep - Static analysis/SAST
- opengrep - Static analysis/SAST (bundled rules in /opt/opengrep-rules)
- ast-grep (sg) - Structural AST/CST-aware code search
- tree-sitter - Syntax-aware parsing and symbol extraction support
- bandit - Python security linter
+26 -2
View File
@@ -6,7 +6,8 @@ import json
import logging
import os
from pathlib import Path
from typing import TYPE_CHECKING, Any
from types import UnionType
from typing import TYPE_CHECKING, Any, Union, get_args, get_origin
from pydantic import AliasChoices, BaseModel
@@ -126,13 +127,36 @@ def _read_json_overrides(path: Path) -> dict[str, dict[str, Any]]:
continue # env wins under some alias; skip the JSON file for this field
for alias in aliases:
if alias in env_block_upper:
sub_data[fname] = env_block_upper[alias]
sub_data[fname] = _decode_saved_value(finfo, env_block_upper[alias])
break
if sub_data:
nested[sub_name] = sub_data
return nested
def _decode_saved_value(finfo: FieldInfo, value: Any) -> Any:
"""Decode a JSON string saved for a dict field, as pydantic does for env vars.
``persist_current`` saves env values verbatim, so a dict setting such as
``LLM_EXTRA_HEADERS`` lands in the file as JSON text. Values that are
already objects, or aren't valid JSON, are left for validation to judge.
"""
if not isinstance(value, str) or not _is_dict_field(finfo):
return value
try:
return json.loads(value)
except ValueError:
return value
def _is_dict_field(finfo: FieldInfo) -> bool:
annotation = finfo.annotation
candidates = (
get_args(annotation) if get_origin(annotation) in (Union, UnionType) else (annotation,)
)
return any(c is dict or get_origin(c) is dict for c in candidates)
def _first_alias_value(aliases: list[str], source: Mapping[str, Any]) -> Any | None:
return next((source[alias] for alias in aliases if alias in source), None)
+140 -30
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import asyncio
import contextlib
import inspect
import json
import logging
import os
import time
@@ -40,6 +41,7 @@ from openai.types.shared import Reasoning
from strix.agents.prompt import CACHE_POINT
from strix.config import codex
from strix.config.loader import load_settings
from strix.config.tool_call_arguments import repair_input
from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input
from strix.config.tool_call_limits import TurnToolCallLimiter
from strix.llm import request_log
@@ -254,6 +256,12 @@ class _TurnGuardModel(Model):
Ids that collide with the history are rewritten before the turn is
recorded, and already-corrupted histories are repaired on the way out.
Tool-call arguments: a turn whose ``arguments`` are not valid JSON fails
that one call with a parse error, but the raw string is recorded and
strict providers then reject every request that replays it. Such
arguments are rewritten to a valid JSON object on the way out (see
:mod:`strix.config.tool_call_arguments`).
Tool-call volume: a degenerate response can queue hundreds of calls that
the run loop then honours one by one. Only the first
``LLM_MAX_TOOL_CALLS_PER_TURN`` calls of a response are kept.
@@ -262,6 +270,8 @@ class _TurnGuardModel(Model):
not covered by the request timeout, which resets on any byte (keepalives
included). ``LLM_STREAM_IDLE_TIMEOUT`` bounds the gap between events so the
turn fails instead of hanging, and the existing retry path replays it.
``LLM_STREAM_FIRST_EVENT_TIMEOUT`` and ``LLM_STREAM_TOTAL_TIMEOUT`` bound
the first event and the whole stream.
"""
def __init__(
@@ -269,11 +279,15 @@ class _TurnGuardModel(Model):
inner: Model,
*,
max_tool_calls_per_turn: int = 0,
stream_idle_timeout: float = 0.0,
stream_idle_timeout: float | None = None,
first_event_timeout: float | None = None,
total_timeout: float | None = None,
) -> None:
self._inner = inner
self._max_tool_calls_per_turn = max_tool_calls_per_turn
self._stream_idle_timeout = stream_idle_timeout
self._first_event_timeout = first_event_timeout
self._total_timeout = total_timeout
def _limiter(self) -> TurnToolCallLimiter:
return TurnToolCallLimiter(self._max_tool_calls_per_turn)
@@ -306,7 +320,7 @@ class _TurnGuardModel(Model):
conversation_id: str | None,
prompt: ResponsePromptParam | None,
) -> ModelResponse:
sanitized = dedupe_input(input)
sanitized = _sanitize_input(input)
rewriter = TurnCallIdRewriter(sanitized)
system_instructions, sanitized = _split_cached_prefix(
system_instructions, sanitized, model_settings
@@ -342,7 +356,7 @@ class _TurnGuardModel(Model):
conversation_id: str | None,
prompt: ResponsePromptParam | None,
) -> AsyncIterator[TResponseStreamEvent]:
sanitized = dedupe_input(input)
sanitized = _sanitize_input(input)
rewriter = TurnCallIdRewriter(sanitized)
system_instructions, sanitized = _split_cached_prefix(
system_instructions, sanitized, model_settings
@@ -360,13 +374,22 @@ class _TurnGuardModel(Model):
conversation_id=conversation_id,
prompt=prompt,
)
async for event in _with_idle_timeout(stream, self._stream_idle_timeout):
async for event in _with_timeouts(
stream,
idle=self._stream_idle_timeout,
first_event=self._first_event_timeout,
total=self._total_timeout,
):
guarded = _guard_event(event, rewriter, limiter)
if guarded is not None:
yield guarded
self._log_dropped(limiter)
def _sanitize_input(model_input: str | list[TResponseInputItem]) -> str | list[Any]:
return repair_input(dedupe_input(model_input))
def _split_cached_prefix(
system_instructions: str | None,
model_input: str | list[Any],
@@ -394,28 +417,68 @@ async def _aclose(stream: AsyncIterator[TResponseStreamEvent]) -> None:
await stream.aclose()
async def _with_idle_timeout(
stream: AsyncIterator[TResponseStreamEvent], timeout: float
async def _with_timeouts(
stream: AsyncIterator[TResponseStreamEvent],
*,
idle: float | None = None,
first_event: float | None = None,
total: float | None = None,
) -> AsyncIterator[TResponseStreamEvent]:
if timeout <= 0:
async for event in stream:
yield event
return
"""Bound the first event, the gap between events and the whole stream; None is no bound."""
iterator = stream.__aiter__()
started = time.monotonic()
while True:
limits: list[tuple[float, float, str]] = []
# Wait until the soonest limit expires: first-event (else idle) until the
# first event arrives, then idle; plus whatever is left of the total.
if first_event is not None:
limits.append((first_event, first_event, "stream_first_event_timeout"))
elif idle is not None:
limits.append((idle, idle, "stream_idle_timeout"))
if total is not None:
left = max(0.0, started + total - time.monotonic())
limits.append((left, total, "stream_total_timeout"))
try:
event = await asyncio.wait_for(iterator.__anext__(), timeout)
if limits:
event = await _next_event(iterator, *min(limits))
else:
event = await iterator.__anext__()
except StopAsyncIteration:
return
except TimeoutError:
except TimeoutError as exc:
await _aclose(stream)
message = f"model stream produced no event for {timeout:.0f}s"
logger.warning("%s; abandoning the turn", message)
raise TimeoutError(message) from None
logger.warning("%s; abandoning the turn", exc)
raise
first_event = None
yield event
async def _next_event(
iterator: AsyncIterator[TResponseStreamEvent], wait: float, limit: float, reason: str
) -> TResponseStreamEvent:
# asyncio.timeout() cancels without a message, so the request log could not
# tell this from a shutdown; this is asyncio.timeout() with a message.
task = asyncio.current_task()
assert task is not None
cancelling = task.cancelling()
expired = False
def expire() -> None:
nonlocal expired
expired = True
task.cancel(msg=f"{request_log.CANCEL_REASON_PREFIX}{reason}")
handle = asyncio.get_running_loop().call_later(wait, expire)
try:
return await iterator.__anext__()
except BaseException as exc:
if expired and task.uncancel() <= cancelling and isinstance(exc, asyncio.CancelledError):
raise TimeoutError(f"model stream hit {reason} ({limit:.0f}s)") from None
raise
finally:
handle.cancel()
def _guard_event(
event: TResponseStreamEvent, rewriter: TurnCallIdRewriter, limiter: TurnToolCallLimiter
) -> TResponseStreamEvent | None:
@@ -550,7 +613,10 @@ class StrixProvider(MultiProvider):
def get_model(self, model_name: str | None) -> Model:
llm = load_settings().llm
slug = codex.subscription_model(model_name)
idle_timeout = float(llm.stream_idle_timeout)
# The settings use 0 for no bound.
idle_timeout = float(llm.stream_idle_timeout) or None
first_event_timeout = float(llm.stream_first_event_timeout) or None
total_timeout = float(llm.stream_total_timeout) or None
if slug:
# The ChatGPT subscription backend is always streamed; it has no
# non-streaming mode to fall back to, so LLM_DISABLE_STREAMING
@@ -591,11 +657,13 @@ class StrixProvider(MultiProvider):
# The wrapper emits its single event only once the whole request
# is done, so an idle gap is meaningless here; the request
# timeout bounds it instead.
idle_timeout = 0.0
idle_timeout = first_event_timeout = total_timeout = None
return _TurnGuardModel(
model,
max_tool_calls_per_turn=llm.max_tool_calls_per_turn,
stream_idle_timeout=idle_timeout,
first_event_timeout=first_event_timeout,
total_timeout=total_timeout,
)
@@ -637,11 +705,15 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
"""Apply Strix config to SDK-native defaults."""
llm = settings.llm
set_tracing_disabled(True)
from strix.report.state import record_llm_request
request_log.install()
if settings.telemetry.run_record_timings:
request_log.register_sink(record_llm_request)
if codex.subscription_model(llm.model):
return
_configure_litellm_compatibility()
_configure_openrouter_attribution(llm.model)
_configure_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)
@@ -745,6 +817,11 @@ def _install_openrouter_stream_cost_capture() -> None:
class _StrixOpenRouterStreamingHandler(OpenRouterChatCompletionStreamingHandler):
def chunk_parser(self, chunk: dict[str, Any]) -> Any:
# Before parsing: LiteLLM raises on an error chunk and drops its
# top-level ``provider``.
request_log.record_upstream_provider(
chunk.get("provider"), _openrouter_error_type(chunk.get("error"))
)
stream = super().chunk_parser(chunk)
usage = chunk.get("usage")
response_id = chunk.get("id") or getattr(stream, "id", None)
@@ -763,12 +840,22 @@ def _install_openrouter_stream_cost_capture() -> None:
json_mode=json_mode,
)
def get_error_class(self, error_message: str, status_code: int, headers: Any) -> Any:
# A non-2xx reply names the provider in ``error.metadata``.
with contextlib.suppress(Exception):
error = json.loads(error_message)["error"]
request_log.record_upstream_provider(
error["metadata"].get("provider_name"), _openrouter_error_type(error)
)
return super().get_error_class(error_message, status_code, headers)
def transform_response(self, *args: Any, **kwargs: Any) -> Any:
# Non-streamed replies (LLM_DISABLE_STREAMING) skip the chunk parser.
response = super().transform_response(*args, **kwargs)
raw_response = kwargs.get("raw_response", args[1] if len(args) > 1 else None)
with contextlib.suppress(Exception):
body = raw_response.json() # type: ignore[union-attr]
request_log.record_upstream_provider(body.get("provider"))
if body.get("usage"):
record_openrouter_provider(body.get("provider"), body["usage"])
return response
@@ -789,9 +876,21 @@ def _install_openrouter_stream_cost_capture() -> None:
litellm.OpenrouterConfig = _StrixOpenrouterConfig # type: ignore[misc]
OPENROUTER_ATTRIBUTION_HEADERS = {
def _openrouter_error_type(error: object) -> object:
if isinstance(error, dict):
metadata = error.get("metadata")
if isinstance(metadata, dict):
return metadata.get("error_type")
return None
APP_ATTRIBUTION_HEADERS = {
"HTTP-Referer": "https://strix.ai",
"X-Title": "Strix",
}
OPENROUTER_ATTRIBUTION_HEADERS = {
**APP_ATTRIBUTION_HEADERS,
"X-OpenRouter-Categories": "cli-agent",
}
@@ -800,20 +899,25 @@ def is_openrouter_model(model_name: str | None) -> bool:
return bool(model_name) and "openrouter/" in (model_name or "").strip().lower()
def _configure_openrouter_attribution(model_name: str | None) -> None:
def is_vercel_ai_gateway_model(model_name: str | None) -> bool:
return (model_name or "").strip().lower().startswith("vercel_ai_gateway/")
def attribution_headers(model_name: str | None) -> dict[str, str]:
if is_openrouter_model(model_name):
return OPENROUTER_ATTRIBUTION_HEADERS
if is_vercel_ai_gateway_model(model_name):
return APP_ATTRIBUTION_HEADERS
return {}
def _configure_attribution(model_name: str | None) -> None:
import litellm
current: object = litellm.headers
existing: dict[str, str] = current if isinstance(current, dict) else {}
if not is_openrouter_model(model_name):
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]
remaining = {k: v for k, v in existing.items() if k not in OPENROUTER_ATTRIBUTION_HEADERS}
litellm.headers = {**remaining, **attribution_headers(model_name)} or None # type: ignore[assignment]
def _configure_extra_headers(llm: LlmSettings) -> None:
@@ -905,6 +1009,12 @@ def model_supports_reasoning(model_name: str) -> bool:
return bool(entry and entry.get("supports_reasoning"))
def model_supports_images(model_name: str) -> bool:
"""Return whether the model accepts image input. Assume yes until proven otherwise."""
entry = _catalog_entry(model_name)
return entry is None or bool(entry.get("supports_vision"))
def _bare_openai_name(model_name: str) -> str:
name = model_name.strip().lower()
for prefix in ("litellm/", "any-llm/", "openai/"):
+30 -2
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from typing import Literal
from typing import Any, Literal
from pydantic import AliasChoices, Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -55,11 +55,29 @@ class LlmSettings(BaseSettings):
alias="LLM_EXTRA_HEADERS",
repr=False,
)
extra_body: dict[str, Any] | None = Field(
default=None,
alias="LLM_EXTRA_BODY",
repr=False,
)
reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
input_cost_per_mtok: float | None = Field(
default=None, ge=0, allow_inf_nan=False, alias="STRIX_LLM_INPUT_COST_PER_MTOK"
)
output_cost_per_mtok: float | None = Field(
default=None, ge=0, allow_inf_nan=False, alias="STRIX_LLM_OUTPUT_COST_PER_MTOK"
)
cached_input_cost_per_mtok: float | None = Field(
default=None, ge=0, allow_inf_nan=False, alias="STRIX_LLM_CACHED_INPUT_COST_PER_MTOK"
)
force_required_tool_choice: bool = Field(
default=False,
alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
)
parallel_tool_calls: bool = Field(
default=False,
alias="STRIX_PARALLEL_TOOL_CALLS",
)
prompt_cache: bool = Field(
default=True,
alias="STRIX_PROMPT_CACHE",
@@ -79,6 +97,10 @@ class LlmSettings(BaseSettings):
timeout: int = Field(default=300, alias="LLM_TIMEOUT")
preflight_timeout: int = Field(default=30, ge=1, alias="LLM_PREFLIGHT_TIMEOUT")
stream_idle_timeout: int = Field(default=300, ge=0, alias="LLM_STREAM_IDLE_TIMEOUT")
stream_first_event_timeout: int = Field(
default=120, ge=0, alias="LLM_STREAM_FIRST_EVENT_TIMEOUT"
)
stream_total_timeout: int = Field(default=600, ge=0, alias="LLM_STREAM_TOTAL_TIMEOUT")
max_tool_calls_per_turn: int = Field(
default=32,
ge=0,
@@ -106,6 +128,11 @@ class DedupeSettings(BaseSettings):
alias="DEDUPE_LLM_EXTRA_HEADERS",
repr=False,
)
extra_body: dict[str, Any] | None = Field(
default=None,
alias="DEDUPE_LLM_EXTRA_BODY",
repr=False,
)
@field_validator("reasoning_effort", mode="before")
@classmethod
@@ -137,7 +164,7 @@ class RuntimeSettings(BaseSettings):
model_config = _BASE_CONFIG
image: str = Field(
default="ghcr.io/usestrix/strix-sandbox:1.3.0",
default="ghcr.io/usestrix/strix-sandbox:1.4.0",
alias="STRIX_IMAGE",
)
backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
@@ -149,6 +176,7 @@ class TelemetrySettings(BaseSettings):
model_config = _BASE_CONFIG
enabled: bool = Field(default=True, alias="STRIX_TELEMETRY")
run_record_timings: bool = Field(default=False, alias="STRIX_RUN_RECORD_TIMINGS")
WebSearchProvider = Literal["auto", "perplexity", "exa"]
+100
View File
@@ -0,0 +1,100 @@
"""Keep replayed tool-call arguments valid JSON.
A model occasionally emits a tool call whose ``arguments`` string is not a
JSON object (truncated, unbalanced, empty). The call itself fails safely: the
tool reports the parse error and the run continues. But the raw string is
recorded in the session as-is, and strict OpenAI-compatible servers (vLLM,
SGLang, ...) validate every assistant tool call in the request, so from then
on each turn is rejected with ``Assistant tool call function.arguments must
be valid JSON`` and the agent can never recover. Rewriting the replayed
arguments to a JSON object that carries the original text keeps the history
valid while the model still sees what it sent.
"""
from __future__ import annotations
import json
from typing import Any
from openai.types.responses import ResponseFunctionToolCall
MALFORMED_ARGUMENTS_KEY = "malformed_arguments"
def _reject_constant(name: str) -> Any:
raise ValueError(f"non-standard JSON constant {name}")
def _loads_strict(text: str) -> Any:
try:
return json.loads(text, parse_constant=_reject_constant)
except RecursionError as exc:
raise ValueError("arguments are nested too deeply to parse") from exc
def describe_malformed_arguments(tool_name: str, arguments: str) -> str | None:
"""Return a model-facing recovery message if ``arguments`` is not a JSON object.
A tool call whose arguments do not parse is almost always one the stream cut
off (the server flushed a prefix of the JSON, or the client closed early),
so instead of the SDK's generic parse-error result the model is told the
call never ran and must be re-issued whole.
"""
if not arguments.strip():
return None
try:
parsed = _loads_strict(arguments)
except ValueError as exc:
detail = str(exc)
else:
if isinstance(parsed, dict):
return None
detail = f"expected a JSON object, got {type(parsed).__name__}"
return (
f"{tool_name}: the tool call was not executed because its arguments were "
f"truncated or otherwise not valid JSON ({detail}). The response was likely "
"cut off mid-call. Re-issue the call with complete, valid JSON arguments."
)
def repair_arguments(arguments: object) -> str | None:
"""Return replacement arguments a strict server accepts, or ``None`` if already valid."""
if not isinstance(arguments, str) or not arguments.strip():
return "{}"
try:
parsed = _loads_strict(arguments)
except ValueError:
parsed = None
if isinstance(parsed, dict):
return None
return json.dumps({MALFORMED_ARGUMENTS_KEY: arguments}, ensure_ascii=False)
def repair_history_arguments(items: list[Any]) -> tuple[list[Any], bool]:
"""Rewrite function calls in a conversation history whose arguments are not a JSON object."""
rebuilt: list[Any] = []
changed = False
for item in items:
if isinstance(item, dict):
if item.get("type") == "function_call":
repaired = repair_arguments(item.get("arguments"))
if repaired is not None:
item = {**item, "arguments": repaired} # noqa: PLW2901
changed = True
elif isinstance(item, ResponseFunctionToolCall):
repaired = repair_arguments(item.arguments)
if repaired is not None:
item = item.model_copy(update={"arguments": repaired}) # noqa: PLW2901
changed = True
rebuilt.append(item)
return rebuilt, changed
def repair_input(model_input: str | list[Any]) -> str | list[Any]:
if isinstance(model_input, str):
return model_input
rebuilt, changed = repair_history_arguments(model_input)
return rebuilt if changed else model_input
+20 -3
View File
@@ -71,9 +71,11 @@ class AgentCoordinator:
self._lock = asyncio.Lock()
self._snapshot_path: Path | None = None
self.is_shutting_down = False
self.root_finish_tool = "finish_scan"
self._budget_stopped = False
self._reserve_stopped = False
self._budget_paused = False
self._operator_paused = False
self._resume_epoch = 0
self._budget_policy: BudgetPolicy = "stop"
self._extend_budget: Callable[[], None] | None = None
@@ -104,6 +106,10 @@ class AgentCoordinator:
def budget_paused(self) -> bool:
return self._budget_paused
@property
def operator_paused(self) -> bool:
return self._operator_paused
@property
def resume_epoch(self) -> int:
"""Bumped by every ``resume_budget``; an agent parks against the value it read."""
@@ -149,6 +155,7 @@ class AgentCoordinator:
"""
async with self._lock:
self._budget_paused = True
self._operator_paused = True
logger.info("scan paused by the operator")
await self._maybe_snapshot()
@@ -164,6 +171,7 @@ class AgentCoordinator:
self._set_budget_limit(max_budget_usd)
async with self._lock:
self._budget_paused = False
self._operator_paused = False
self._resume_epoch += 1
woken = [aid for aid, status in self.statuses.items() if status == "budget_paused"]
for aid in woken:
@@ -231,7 +239,7 @@ class AgentCoordinator:
self._budget_stopped = budget_stopped
self._reserve_stopped = reserve_stopped
if not budget_paused:
self._budget_paused = False
self._budget_paused = self._operator_paused
for aid, status in self.statuses.items():
if status == "budget_paused":
self.statuses[aid] = "waiting"
@@ -423,7 +431,12 @@ class AgentCoordinator:
unknown, or it is terminal and its loop does not park for wake-ups.
"""
from_user = message.get("from") == "user"
if from_user and self._budget_paused and self._budget_policy != "pause":
if (
from_user
and self._budget_paused
and not self._operator_paused
and self._budget_policy != "pause"
):
await self.resume_from_budget_pause(exclude=target_agent_id)
async with self._lock:
if target_agent_id not in self.statuses:
@@ -446,7 +459,9 @@ class AgentCoordinator:
self.recovery_counts.pop(target_agent_id, None)
self.idle_resume_counts.pop(target_agent_id, None)
self._parent_notified.discard(target_agent_id)
self.statuses[target_agent_id] = "waiting"
held = self._operator_paused and self.statuses[target_agent_id] == "budget_paused"
if not held:
self.statuses[target_agent_id] = "waiting"
runtime.wake.set()
stream = runtime.stream
interrupt_on_message = runtime.interrupt_on_message
@@ -639,6 +654,7 @@ class AgentCoordinator:
"budget_stopped": self._budget_stopped,
"reserve_stopped": self._reserve_stopped,
"budget_paused": self._budget_paused,
"operator_paused": self._operator_paused,
}
async def restore(self, snap: dict[str, Any]) -> None:
@@ -661,6 +677,7 @@ class AgentCoordinator:
self._budget_stopped = bool(snap.get("budget_stopped", False))
self._reserve_stopped = bool(snap.get("reserve_stopped", False))
self._budget_paused = bool(snap.get("budget_paused", False))
self._operator_paused = bool(snap.get("operator_paused", False))
for aid in self.statuses:
self.runtimes.setdefault(aid, AgentRuntime())
+69 -18
View File
@@ -4,9 +4,12 @@ from __future__ import annotations
import asyncio
import contextlib
import inspect
import logging
import re
import uuid
from collections.abc import Callable
from dataclasses import replace
from functools import cache
from typing import TYPE_CHECKING, Any, cast
@@ -21,6 +24,8 @@ from openai import (
from strix.config import codex
from strix.core.hooks import (
SUBAGENT_TURN_LIMITS_KEY,
TURN_LIMIT_KEY,
BudgetExceededError,
BudgetPausedError,
SubagentBudgetReservedError,
@@ -30,11 +35,13 @@ from strix.core.sessions import (
enforce_image_budget,
open_agent_session,
replace_session_items,
scrub_images_from_items,
seed_initial_input,
strip_all_images_from_session,
)
from strix.llm import request_log
from strix.llm.compaction import is_context_overflow, maybe_compact
from strix.skills import skill_turn_limit
if TYPE_CHECKING:
@@ -44,6 +51,7 @@ if TYPE_CHECKING:
from agents.lifecycle import RunHooks
from agents.memory import Session, SQLiteSession
from agents.result import RunResultBase
from agents.run_config import CallModelData, ModelInputData
from strix.core.agents import AgentCoordinator, Status
@@ -53,6 +61,18 @@ logger = logging.getLogger(__name__)
StreamEventSink = Callable[[str, Any], None]
_INPUT_REJECTION_CODES = frozenset({400, 404, 422})
# Replies meaning "this model takes no images", not image errors in general: a
# context overflow that counts "image/vision expansion" must not match.
_IMAGE_REJECTION = re.compile(
r"no endpoints found that support image input" # OpenRouter
r"|image_url is only supported by certain models" # OpenAI
r"|is not a multimodal model|at most 0 image\(s\)" # vLLM
r"|does not support image input" # LiteLLM's Fireworks check
r"|doesn't support the image field" # Bedrock Converse
r"|unknown variant `image_url`" # DeepSeek, e.g. via Vercel AI Gateway
r"|'[^']*image[^']*' functionality not supported", # Vercel AI Gateway (AI SDK), unverified
re.IGNORECASE,
)
_MAX_COMPACTIONS_PER_CYCLE = 2
@@ -120,6 +140,28 @@ async def _compact_session(
)
_TEXT_ONLY_IMAGE_TEXT = "[error: this model cannot view images; use `snapshot -i` instead]"
def _with_image_scrub(run_config: RunConfig, context: dict[str, Any]) -> RunConfig:
if context.get("supports_images", True):
return run_config
# Chain any filter already set; it sees the scrubbed input.
inner = run_config.call_model_input_filter
async def _scrub(data: CallModelData[Any]) -> ModelInputData:
model_data = replace(
data.model_data,
input=scrub_images_from_items(data.model_data.input, text=_TEXT_ONLY_IMAGE_TEXT),
)
if inner is None:
return model_data
result = inner(replace(data, model_data=model_data))
return await result if inspect.isawaitable(result) else result
return replace(run_config, call_model_input_filter=_scrub)
_MAX_TRANSIENT_MODEL_RETRIES = 5
_TRANSIENT_MODEL_RETRY_BASE_DELAY_S = 2.0
_TRANSIENT_MODEL_RETRY_MAX_DELAY_S = 90.0
@@ -130,6 +172,12 @@ def _model_error_status_code(exc: BaseException) -> int | None:
return code if isinstance(code, int) else None
def _is_image_rejection(exc: BaseException) -> bool:
return _model_error_status_code(exc) in _INPUT_REJECTION_CODES and bool(
_IMAGE_REJECTION.search(str(exc))
)
def _is_transient_model_error(exc: BaseException) -> bool:
if codex.is_content_guardrail_error(exc):
return False
@@ -139,9 +187,7 @@ def _is_transient_model_error(exc: BaseException) -> bool:
return True
code = _model_error_status_code(exc)
if code is not None:
import litellm
return bool(litellm._should_retry(code))
return code >= 400 and code not in (401, 402, 403, 404)
return isinstance(exc, APIError)
@@ -259,7 +305,7 @@ async def _run_agent_loop(
raise SubagentBudgetReservedError("scan reached the sub-agent budget reserve")
if reserve_stopped and start_parked and interactive and context.get("parent_id") is None:
await coordinator.send(agent_id, _reserve_notice())
await coordinator.send(agent_id, _reserve_notice(coordinator.root_finish_tool))
if not (start_parked and interactive):
with contextlib.suppress(BudgetPausedError):
@@ -384,6 +430,7 @@ async def spawn_child_agent(
name=name,
parent_id=parent_id,
task=task,
skills=skills,
initial_input=child_initial_input(
name=name,
child_id=child_id,
@@ -467,6 +514,7 @@ async def respawn_subagents(
name=name,
parent_id=parent_id,
task=str(md.get("task", "")),
skills=child_skills,
initial_input=[],
start_parked=start_parked,
event_sink=event_sink,
@@ -556,7 +604,7 @@ async def _run_until_lifecycle(
hooks=hooks,
)
except BudgetPausedError as exc:
if coordinator.budget_policy != "pause":
if coordinator.budget_policy != "pause" and not exc.operator_pause:
raise
# The agent parked right before an LLM call; everything up to that
# point is already in its session. Once resumed, the same call goes
@@ -569,6 +617,7 @@ async def _run_until_lifecycle(
# Stopped while parked (operator stop or a parent's stop_agent).
await coordinator.reset_recovery(agent_id)
return result
await coordinator.consume_pending(agent_id)
if session is not None:
input_data = []
continue
@@ -599,6 +648,7 @@ async def _run_until_lifecycle(
input_data = await _append_tool_required_message(
session=session,
context=context,
root_finish_tool=coordinator.root_finish_tool,
attempt=recoveries,
limit=recovery_limit,
interactive=interactive,
@@ -747,7 +797,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
stream = Runner.run_streamed(
agent,
input=input_data,
run_config=run_config,
run_config=_with_image_scrub(run_config, context),
context=context,
max_turns=max_turns,
session=session,
@@ -787,7 +837,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
await coordinator.detach_stream(agent_id, stream)
except BudgetPausedError as exc:
logger.info("agent %s paused at the scan budget limit: %s", agent_id, exc)
if coordinator.budget_policy == "pause":
if coordinator.budget_policy == "pause" or exc.operator_pause:
await coordinator.park_for_budget(agent_id)
else:
await coordinator.pause_for_budget(agent_id)
@@ -805,11 +855,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
await coordinator.trigger_budget_stop()
raise
except Exception as exc:
if (
image_strips < 3
and session is not None
and getattr(exc, "status_code", None) in _INPUT_REJECTION_CODES
):
if image_strips < 3 and session is not None and _is_image_rejection(exc):
try:
stripped = await strip_all_images_from_session(session)
except Exception:
@@ -949,8 +995,9 @@ async def _append_tool_required_message(
limit: int,
interactive: bool,
silent_yield: bool = False,
root_finish_tool: str = "finish_scan",
) -> list[dict[str, str]]:
finish_tool = "finish_scan" if context.get("parent_id") is None else "agent_finish"
finish_tool = root_finish_tool if context.get("parent_id") is None else "agent_finish"
if silent_yield:
message = (
"You called wait_for_user without having written anything to the user since "
@@ -962,7 +1009,7 @@ async def _append_tool_required_message(
message = (
"Your previous message ended a turn without a tool call. Plain text never ends "
"execution and never hands control to the user: it is shown to the user, and the "
"run continues. Continue immediately and call exactly one tool. "
"run continues. Continue immediately and call a tool. "
"If you have nothing to do until the user replies, call wait_for_user; your "
"text already reached them, so do not repeat it. "
"If you are blocked waiting for another agent, call wait_for_agents. "
@@ -974,7 +1021,7 @@ async def _append_tool_required_message(
message = (
"Your previous response ended the autonomous run without a lifecycle tool "
"call. That is invalid in non-interactive mode; plain text final answers are "
"ignored. Continue immediately and call exactly one tool. "
"ignored. Continue immediately and call a tool. "
f"If your work is complete, call {finish_tool}. "
"If you are blocked waiting for another agent, call wait_for_agents. "
"Otherwise use the appropriate execution or planning tool. "
@@ -1068,7 +1115,7 @@ async def notify_parent_on_terminal(
)
def _reserve_notice() -> dict[str, Any]:
def _reserve_notice(finish_tool: str) -> dict[str, Any]:
return {
"from": "system",
"type": "budget_reserve_stop",
@@ -1078,7 +1125,7 @@ def _reserve_notice() -> dict[str, Any]:
"sub-agent is being force-stopped as soon as its in-flight turn completes, and "
"none will send a completion report. Their confirmed vulnerabilities are "
"already filed as they were found. Do not wait on any sub-agents and do not "
"spawn new ones — wrap up now and call finish_scan."
f"spawn new ones — wrap up now and call {finish_tool}."
),
}
@@ -1087,7 +1134,7 @@ async def _notify_root_on_budget_reserve(coordinator: AgentCoordinator) -> None:
root = await coordinator.claim_reserve_notification()
if root is None:
return
await coordinator.send(root, _reserve_notice())
await coordinator.send(root, _reserve_notice(coordinator.root_finish_tool))
async def _notify_parent_on_exit(
@@ -1119,6 +1166,7 @@ async def _start_child_runner(
name: str,
parent_id: str | None,
task: str,
skills: list[str],
initial_input: Any,
start_parked: bool = False,
event_sink: StreamEventSink | None = None,
@@ -1132,6 +1180,9 @@ async def _start_child_runner(
child_ctx["agent_id"] = child_id
child_ctx["parent_id"] = parent_id
child_ctx["task"] = task
child_ctx[TURN_LIMIT_KEY] = skill_turn_limit(
skills, parent_ctx.get(SUBAGENT_TURN_LIMITS_KEY) or {}
)
async def _child_loop() -> None:
# A budget stop is a clean scan-wide shutdown, not a child failure: the
+102 -7
View File
@@ -4,24 +4,32 @@ from __future__ import annotations
import logging
import math
import time
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from agents.lifecycle import RunHooks
from strix.core.agents import BudgetPolicy, coordinator_from_context
from strix.report.state import get_global_report_state
from strix.report.state import get_global_report_state, record_tool_call
if TYPE_CHECKING:
from agents import RunContextWrapper
from agents.agent import Agent
from agents.items import ModelResponse, TResponseInputItem
from agents.tool import Tool
logger = logging.getLogger(__name__)
LLM_TURN_KEY = "llm_turn"
# ``run_strix_scan(subagent_turn_limits=...)``, and the limit it gives one sub-agent.
SUBAGENT_TURN_LIMITS_KEY = "subagent_turn_limits"
TURN_LIMIT_KEY = "turn_limit"
_TURN_LIMIT_WARN_BEFORE = 10
_TURN_LIMIT_REPEAT = 5
# ``BudgetPolicy`` decides what happens when the accumulated LLM cost reaches
# ``max_budget_usd``.
@@ -35,6 +43,8 @@ LLM_TURN_KEY = "llm_turn"
# same call when the operator raises the limit or resumes.
__all__ = [
"LLM_TURN_KEY",
"SUBAGENT_TURN_LIMITS_KEY",
"TURN_LIMIT_KEY",
"BudgetExceededError",
"BudgetPausedError",
"BudgetPolicy",
@@ -65,9 +75,12 @@ class BudgetPausedError(RuntimeError):
decided to park; the agent waits for a resume newer than that.
"""
def __init__(self, message: str, *, resume_epoch: int = 0) -> None:
def __init__(
self, message: str, *, resume_epoch: int = 0, operator_pause: bool = False
) -> None:
super().__init__(message)
self.resume_epoch = resume_epoch
self.operator_pause = operator_pause
def _validate_budget(max_budget_usd: float | None) -> None:
@@ -104,16 +117,16 @@ _ROOT_DIRECTIVES: tuple[str, ...] = (
(
"As the root agent, begin planning your wind-down of the whole scan: avoid "
"starting large new lines of investigation, and keep your required objectives on "
"track so you can call finish_scan comfortably before the limit."
"track so you can call {finish_tool} comfortably before the limit."
),
(
"As the root agent, prioritize wrapping up the whole scan now: stop opening new "
"lines of investigation, close out only what is essential, and move toward calling "
"finish_scan to compile and deliver the final report."
"{finish_tool} to compile and deliver the final report."
),
(
"As the root agent, STOP all other work on the whole scan and finish immediately: "
"secure your findings and call finish_scan now — anything left unfinished when the "
"secure your findings and call {finish_tool} now — anything left unfinished when the "
"limit is hit is discarded."
),
)
@@ -138,8 +151,11 @@ _SUBAGENT_DIRECTIVES: tuple[str, ...] = (
def _wrapup_directive(context: RunContextWrapper[dict[str, Any]], stage: int) -> str:
is_root = context.context.get("parent_id") is None
directives = _ROOT_DIRECTIVES if is_root else _SUBAGENT_DIRECTIVES
return directives[stage]
if not is_root:
return _SUBAGENT_DIRECTIVES[stage]
coordinator = coordinator_from_context(context.context)
finish_tool = coordinator.root_finish_tool if coordinator is not None else "finish_scan"
return _ROOT_DIRECTIVES[stage].format(finish_tool=finish_tool)
def _urgency(stage: int) -> str:
@@ -157,6 +173,7 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
max_turns: int | None = None,
interactive: bool = False,
budget_policy: BudgetPolicy = "stop",
record_tool_timings: bool = False,
) -> None:
_validate_budget(max_budget_usd)
if max_turns is not None and max_turns <= 0:
@@ -169,6 +186,8 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
self._max_turns = max_turns
self._interactive = interactive
self._budget_policy: BudgetPolicy = budget_policy
self._record_tool_timings = record_tool_timings
self._tool_starts: dict[str, tuple[datetime, float]] = {}
@property
def max_budget_usd(self) -> float | None:
@@ -195,15 +214,26 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
system_prompt: str | None, # noqa: ARG002
input_items: list[TResponseInputItem],
) -> None:
self._pause_if_operator_paused(context)
if self._budget_policy == "pause":
self._pause_if_limited(context)
context.context[LLM_TURN_KEY] = int(context.context.get(LLM_TURN_KEY, 0)) + 1
try:
self._maybe_warn_turns(context, input_items)
self._maybe_warn_soft_turn_limit(context, input_items)
self._maybe_warn_budget(context, input_items)
except Exception:
logger.exception("budget/turn warning injection failed")
def _pause_if_operator_paused(self, context: RunContextWrapper[dict[str, Any]]) -> None:
coordinator = coordinator_from_context(context.context)
if coordinator is not None and coordinator.operator_paused:
raise BudgetPausedError(
"scan paused; waiting for the operator to resume",
resume_epoch=coordinator.resume_epoch,
operator_pause=True,
)
def _pause_if_limited(self, context: RunContextWrapper[dict[str, Any]]) -> None:
"""Park the agent before a paid call when the scan is at its limit or paused.
@@ -254,6 +284,40 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
)
input_items.append({"role": "user", "content": content})
def _maybe_warn_soft_turn_limit(
self,
context: RunContextWrapper[dict[str, Any]],
input_items: list[TResponseInputItem],
) -> None:
"""A soft limit, never a stop: warn 10 turns before it, then every few turns past it."""
limit = context.context.get(TURN_LIMIT_KEY)
requests = getattr(getattr(context, "usage", None), "requests", None)
if not isinstance(limit, int) or not isinstance(requests, int):
return
turns_used = requests + 1
if turns_used == limit - _TURN_LIMIT_WARN_BEFORE:
content = (
f"[NOTICE] You have used {turns_used} of your {limit}-turn limit. Start "
"wrapping up: take only the steps you still need to answer your task, then "
"call agent_finish. You may go past the limit only while you are still "
"making real progress."
)
elif turns_used >= limit and (turns_used - limit) % _TURN_LIMIT_REPEAT == 0:
content = (
f"[NOTICE] You have used {turns_used} turns, past your {limit}-turn limit. "
"Call agent_finish now with what you have, listing anything unresolved in "
"open_items, unless your next steps are clearly making progress on your task."
)
else:
return
logger.info(
"Turn-limit notice for agent %s: turn %d, limit %d",
context.context.get("agent_id"),
turns_used,
limit,
)
input_items.append({"role": "user", "content": content})
def _maybe_warn_budget(
self,
context: RunContextWrapper[dict[str, Any]],
@@ -299,6 +363,37 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
)
input_items.append({"role": "user", "content": content})
async def on_tool_start(
self,
context: RunContextWrapper[dict[str, Any]],
agent: Agent[dict[str, Any]], # noqa: ARG002
tool: Tool, # noqa: ARG002
) -> None:
call_id = getattr(context, "tool_call_id", None)
if self._record_tool_timings and isinstance(call_id, str):
self._tool_starts[call_id] = (datetime.now(UTC), time.monotonic())
async def on_tool_end(
self,
context: RunContextWrapper[dict[str, Any]],
agent: Agent[dict[str, Any]], # noqa: ARG002
tool: Tool,
result: object, # noqa: ARG002
) -> None:
call_id = getattr(context, "tool_call_id", None)
start = self._tool_starts.pop(call_id, None) if isinstance(call_id, str) else None
if start is None:
return
started_at, started = start
ctx = context.context if isinstance(context.context, dict) else {}
agent_id = ctx.get("agent_id")
record_tool_call(
started_at=started_at,
agent_id=agent_id if isinstance(agent_id, str) else None,
tool=tool.name,
duration_ms=round((time.monotonic() - started) * 1000),
)
async def on_llm_end(
self,
context: RunContextWrapper[dict[str, Any]],
+16 -10
View File
@@ -10,12 +10,11 @@ from openai.types.shared import Reasoning
from strix.config.models import (
DEFAULT_MODEL_RETRY,
OPENROUTER_ATTRIBUTION_HEADERS,
attribution_headers,
bedrock_route_supports_prompt_caching,
is_bedrock_route,
is_claude_model,
is_known_openai_bare_model,
is_openrouter_model,
model_supports_reasoning,
request_timeout_extra_args,
routes_through_litellm,
@@ -247,25 +246,30 @@ def make_model_settings(
*,
model_name: str,
force_required_tool_choice: bool = False,
parallel_tool_calls: bool = False,
request_timeout: float | None = None,
prompt_cache: bool = True,
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, Any] | None = None,
has_tools: bool = True,
) -> ModelSettings:
headers = _request_headers(model_name, extra_headers)
model_settings = ModelSettings(
parallel_tool_calls=False if has_tools else None,
parallel_tool_calls=parallel_tool_calls if has_tools else None,
retry=DEFAULT_MODEL_RETRY,
include_usage=True,
extra_args=request_timeout_extra_args(request_timeout),
extra_headers=headers,
extra_body=dict(extra_body) if extra_body else None,
)
if (
reasoning_effort is not None
and reasoning_effort != "none"
and model_supports_reasoning(model_name)
):
model_settings = model_settings.resolve(_reasoning_settings(reasoning_effort))
model_settings = model_settings.resolve(
_reasoning_settings(reasoning_effort, extra_body),
)
if force_required_tool_choice and _accepts_required_tool_choice(model_name):
model_settings = model_settings.resolve(ModelSettings(tool_choice="required"))
@@ -282,15 +286,15 @@ def make_model_settings(
def _request_headers(
model_name: str, extra_headers: dict[str, str] | None
) -> dict[str, str] | None:
headers: dict[str, str] = {}
if is_openrouter_model(model_name):
headers.update(OPENROUTER_ATTRIBUTION_HEADERS)
headers: dict[str, str] = dict(attribution_headers(model_name))
if extra_headers:
headers.update(extra_headers)
return headers or None
def _reasoning_settings(effort: ReasoningEffort) -> ModelSettings:
def _reasoning_settings(
effort: ReasoningEffort, extra_body: dict[str, Any] | None = None
) -> ModelSettings:
"""``max`` is not in the OpenAI SDK's ``Reasoning.effort`` enum, so send it as
a raw body field instead — also keeping it clear of LiteLLM's DeepSeek mapping,
which collapses every ``reasoning_effort`` level to plain thinking-enabled.
@@ -298,11 +302,13 @@ def _reasoning_settings(effort: ReasoningEffort) -> ModelSettings:
It goes in ``extra_body``, the field every model implementation forwards as the
request's ``extra_body``; the same value under ``extra_args`` collides with that
keyword and raises before a request is ever sent.
keyword and raises before a request is ever sent. ``resolve`` replaces
``extra_body`` wholesale, so the user's ``LLM_EXTRA_BODY`` fields are carried
over, with ``reasoning_effort`` taking precedence.
"""
if effort != "max":
return ModelSettings(reasoning=Reasoning(effort=effort))
return ModelSettings(extra_body={"reasoning_effort": "max"})
return ModelSettings(extra_body={**(extra_body or {}), "reasoning_effort": "max"})
def _prompt_cache_extra_args(model_name: str) -> dict[str, Any] | None:
+86 -14
View File
@@ -8,7 +8,7 @@ import io
import json
import logging
import uuid
from collections.abc import Callable
from collections.abc import Awaitable, Callable, Mapping
from pathlib import Path
from typing import TYPE_CHECKING, Any
@@ -23,6 +23,7 @@ from strix.config.models import (
StrixProvider,
configure_sdk_api_route,
configure_sdk_model_defaults,
model_supports_images,
supports_strict_tool_schemas,
uses_chat_completions_tool_schema,
)
@@ -35,7 +36,12 @@ from strix.core.execution import (
from strix.core.execution import (
spawn_child_agent as start_child_agent,
)
from strix.core.hooks import BudgetExceededError, ReportUsageHooks, recomputed_budget_flags
from strix.core.hooks import (
SUBAGENT_TURN_LIMITS_KEY,
BudgetExceededError,
ReportUsageHooks,
recomputed_budget_flags,
)
from strix.core.inputs import (
build_root_task,
build_scan_targets,
@@ -48,15 +54,18 @@ from strix.report.state import get_global_report_state
from strix.runtime import session_manager
from strix.telemetry import set_scan_phase
from strix.telemetry.logging import set_scan_id, setup_scan_logging
from strix.tools.finish.tool import finish_scan
from strix.tools.output_store import (
WORKSPACE_SPILL_DIR,
configure_spill_writer,
)
from strix.tools.reporting.tool import create_vulnerability_report
if TYPE_CHECKING:
from agents.memory import SQLiteSession
from agents.result import RunResultBase
from agents.tool import Tool
from strix.runtime.status import StatusSink
from strix.tools.mcp import (
@@ -67,6 +76,8 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
_MCP_PROMPT_WARMUP_TIMEOUT_SECONDS = 30.0
StreamEventSink = Callable[[str, Any], None]
# Receives the run's MCP connection roster as a list of non-secret status dicts
@@ -92,6 +103,19 @@ def _mcp_roster_payload(registry: McpRegistry) -> list[dict[str, Any]]:
]
def _mcp_prompt_roster(registry: McpRegistry) -> list[dict[str, Any]]:
"""The MCP roster rendered in agent prompts, with only verified counts."""
return [
{
"name": summary.name,
"purpose": summary.purpose,
"tool_count": summary.tool_count if summary.state == "catalog_ready" else None,
"state": summary.state,
}
for summary in registry.summaries()
]
def _record_mcp_connections(connection_names: list[str]) -> None:
"""Record which MCP servers this run configured, for the interfaces.
@@ -152,10 +176,17 @@ def _compose_root_instructions_override(
is_whitebox: bool,
is_diff_scoped: bool,
interactive: bool,
parallel_tool_calls: bool,
system_prompt_context: dict[str, Any],
supports_images: bool,
report_tool: str,
coverage_tools: bool,
replace_scan_prompt: bool = False,
) -> str | None:
if root_instructions_override is None:
return None
if replace_scan_prompt:
return f"{root_instructions_override}\n\n{render_scope_prompt(system_prompt_context)}"
base_instructions = render_system_prompt(
skills=skills,
@@ -164,8 +195,12 @@ def _compose_root_instructions_override(
is_root=True,
is_diff_scoped=is_diff_scoped,
interactive=interactive,
parallel_tool_calls=parallel_tool_calls,
system_prompt_context=system_prompt_context,
include_scope=False,
supports_images=supports_images,
report_tool=report_tool,
coverage_tools=coverage_tools,
)
return (
f"{base_instructions}\n\n"
@@ -183,7 +218,7 @@ async def run_strix_scan(
scan_id: str | None = None,
image: str,
local_sources: list[dict[str, Any]] | None = None,
extra_files: list[dict[str, Any]] | None = None,
extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None = None,
coordinator: AgentCoordinator | None = None,
interactive: bool = False,
max_turns: int = DEFAULT_MAX_TURNS,
@@ -193,15 +228,22 @@ async def run_strix_scan(
cleanup_on_exit: bool = True,
event_sink: StreamEventSink | None = None,
root_instructions_override: str | None = None,
replace_root_scan_prompt: bool = False,
extra_system_prompt_context: dict[str, Any] | None = None,
status_sink: StatusSink | None = None,
mcp_connection_requests: list[McpConnectionRequest] | None = None,
mcp_status_sink: McpStatusSink | None = None,
root_finish_tool: Tool = finish_scan,
report_tool: Tool = create_vulnerability_report,
coverage_tools: bool = True,
subagent_turn_limits: Mapping[str, int] | None = None,
) -> RunResultBase | None:
"""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.
``replace_root_scan_prompt=True`` makes it the root's whole prompt, followed
by the scope, instead of adding it to the rendered scan prompt.
``extra_files`` entries (``{"workspace_path", "content"}``) are placed into
the sandbox workspace at session bring-up; see
:func:`strix.runtime.session_manager.create_or_reuse`.
@@ -219,6 +261,14 @@ async def run_strix_scan(
command-line default) it reads ``~/.strix/mcp-servers.json`` itself. Either
way the engine does the connecting, so the caller passes inert configs plus
metadata and never live sessions.
``root_finish_tool`` is the tool the root agent ends the run with.
``report_tool`` is the tool every agent files findings with, in place of
``create_vulnerability_report``.
``coverage_tools=False`` drops the coverage ledger tools and their prompt
rules from every agent.
``subagent_turn_limits`` maps a skill name to a soft turn limit for sub-agents
spawned with that skill (the largest applies). Agents are warned 10 turns
before it and reminded past it, but never stopped.
"""
def report(phase: str) -> None:
@@ -265,6 +315,9 @@ async def run_strix_scan(
strict_tool_schemas = supports_strict_tool_schemas(resolved_model)
if not strict_tool_schemas:
logger.info("Sending non-strict tool schemas: %s caps strict tools", resolved_model)
supports_images = model_supports_images(resolved_model)
if not supports_images:
logger.info("Leaving out image tools: %s does not accept images", resolved_model)
if budget_policy not in ("stop", "pause"):
raise ValueError(f"unknown budget_policy: {budget_policy!r}")
@@ -272,6 +325,7 @@ async def run_strix_scan(
coordinator = AgentCoordinator()
coordinator.set_snapshot_path(agents_path)
coordinator.set_budget_policy(budget_policy)
coordinator.root_finish_tool = root_finish_tool.name
from strix.tools.coverage.tools import hydrate_coverage_from_disk
from strix.tools.notes.tools import hydrate_notes_from_disk
@@ -372,9 +426,11 @@ async def run_strix_scan(
settings.llm.reasoning_effort,
model_name=resolved_model,
force_required_tool_choice=settings.llm.force_required_tool_choice,
parallel_tool_calls=settings.llm.parallel_tool_calls,
request_timeout=settings.llm.timeout,
prompt_cache=settings.llm.prompt_cache,
extra_headers=settings.llm.extra_headers,
extra_body=settings.llm.extra_body,
)
run_config = RunConfig(
model=resolved_model,
@@ -392,6 +448,7 @@ async def run_strix_scan(
max_turns=max_turns,
interactive=interactive,
budget_policy=budget_policy,
record_tool_timings=settings.telemetry.run_record_timings,
)
coordinator.set_budget_limit_setter(hooks.set_max_budget_usd)
if interactive and budget_policy != "pause":
@@ -432,19 +489,13 @@ async def run_strix_scan(
_record_mcp_connections(mcp_registry.names())
report(
f"MCP: configured {len(mcp_registry)} connection(s); "
"warming them in the background"
"connecting and listing tools"
)
scope_context["mcp_available"] = True
scope_context["mcp_connections"] = [
{
"name": summary.name,
"purpose": summary.purpose,
"tool_count": summary.tool_count,
}
for summary in mcp_registry.summaries()
]
scope_context["mcp_connections"] = _mcp_prompt_roster(mcp_registry)
def _emit_mcp_status() -> None:
scope_context["mcp_connections"] = _mcp_prompt_roster(mcp_registry)
roster = _mcp_roster_payload(mcp_registry)
_persist_mcp_status(roster)
if mcp_status_sink is not None:
@@ -455,7 +506,11 @@ async def run_strix_scan(
mcp_registry.set_status_sink(_emit_mcp_status)
_emit_mcp_status()
mcp_registry.start_warmup(max_concurrency=6)
warmup_task = mcp_registry.start_warmup(max_concurrency=6)
await asyncio.wait(
{warmup_task},
timeout=_MCP_PROMPT_WARMUP_TIMEOUT_SECONDS,
)
except Exception:
logger.exception("Failed to configure user MCP servers; continuing without them")
@@ -467,7 +522,12 @@ async def run_strix_scan(
is_whitebox=is_whitebox,
is_diff_scoped=is_diff_scoped,
interactive=interactive,
parallel_tool_calls=settings.llm.parallel_tool_calls,
system_prompt_context=root_context,
supports_images=supports_images,
report_tool=report_tool.name,
coverage_tools=coverage_tools,
replace_scan_prompt=replace_root_scan_prompt,
)
root_agent = build_strix_agent(
@@ -478,10 +538,15 @@ async def run_strix_scan(
is_whitebox=is_whitebox,
is_diff_scoped=is_diff_scoped,
interactive=interactive,
parallel_tool_calls=settings.llm.parallel_tool_calls,
chat_completions_tools=chat_completions_tools,
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=root_context,
instructions_override=root_instructions,
supports_images=supports_images,
finish_tool=root_finish_tool,
report_tool=report_tool,
coverage_tools=coverage_tools,
)
if not is_resume:
@@ -498,9 +563,13 @@ async def run_strix_scan(
is_whitebox=is_whitebox,
is_diff_scoped=is_diff_scoped,
interactive=interactive,
parallel_tool_calls=settings.llm.parallel_tool_calls,
chat_completions_tools=chat_completions_tools,
strict_tool_schemas=strict_tool_schemas,
system_prompt_context=scope_context,
supports_images=supports_images,
report_tool=report_tool,
coverage_tools=coverage_tools,
)
async def spawn_child_agent(**kwargs: Any) -> dict[str, Any]:
@@ -528,6 +597,9 @@ async def run_strix_scan(
"spawn_child_agent": spawn_child_agent,
"scan_targets": build_scan_targets(scan_config),
"max_context_images": settings.runtime.max_context_images,
"supports_images": supports_images,
"coverage_tools": coverage_tools,
SUBAGENT_TURN_LIMITS_KEY: dict(subagent_turn_limits or {}),
}
root_session = open_agent_session(root_id, agents_db)
@@ -592,7 +664,7 @@ async def run_strix_scan(
)
if not interactive and result is not None:
final = getattr(result, "final_output", None)
# Lifecycle tools mark the root completed.
# Lifecycle tools mark the root completed.
async with coordinator._lock:
root_completed = coordinator.statuses.get(root_id) == "completed"
if not root_completed:
+2 -2
View File
@@ -199,13 +199,13 @@ async def enforce_image_budget(session: Session, max_images: int) -> bool:
return await _rewrite_session(session, _transform)
def scrub_images_from_items(items: list[Any]) -> list[Any]:
def scrub_images_from_items(items: list[Any], *, text: str = _INHERITED_IMAGE_TEXT) -> list[Any]:
"""Return a copy of ``items`` with every image block replaced by text."""
def _scrub(obj: Any) -> Any:
if isinstance(obj, dict):
if obj.get("type") == "input_image":
return {"type": "input_text", "text": _INHERITED_IMAGE_TEXT}
return {"type": "input_text", "text": text}
return {k: _scrub(v) for k, v in obj.items()}
if isinstance(obj, list):
return [_scrub(v) for v in obj]
+3 -2
View File
@@ -164,12 +164,13 @@ async def warm_up_llm() -> None:
raw_model = dedupe_model
deduper = resolve_dedupe_model(settings.dedupe, dedupe_model)
# A dedicated dedupe model may route to another provider, which must
# never receive the main endpoint's headers; it has its own
# DEDUPE_LLM_EXTRA_HEADERS.
# never receive the main endpoint's headers or body fields; it has
# its own DEDUPE_LLM_EXTRA_HEADERS and DEDUPE_LLM_EXTRA_BODY.
await preflight_request(
deduper,
model_name=dedupe_model,
extra_headers=settings.dedupe.extra_headers,
extra_body=settings.dedupe.extra_body,
timeout=llm.preflight_timeout,
api_base_setting="DEDUPE_LLM_API_BASE",
)
+3
View File
@@ -116,6 +116,7 @@ async def preflight_model_connection(
model,
model_name=model_name,
extra_headers=resolved_settings.llm.extra_headers,
extra_body=resolved_settings.llm.extra_body,
timeout=resolved_settings.llm.preflight_timeout,
api_base_setting="LLM_API_BASE",
)
@@ -126,6 +127,7 @@ async def preflight_request(
*,
model_name: str,
extra_headers: dict[str, str] | None,
extra_body: dict[str, Any] | None,
timeout: int,
api_base_setting: str,
) -> None:
@@ -144,6 +146,7 @@ async def preflight_request(
request_timeout=timeout,
prompt_cache=False,
extra_headers=extra_headers,
extra_body=extra_body,
has_tools=False,
)
try:
+2 -2
View File
@@ -66,7 +66,7 @@ func panelFocus(panel sidebarPanel) (focusMode, bool) {
func (m Model) sidebarPanels() []panelRect {
statsHeight, vulnHeight, mcpHeight, agentHeight := m.sidebarHeights()
top := m.viewerHeight()
top := m.sidebarTopHeight()
rects := []panelRect{{panelAgents, top, agentHeight}}
top += agentHeight
if vulnHeight > 0 {
@@ -83,7 +83,7 @@ func (m Model) sidebarPanels() []panelRect {
func (m Model) sidebarGap() int {
statsHeight, vulnHeight, mcpHeight, agentHeight := m.sidebarHeights()
return max(0, m.height-m.viewerHeight()-agentHeight-vulnHeight-mcpHeight-statsHeight)
return max(0, m.height-m.sidebarTopHeight()-agentHeight-vulnHeight-mcpHeight-statsHeight)
}
func (m Model) panelAt(y int) (panelRect, bool) {
@@ -355,3 +355,77 @@ func TestLonePanelHasNoControls(t *testing.T) {
t.Fatalf("controls should return once a second panel exists")
}
}
func TestTargetsBoxSitsUnderViewer(t *testing.T) {
m := panelsModel(t)
m.snapshot.Targets = []string{"https://example.com", "./app"}
m.snapshot.TargetCount = 2
m.resizeViewport()
_, sidebarWidth, _, _ := m.layout()
rows := strings.Split(ansi.Strip(m.sidebarView(sidebarWidth, m.height)), "\n")
if len(rows) != m.height {
t.Fatalf("sidebar renders %d rows, want %d", len(rows), m.height)
}
top := m.viewerHeight()
for i, want := range []string{"Targets 2", "▸ https://example.com", "▸ ./app"} {
if !strings.Contains(rows[top+1+i], want) {
t.Fatalf("row %d missing %q:\n%s", top+1+i, want, strings.Join(rows, "\n"))
}
}
if !strings.Contains(rows[panelRectOf(t, m, panelAgents).top+1], "Agents") {
t.Fatalf("agents panel not where its rect says:\n%s", strings.Join(rows, "\n"))
}
}
func TestTargetsBoxDropsOnShortTerminal(t *testing.T) {
m := panelsModel(t)
m.height = 19
m.snapshot.Targets = []string{"a", "b", "c", "d", "e"}
m.snapshot.TargetCount = 5
m.resizeViewport()
_, sidebarWidth, _, _ := m.layout()
if got := lipgloss.Height(m.sidebarView(sidebarWidth, m.height)); got > m.height {
t.Fatalf("sidebar renders %d rows on a %d-row terminal", got, m.height)
}
}
// The targets box only takes rows the agent tree would otherwise grow into, so
// the stats panel keeps the Caido URL and the other panels keep their heights.
func TestTargetsBoxNeverSqueezesPanels(t *testing.T) {
for height := 19; height <= 40; height++ {
m := panelsModel(t)
m.height = height
m.snapshot.Usage = map[string]any{"total_tokens": 1234, "cost": 0.5}
m.snapshot.CaidoURL = "http://127.0.0.1:8080"
m.resizeViewport()
stats, vuln, mcp, _ := m.sidebarHeights()
m.snapshot.Targets = []string{"a", "b", "c", "d", "e"}
m.snapshot.TargetCount = 5
m.resizeViewport()
gotStats, gotVuln, gotMcp, _ := m.sidebarHeights()
if gotStats != stats || gotVuln != vuln || gotMcp != mcp {
t.Fatalf("height %d: targets squeezed panels: stats %d->%d findings %d->%d mcp %d->%d",
height, stats, gotStats, vuln, gotVuln, mcp, gotMcp)
}
_, sidebarWidth, _, _ := m.layout()
if got := lipgloss.Height(m.sidebarView(sidebarWidth, m.height)); got != m.height {
t.Fatalf("height %d: sidebar renders %d rows", height, got)
}
}
}
func TestTargetsViewCountsTheTargetsItCannotShow(t *testing.T) {
m := panelsModel(t)
m.snapshot.Targets = []string{"a", "b", "c", "d", "e"}
m.snapshot.TargetCount = 5
for rows, want := range map[int]string{
2: "Targets 5\n+5 more",
3: "Targets 5\n▸ a\n+4 more",
6: "Targets 5\n▸ a\n▸ b\n▸ c\n▸ d\n+1 more",
9: "Targets 5\n▸ a\n▸ b\n▸ c\n▸ d\n+1 more",
} {
if got := ansi.Strip(m.targetsView(20, rows)); got != want {
t.Fatalf("%d rows:\n%s\nwant:\n%s", rows, got, want)
}
}
}
+13 -6
View File
@@ -421,7 +421,7 @@ func (m Model) setupComposer(width int) string {
inner := max(1, width-4)
body := m.highlightInputSelection(m.input.View())
body += "\n\n" + m.setupSummaryView(inner)
if targets := m.setupTargetsView(inner); targets != "" {
if targets := m.targetsView(inner, maxTargetRows); targets != "" {
body += "\n" + targets
}
return lipgloss.NewStyle().Width(max(1, width-2)).Padding(0, 1).
@@ -454,15 +454,22 @@ func (m Model) setupSummaryView(width int) string {
return truncate(strings.Join(chips, render.Dim().Render(" · ")), max(1, width))
}
// setupTargetsView lists what the scan is pointed at, once anything is queued.
func (m Model) setupTargetsView(width int) string {
if len(m.snapshot.Targets) == 0 {
// maxTargetRows fits the header, four targets and the count of the rest.
const maxTargetRows = 6
// targetsView lists what the scan is pointed at, once anything is queued, in at
// most maxRows rows. Targets that do not fit are counted on the last row.
func (m Model) targetsView(width, maxRows int) string {
if len(m.snapshot.Targets) == 0 || maxRows < 2 {
return ""
}
const visible = 4
total := max(m.snapshot.TargetCount, len(m.snapshot.Targets))
visible := min(len(m.snapshot.Targets), maxTargetRows-2, maxRows-1)
if visible < total {
visible = min(visible, maxRows-2)
}
rows := []string{render.Bold(green).Render("Targets") + render.Dim().Render(fmt.Sprintf(" %d", total))}
for _, target := range m.snapshot.Targets[:min(visible, len(m.snapshot.Targets))] {
for _, target := range m.snapshot.Targets[:visible] {
rows = append(rows, render.Col(dim).Render("▸ ")+render.Col(white).Render(truncate(target, max(1, width-2))))
}
if hidden := total - visible; hidden > 0 {
+35
View File
@@ -542,6 +542,15 @@ func (m Model) sidebarView(width, height int) string {
BorderForeground(dark).
Padding(0, 1).
Render(m.viewerBox(width - 4))}
if targetsHeight := m.targetsBoxHeight(); targetsHeight > 0 {
parts = append(parts, lipgloss.NewStyle().
Width(width-2).
Height(targetsHeight-2).
Border(lipgloss.RoundedBorder()).
BorderForeground(dark).
Padding(0, 1).
Render(m.targetsView(width-4, targetsHeight-2)))
}
agents := ""
if agentHeight > 1 {
agentRows := max(1, agentHeight-4)
@@ -592,6 +601,13 @@ func (m Model) sidebarView(width, height int) string {
}
func (m Model) sidebarHeights() (statsHeight, vulnHeight, mcpHeight, agentHeight int) {
_, statsHeight, vulnHeight, mcpHeight, agentHeight = m.sidebarLayout()
return
}
// sidebarLayout sizes the targets box and the panels. The targets box only
// takes rows no panel needs, so it never squeezes the panels below it.
func (m Model) sidebarLayout() (targetsHeight, statsHeight, vulnHeight, mcpHeight, agentHeight int) {
// Measure the stats panel the way its box will render it: a long model name
// wraps inside the sidebar, and counting only its newlines would size the
// box short and push the whole frame past the bottom of the terminal.
@@ -628,6 +644,14 @@ func (m Model) sidebarHeights() (statsHeight, vulnHeight, mcpHeight, agentHeight
}
}
spare := max(0, m.height-m.viewerHeight()-statsHeight-vulnHeight-mcpHeight-agentHeight)
// The box needs its border and two rows: the header and a target or the
// count of hidden ones. A zoomed panel keeps the whole sidebar.
if targets := m.targetsView(m.sidebarInnerWidth(), maxTargetRows); targets != "" && m.zoomedPanel == panelNone {
if fit := min(lipgloss.Height(targets)+2, spare); fit >= 4 {
targetsHeight = fit
spare -= fit
}
}
switch {
case m.zoomedPanel == panelFindings && vulnHeight > 1:
vulnHeight += spare
@@ -643,6 +667,17 @@ func (m Model) viewerHeight() int {
return strings.Count(m.viewerView(m.viewerContentWidth()), "\n") + 3
}
// sidebarTopHeight is the rows above the panels: the viewer box and, when the
// run has targets, the targets box under it.
func (m Model) sidebarTopHeight() int {
return m.viewerHeight() + m.targetsBoxHeight()
}
func (m Model) targetsBoxHeight() int {
targetsHeight, _, _, _, _ := m.sidebarLayout()
return targetsHeight
}
func (m Model) sidebarInnerWidth() int {
_, sidebarWidth, _, _ := m.layout()
if sidebarWidth == 0 {
+2
View File
@@ -0,0 +1,2 @@
# Supply-chain cooldown: only install versions published at least 5 days ago (npm >= 11.10).
min-release-age=5
+70 -4
View File
@@ -29,7 +29,7 @@
"@vitejs/plugin-react": "^5.0.0",
"tailwindcss": "^4.1.0",
"typescript": "^5.6.0",
"vite": "^6.0.0"
"vite": "^6.4.4"
}
},
"node_modules/@babel/code-frame": {
@@ -1401,6 +1401,72 @@
"node": ">=14.0.0"
}
},
"node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/core": {
"version": "1.11.1",
"dev": true,
"inBundle": true,
"license": "MIT",
"optional": true,
"dependencies": {
"@emnapi/wasi-threads": "1.2.2",
"tslib": "^2.4.0"
}
},
"node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/runtime": {
"version": "1.11.1",
"dev": true,
"inBundle": true,
"license": "MIT",
"optional": true,
"dependencies": {
"tslib": "^2.4.0"
}
},
"node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/wasi-threads": {
"version": "1.2.2",
"dev": true,
"inBundle": true,
"license": "MIT",
"optional": true,
"dependencies": {
"tslib": "^2.4.0"
}
},
"node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@napi-rs/wasm-runtime": {
"version": "1.1.4",
"dev": true,
"inBundle": true,
"license": "MIT",
"optional": true,
"dependencies": {
"@tybys/wasm-util": "^0.10.1"
},
"funding": {
"type": "github",
"url": "https://github.com/sponsors/Brooooooklyn"
},
"peerDependencies": {
"@emnapi/core": "^1.7.1",
"@emnapi/runtime": "^1.7.1"
}
},
"node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@tybys/wasm-util": {
"version": "0.10.2",
"dev": true,
"inBundle": true,
"license": "MIT",
"optional": true,
"dependencies": {
"tslib": "^2.4.0"
}
},
"node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/tslib": {
"version": "2.8.1",
"dev": true,
"inBundle": true,
"license": "0BSD",
"optional": true
},
"node_modules/@tailwindcss/oxide-win32-arm64-msvc": {
"version": "4.3.3",
"resolved": "https://registry.npmjs.org/@tailwindcss/oxide-win32-arm64-msvc/-/oxide-win32-arm64-msvc-4.3.3.tgz",
@@ -4096,9 +4162,9 @@
}
},
"node_modules/vite": {
"version": "6.4.3",
"resolved": "https://registry.npmjs.org/vite/-/vite-6.4.3.tgz",
"integrity": "sha512-NTKlcQjlAK7MlQoyb6LgaqHc8sso/pVyUJYWMws3jg21uTJw/LddqIFPcPqP6PzpgbIcZyKI85sFE4HBrQDA8A==",
"version": "6.4.4",
"resolved": "https://registry.npmjs.org/vite/-/vite-6.4.4.tgz",
"integrity": "sha512-xGjU/SyjeNyb0E4KxXBvYz/1C+3JXuujvf2hXNOtnlcE+58o5ajd9N2/n2VhkiJvpCP2qQasnUpx5PXNeI+p1Q==",
"dev": true,
"license": "MIT",
"dependencies": {
+1 -1
View File
@@ -30,6 +30,6 @@
"@vitejs/plugin-react": "^5.0.0",
"tailwindcss": "^4.1.0",
"typescript": "^5.6.0",
"vite": "^6.0.0"
"vite": "^6.4.4"
}
}
+1
View File
@@ -307,6 +307,7 @@ async def _summarize(model: str, prompt: str, max_tokens: int) -> str | None:
request_timeout=llm.timeout,
prompt_cache=False,
extra_headers=llm.extra_headers,
extra_body=llm.extra_body,
has_tools=False,
).resolve(ModelSettings(max_tokens=max_tokens))
try:
+47 -4
View File
@@ -159,11 +159,33 @@ class HttpReply:
status_code: int | None = None
headers: dict[str, str] | None = None
upstream_provider: str | None = None
upstream_error_type: str | None = None
_http_reply: ContextVar[HttpReply | None] = ContextVar("strix_llm_http_reply", default=None)
def record_upstream_provider(provider: object, error_type: object = None) -> None:
"""Remember which provider behind a gateway served the attempt in flight."""
reply = _http_reply.get()
if reply is None:
return
if isinstance(provider, str) and provider:
reply.upstream_provider = provider
if isinstance(error_type, str) and error_type:
reply.upstream_error_type = error_type
def _upstream_details(reply: HttpReply | None) -> list[tuple[str, object]]:
if reply is None:
return []
return [
("upstream_provider", reply.upstream_provider),
("upstream_error_type", reply.upstream_error_type),
]
async def record_http_reply(response: Response) -> None:
"""httpx ``response`` event hook: remember the reply for the attempt in flight.
@@ -772,6 +794,7 @@ def _litellm_success(
("request", _litellm_request_details(kwargs)),
("response", _litellm_response_details(response)),
("litellm", _litellm_hidden_details(hidden, slo)),
*_upstream_details(_http_reply.get()),
),
)
@@ -821,6 +844,7 @@ def _litellm_failure(
("request", _litellm_request_details(kwargs)),
("error", error or None),
("litellm", _litellm_hidden_details(hidden, slo)),
*_upstream_details(_http_reply.get()),
),
)
@@ -894,15 +918,18 @@ def _observe_sdk_shared_http_client() -> None:
def _log_line_sink(event: LlmRequestEvent) -> None:
details = event.details or {}
level = logging.DEBUG if event.outcome == "success" else logging.WARNING
logger.log(
level,
"llm_request route=%s provider=%s model=%s host=%s outcome=%s status=%s "
"request_id=%s response_id=%s stream=%s duration_ms=%d ttft_ms=%s "
"llm_request route=%s provider=%s upstream=%s upstream_error=%s model=%s host=%s "
"outcome=%s status=%s request_id=%s response_id=%s stream=%s duration_ms=%d ttft_ms=%s "
"req_bytes=%s res_bytes=%s finish=%s "
"in=%s out=%s cached=%s cost=%s agent=%s attempt=%d%s",
event.route,
event.provider or "-",
details.get("upstream_provider") or "-",
details.get("upstream_error_type") or "-",
event.model,
event.api_host or "-",
event.outcome,
@@ -1070,6 +1097,7 @@ class RequestLoggingModel(Model):
details=merge_details(
("request", request.details),
("response", _openai_response_details(response, raw_response)),
*_upstream_details(reply),
),
)
status, request_id = _openai_error_fields(exc)
@@ -1082,13 +1110,14 @@ class RequestLoggingModel(Model):
outcome="error",
status_code=status,
provider_request_id=request_id or base.provider_request_id,
error_type=type(exc).__name__,
error_type=_cancel_reason(exc) or type(exc).__name__,
error_message=_abandonment_message(exc) or clean_error_message(exc),
response_bytes=_exception_body_size(exc, status),
response_headers=headers_from_response(error_headers) or base.response_headers,
details=merge_details(
("request", request.details),
("error", _exception_details(exc)),
*_upstream_details(reply),
),
)
@@ -1318,9 +1347,23 @@ def _is_abandonment(exc: BaseException | None) -> bool:
return isinstance(exc, asyncio.CancelledError | GeneratorExit)
# Prefixes the reason a stream timeout gives when it cancels an attempt.
CANCEL_REASON_PREFIX = "strix:"
def _cancel_reason(exc: BaseException) -> str | None:
"""The stream timeout that cancelled the attempt, e.g. ``stream_idle_timeout``."""
if isinstance(exc, asyncio.CancelledError) and exc.args:
message = str(exc.args[0])
if message.startswith(CANCEL_REASON_PREFIX):
return message.removeprefix(CANCEL_REASON_PREFIX)
return None
def _abandonment_message(exc: BaseException) -> str | None:
if isinstance(exc, asyncio.CancelledError):
return "attempt cancelled before the reply was consumed (stream idle timeout or shutdown)"
reason = f": {exc.args[0]}" if exc.args else ""
return f"attempt cancelled before the reply was consumed{reason}"
if isinstance(exc, GeneratorExit):
return "stream closed by the caller before it finished"
return None
+26 -6
View File
@@ -39,11 +39,13 @@ def _dedupe_model_settings(
model_name=model_name,
force_required_tool_choice=False,
request_timeout=request_timeout,
# The main model's headers apply only when dedupe falls back to the main
# model; a dedicated dedupe model may route to another provider, which
# must never receive the main endpoint's credentials. A dedicated model
# gets its own DEDUPE_LLM_EXTRA_HEADERS instead.
# The main model's headers and body fields apply only when dedupe falls
# back to the main model; a dedicated dedupe model may route to another
# provider, which must never receive the main endpoint's credentials or
# identifiers. A dedicated model gets its own DEDUPE_LLM_EXTRA_HEADERS
# and DEDUPE_LLM_EXTRA_BODY instead.
extra_headers=dedupe.extra_headers if dedupe.model else llm.extra_headers,
extra_body=dedupe.extra_body if dedupe.model else llm.extra_body,
has_tools=False,
)
@@ -298,7 +300,15 @@ def _parse_dedupe_response(content: str) -> dict[str, Any]:
raise ValueError(f"No JSON object found in dedupe response: {content[:500]}")
parsed = json.loads(text[start : end + 1])
duplicate_id = str(parsed.get("duplicate_id") or "")[:64]
is_duplicate = parsed.get("is_duplicate")
if not isinstance(is_duplicate, bool):
raise TypeError("is_duplicate must be a boolean")
duplicate_id = parsed.get("duplicate_id") or ""
if is_duplicate and (not isinstance(duplicate_id, str) or not duplicate_id):
raise ValueError("A duplicate decision must name an existing report ID")
# Keep the ID verbatim: truncating or stripping it could turn an unknown ID
# into a different, existing one before the membership check below.
duplicate_id = str(duplicate_id)
reason = str(parsed.get("reason") or "")[:500]
try:
confidence = float(parsed.get("confidence", 0.0))
@@ -306,13 +316,22 @@ def _parse_dedupe_response(content: str) -> dict[str, Any]:
confidence = 0.0
return {
"is_duplicate": bool(parsed.get("is_duplicate", False)),
"is_duplicate": is_duplicate,
"duplicate_id": duplicate_id,
"confidence": confidence,
"reason": reason,
}
def _validate_duplicate_reference(
result: dict[str, Any], existing_reports: list[dict[str, Any]]
) -> None:
if result["is_duplicate"] and not any(
report.get("id") == result["duplicate_id"] for report in existing_reports
):
raise ValueError("Duplicate decision refers to an unknown report ID")
def _extract_text(response: ModelResponse) -> str:
parts: list[str] = []
for item in response.output:
@@ -395,6 +414,7 @@ async def check_duplicate(
}
result = _parse_dedupe_response(content)
_validate_duplicate_reference(result, existing_reports)
logger.info(
"Deduplication check: is_duplicate=%s, confidence=%.2f, reason=%s",
+43
View File
@@ -28,6 +28,8 @@ from strix.telemetry import posthog, scarf
if TYPE_CHECKING:
from agents.usage import Usage
from strix.llm.request_log import LlmRequestEvent
logger = logging.getLogger(__name__)
@@ -1019,6 +1021,47 @@ class StreamedOpenRouterCosts:
streamed_openrouter_costs = StreamedOpenRouterCosts()
def record_llm_request(event: "LlmRequestEvent") -> None:
"""Request-log sink: each provider attempt's timing, kept in run.json's ``llm_requests``."""
report_state = get_global_report_state()
if report_state is None:
return
details = event.details or {}
report_state.run_record.setdefault("llm_requests", []).append(
{
"started_at": event.started_at.isoformat(),
"agent_id": event.agent_id,
"model": event.model,
"provider": details.get("upstream_provider") or event.provider,
"outcome": event.outcome,
"status": event.status_code,
"duration_ms": event.duration_ms,
"ttft_ms": event.time_to_first_token_ms,
"input_tokens": event.input_tokens,
"cached_tokens": event.cached_input_tokens,
"output_tokens": event.output_tokens,
"retry_attempt": event.retry_attempt,
}
)
def record_tool_call(
*, started_at: datetime, agent_id: str | None, tool: str, duration_ms: int
) -> None:
"""Each tool call's timing, kept in run.json's ``tool_calls``."""
report_state = get_global_report_state()
if report_state is None:
return
report_state.run_record.setdefault("tool_calls", []).append(
{
"started_at": started_at.isoformat(),
"agent_id": agent_id,
"tool": tool,
"duration_ms": duration_ms,
}
)
def record_openrouter_provider(provider: Any, usage: Any) -> None:
"""Tally which upstream provider served a stream, from its final usage chunk.
+51 -2
View File
@@ -8,11 +8,38 @@ from typing import Any
from agents.usage import Usage, deserialize_usage, serialize_usage
from pydantic import BaseModel, TypeAdapter, ValidationError
from strix.config.loader import load_settings
from strix.report.pricing import resolve_litellm_model
logger = logging.getLogger(__name__)
Prices = tuple[float | None, float | None, float | None]
def _configured_prices() -> tuple[str | None, Prices | None]:
llm = load_settings().llm
prices = (llm.input_cost_per_mtok, llm.output_cost_per_mtok, llm.cached_input_cost_per_mtok)
if all(price is None for price in prices):
return None, None
return (llm.model or "").strip() or None, prices
def _configured_usage_cost(usage: Usage, prices: Prices) -> float:
input_rate, output_rate, cached_rate = prices
input_tokens = max(0, _int_or_zero(getattr(usage, "input_tokens", 0)))
cached_details = getattr(usage, "input_tokens_details", None)
cached_tokens = max(0, _int_or_zero(getattr(cached_details, "cached_tokens", 0)))
cached_tokens = min(cached_tokens, input_tokens)
output_tokens = max(0, _int_or_zero(getattr(usage, "output_tokens", 0)))
billable_input = input_tokens - cached_tokens
cost = (
billable_input * (input_rate or 0)
+ cached_tokens * (cached_rate if cached_rate is not None else input_rate or 0)
+ output_tokens * (output_rate or 0)
) / 1_000_000
return _round_cost(cost)
class ProviderUsage(BaseModel):
"""Running totals for one upstream provider OpenRouter routed calls to."""
@@ -37,6 +64,7 @@ class LLMUsageLedger:
self._total_usage = Usage()
self._agent_usage: dict[str, Usage] = {}
self._agent_metadata: dict[str, dict[str, str]] = {}
self._agent_estimated_cost: dict[str, float] = {}
self._observed_cost = 0.0
self._estimated_cost = 0.0
self._has_observed_cost = False
@@ -73,6 +101,9 @@ class LLMUsageLedger:
estimated = _estimate_litellm_cost(usage, model)
if estimated:
self._estimated_cost += estimated
self._agent_estimated_cost[normalized_agent_id] = (
self._agent_estimated_cost.get(normalized_agent_id, 0.0) + estimated
)
return True
@@ -114,11 +145,23 @@ class LLMUsageLedger:
def total_cost(self) -> float:
if self.zero_cost:
return 0.0
model, prices = _configured_prices()
if prices is not None:
return _round_cost(
sum(self._configured_agent_cost(aid, model, prices) for aid in self._agent_usage)
)
return _round_cost(self._observed_cost if self._has_observed_cost else self._estimated_cost)
def _configured_agent_cost(self, agent_id: str, model: str | None, prices: Prices) -> float:
if self._agent_metadata.get(agent_id, {}).get("model") == model:
return _configured_usage_cost(self._agent_usage[agent_id], prices)
return self._agent_estimated_cost.get(agent_id, 0.0)
def to_record(self) -> dict[str, Any]:
record = serialize_usage(self._total_usage)
record["cost"] = self.total_cost
model, prices = _configured_prices()
total_cost = self.total_cost
record["cost"] = total_cost
record["providers"] = {name: tally.model_dump() for name, tally in self._providers.items()}
record["agents"] = []
@@ -128,7 +171,11 @@ 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
self._configured_agent_cost(agent_id, model, prices)
if prices is not None and not self.zero_cost
else total_cost * (agent_tokens[agent_id] / total_tokens)
if total_tokens
else 0.0
)
agent_record = serialize_usage(usage)
@@ -148,6 +195,7 @@ class LLMUsageLedger:
self._total_usage = Usage()
self._agent_usage.clear()
self._agent_metadata.clear()
self._agent_estimated_cost.clear()
self._observed_cost = 0.0
self._estimated_cost = 0.0
self._has_observed_cost = False
@@ -183,6 +231,7 @@ class LLMUsageLedger:
logger.exception("Failed to hydrate llm_usage for agent %s", agent_id)
self._agent_usage[agent_id] = Usage()
self._agent_estimated_cost[agent_id] = _float_or_zero(raw_agent.get("cost"))
metadata: dict[str, str] = {}
agent_name = raw_agent.get("agent_name")
model = raw_agent.get("model")
+255
View File
@@ -0,0 +1,255 @@
from __future__ import annotations
import asyncio
import contextlib
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any
from agents.sandbox.session.manager import Instrumentation
from agents.sandbox.session.sandbox_session import SandboxSession
from agents.sandbox.session.sandbox_session_state import SandboxSessionState
from agents.sandbox.snapshot import NoopSnapshot
if TYPE_CHECKING:
import io
from collections.abc import Sequence
from agents.sandbox.entries import BaseEntry
from agents.sandbox.manifest import Manifest
from agents.sandbox.materialization import MaterializedFile
from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
from agents.sandbox.types import ExecResult, ExposedPortEndpoint, User
logger = logging.getLogger(__name__)
class _PendingSessionState(SandboxSessionState):
type: str = "pending"
class DeferredSandboxSession(SandboxSession):
def __init__(
self,
bring_up: asyncio.Task[tuple[Any, BaseSandboxSession]],
*,
manifest: Manifest,
exposed_ports: tuple[int, ...] = (),
) -> None:
self._bring_up = bring_up
self._pending_state: SandboxSessionState = _PendingSessionState(
snapshot=NoopSnapshot(id="pending"),
manifest=manifest,
exposed_ports=exposed_ports,
)
self._pending_limits: Any = None
self._pending_archive_limits: Any = None
self._resolved: BaseSandboxSession | None = None
self._client: Any = None
self._instrumentation: Instrumentation = Instrumentation()
self._seq = 0
@property
def pending(self) -> bool:
return self._resolved is None
@property
def bring_up(self) -> asyncio.Task[tuple[Any, BaseSandboxSession]]:
return self._bring_up
async def resolve(self) -> BaseSandboxSession:
if self._resolved is not None:
return self._resolved
client, session = await asyncio.shield(self._bring_up)
self._adopt(client, session)
return session
def _adopt(self, client: Any, session: BaseSandboxSession) -> None:
if self._resolved is not None:
return
self._client = client
self._resolved = session
self._inner = session
if self._pending_limits is not None:
session._set_concurrency_limits(self._pending_limits)
if self._pending_archive_limits is not None:
session._set_archive_limits(self._pending_archive_limits)
if self._pending_state.manifest != session.state.manifest:
session.state = session.state.model_copy(
update={"manifest": self._pending_state.manifest}
)
@property
def client(self) -> Any:
return self._client
@property
def state(self) -> SandboxSessionState:
if self._resolved is None:
return self._pending_state
return self._resolved.state
@state.setter
def state(self, value: SandboxSessionState) -> None:
if self._resolved is None:
self._pending_state = value
else:
self._resolved.state = value
def _set_concurrency_limits(self, limits: Any) -> None:
if self._resolved is None:
self._pending_limits = limits
else:
self._resolved._set_concurrency_limits(limits)
def _set_archive_limits(self, limits: Any) -> None:
if self._resolved is None:
self._pending_archive_limits = limits
else:
self._resolved._set_archive_limits(limits)
def _manifest_base_dir(self) -> Path:
if self._resolved is None:
return Path.cwd()
return self._resolved._manifest_base_dir()
def normalize_path(self, path: Path | str, *, for_write: bool = False) -> Path:
if self._resolved is None:
return Path(path)
return self._resolved.normalize_path(path, for_write=for_write)
def register_persist_workspace_skip_path(self, path: Path | str) -> Path:
if self._resolved is None:
return Path(path)
return self._resolved.register_persist_workspace_skip_path(path)
def supports_pty(self) -> bool:
return self._resolved.supports_pty() if self._resolved is not None else False
async def running(self) -> bool:
if self._resolved is None:
if not self._bring_up.done():
return True
return not self._bring_up.cancelled() and self._bring_up.exception() is None
return await self._resolved.running()
async def start(self) -> None:
if self._resolved is None:
return
await self._resolved.start()
async def stop(self) -> None:
if self._resolved is None:
return
await self._resolved.stop()
async def shutdown(self) -> None:
if self._resolved is None:
return
await self._resolved.shutdown()
async def aclose(self) -> None:
if self._resolved is None:
return
await self._resolved.aclose()
async def _validate_manifest_application(self, *, only_ephemeral: bool = False) -> None:
session = await self.resolve()
await session._validate_manifest_application(only_ephemeral=only_ephemeral)
async def apply_manifest(self, *, only_ephemeral: bool = False) -> Any:
session = await self.resolve()
return await session.apply_manifest(only_ephemeral=only_ephemeral)
async def _apply_entry_batch(
self,
entries: Sequence[tuple[Path, BaseEntry]],
*,
base_dir: Path,
) -> list[MaterializedFile]:
del base_dir
session = await self.resolve()
return await session._apply_entry_batch(entries, base_dir=session._manifest_base_dir())
async def exec(self, *command: str | Path, **kwargs: Any) -> ExecResult:
await self.resolve()
return await super().exec(*command, **kwargs)
async def ls(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().ls(*args, **kwargs)
async def rm(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().rm(*args, **kwargs)
async def mkdir(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().mkdir(*args, **kwargs)
async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase:
await self.resolve()
return await super().read(path, user=user)
async def write(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().write(*args, **kwargs)
async def resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
await self.resolve()
return await super().resolve_exposed_port(port)
async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
session = await self.resolve()
return await session._resolve_exposed_port(port)
async def pty_exec_start(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().pty_exec_start(*args, **kwargs)
async def pty_write_stdin(self, *args: Any, **kwargs: Any) -> Any:
await self.resolve()
return await super().pty_write_stdin(*args, **kwargs)
async def pty_terminate_all(self) -> None:
if self._resolved is None:
return
await super().pty_terminate_all()
async def persist_workspace(self) -> io.IOBase:
await self.resolve()
return await super().persist_workspace()
async def hydrate_workspace(self, data: io.IOBase) -> None:
await self.resolve()
await super().hydrate_workspace(data)
async def teardown(self) -> None:
if self._resolved is None:
self._bring_up.cancel()
with contextlib.suppress(BaseException):
client, session = await self._bring_up
self._adopt(client, session)
if self._resolved is None:
return
await self._client.delete(self._resolved)
class DeferredSandboxClient:
def __init__(self, session: DeferredSandboxSession) -> None:
self._session = session
def __getattr__(self, name: str) -> Any:
client = self._session.client
if client is None:
raise AttributeError(f"sandbox client unavailable before bring-up finishes ({name})")
return getattr(client, name)
async def delete(self, session: Any) -> None:
if isinstance(session, DeferredSandboxSession):
await session.teardown()
return
real = await self._session.resolve()
await self._session.client.delete(real if session is self._session else session)
+28
View File
@@ -66,6 +66,33 @@ def _apply_sandbox_network(create_kwargs: dict[str, Any]) -> None:
create_kwargs.pop("ports", None)
_SANDBOX_DEVICES_ENV = "STRIX_DOCKER_SANDBOX_DEVICES"
def _sandbox_devices() -> list[str]:
raw = os.environ.get(_SANDBOX_DEVICES_ENV, "")
return [item.strip() for item in raw.split(",") if item.strip()]
def _apply_sandbox_devices(create_kwargs: dict[str, Any]) -> None:
"""Pass the host devices listed in ``STRIX_DOCKER_SANDBOX_DEVICES`` through.
Comma-separated device paths, each mapped to the same path in the container
with ``rwm`` access. A tunnel brought up inside the sandbox (OpenVPN,
Tailscale) needs ``/dev/net/tun``: Docker's default device cgroup denies it
even with ``NET_ADMIN``. Unset means no extra devices.
"""
devices = _sandbox_devices()
if not devices:
return
existing = [str(item) for item in (create_kwargs.get("devices") or [])]
for path in devices:
spec = f"{path}:{path}:rwm"
if spec not in existing and path not in existing:
existing.append(spec)
create_kwargs["devices"] = existing
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."""
@@ -233,6 +260,7 @@ class StrixDockerSandboxClient(DockerSandboxClient):
extra_hosts["host.docker.internal"] = "host-gateway"
_apply_sandbox_network(create_kwargs)
_apply_sandbox_devices(create_kwargs)
_apply_resource_limits(create_kwargs)
_apply_log_limits(create_kwargs)
_apply_run_labels(create_kwargs)
+93 -32
View File
@@ -3,6 +3,8 @@
from __future__ import annotations
import asyncio
import contextlib
import inspect
import io
import logging
import os
@@ -19,9 +21,12 @@ from strix.config import load_settings
from strix.runtime.backends import backend_supports_bind_mounts, get_backend
from strix.runtime.caido_bootstrap import bootstrap_caido
from strix.runtime.caido_handle import CaidoBootstrapHandle
from strix.runtime.deferred_session import DeferredSandboxClient, DeferredSandboxSession
if TYPE_CHECKING:
from collections.abc import Awaitable
from agents.sandbox.session import BaseSandboxSession
from strix.runtime.status import StatusSink
@@ -255,13 +260,22 @@ def _gitdir_from_pointer(git_file: Path) -> Path | None:
return None
async def _resolve_extra_file_archive(
extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None,
local_sources: list[dict[str, Any]],
) -> bytes | None:
files = await extra_files if inspect.isawaitable(extra_files) else extra_files
return build_extra_file_archive(files, local_sources) if files else None
async def create_or_reuse(
scan_id: str,
*,
image: str,
local_sources: list[dict[str, Any]],
extra_files: list[dict[str, Any]] | None = None,
extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None = None,
status_sink: StatusSink | None = None,
defer: bool = False,
) -> dict[str, Any]:
"""Return the existing session bundle for ``scan_id`` or create a new one.
@@ -271,7 +285,9 @@ async def create_or_reuse(
Each ``extra_files`` entry (``{"workspace_path": "/workspace/<rel>",
"content": bytes | str}``) lands as a regular, agent-writable file at its
``workspace_path`` on every backend: the files are uploaded as one archive
and unpacked inside the sandbox right after bring-up.
and unpacked inside the sandbox right after bring-up. ``extra_files`` may
also be an awaitable resolving to that list: provisioning starts at once and
the files are awaited only when the sandbox is ready to receive them.
"""
def report(phase: str) -> None:
@@ -292,9 +308,6 @@ async def create_or_reuse(
else:
bind_mounts = []
entries = build_manifest_entries(local_sources)
extra_file_archive = (
build_extra_file_archive(extra_files, local_sources) if extra_files else None
)
# Caido runs as an in-container sidecar; HTTP(S) traffic from any
# process started via ``session.exec`` (the SDK's Shell tool, etc.)
@@ -323,41 +336,69 @@ async def create_or_reuse(
backend_name,
image,
)
report("Starting sandbox container")
client, session = await backend(
image=image,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
)
if extra_file_archive is not None:
report("Placing workspace files")
async def _bring_up() -> tuple[Any, Any]:
report("Starting sandbox container")
client, session = await _create_or_discard(
backend(
image=image,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
)
)
try:
await stage_extra_files(session, extra_file_archive)
extra_file_archive = await _resolve_extra_file_archive(extra_files, local_sources)
if extra_file_archive is not None:
report("Placing workspace files")
await stage_extra_files(session, extra_file_archive)
except BaseException:
await _discard_session(client, session)
raise
return client, session
report("Setting up the proxy")
caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
scheme = "https" if caido_endpoint.tls else "http"
host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
logger.debug("Caido host endpoint resolved: %s", host_caido_url)
async def _caido_url(session: Any) -> str:
report("Setting up the proxy")
caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
scheme = "https" if caido_endpoint.tls else "http"
host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
logger.debug("Caido host endpoint resolved: %s", host_caido_url)
return host_caido_url
# The Caido login + project setup polls the guest for a couple of seconds
# and nothing needs the client before the first proxy tool call, so it
# runs concurrently with the rest of scan start; consumers resolve the
# handle at first use (see CaidoBootstrapHandle).
caido_client = CaidoBootstrapHandle(
asyncio.create_task(
bootstrap_caido(
session,
host_url=host_caido_url,
container_url=container_caido_url,
),
name=f"caido-bootstrap-{scan_id}",
def _caido(session: Any, host_caido_url: str) -> Any:
return bootstrap_caido(
session,
host_url=host_caido_url,
container_url=container_caido_url,
)
if defer:
bring_up = asyncio.create_task(_bring_up(), name=f"sandbox-bring-up-{scan_id}")
deferred = DeferredSandboxSession(
bring_up,
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
)
async def _caido_when_up() -> Any:
session = await deferred.resolve()
return await _caido(session, await _caido_url(session))
bundle = {
"client": DeferredSandboxClient(deferred),
"session": deferred,
"caido_client": CaidoBootstrapHandle(
asyncio.create_task(_caido_when_up(), name=f"caido-bootstrap-{scan_id}")
),
}
_SESSION_CACHE[scan_id] = bundle
logger.info("Sandbox session for scan %s deferred and cached", scan_id)
return bundle
client, session = await _bring_up()
host_caido_url = await _caido_url(session)
caido_client = CaidoBootstrapHandle(
asyncio.create_task(_caido(session, host_caido_url), name=f"caido-bootstrap-{scan_id}")
)
bundle = {
@@ -370,6 +411,26 @@ async def create_or_reuse(
return bundle
_DISCARD_TASKS: set[asyncio.Task[None]] = set()
async def _create_or_discard(create: Awaitable[tuple[Any, Any]]) -> tuple[Any, Any]:
pending = asyncio.ensure_future(create)
try:
return await asyncio.shield(pending)
except asyncio.CancelledError:
discard = asyncio.create_task(_discard_when_created(pending))
_DISCARD_TASKS.add(discard)
discard.add_done_callback(_DISCARD_TASKS.discard)
raise
async def _discard_when_created(create: asyncio.Future[tuple[Any, Any]]) -> None:
with contextlib.suppress(BaseException):
client, session = await create
await _discard_session(client, session)
async def _discard_session(client: Any, session: Any) -> None:
"""Best-effort teardown of a session that never made it into the cache."""
try:
+1 -1
View File
@@ -40,7 +40,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
- `source_aware_sast` (custom): opengrep/AST/secrets/supply-chain static triage workflow
- `dependency_cve_scanning` (custom): trivy-based SCA workflow for reporting known dependency CVEs via `create_dependency_report`
- `npx_confusion` (custom): npx/npm exec/bunx fallback and adjacent package-runner identity confusion, with runner-specific registry and reporting gates
- `semantic_confusion` (vulnerabilities): cross-boundary parser, normalization, and representation mismatch analysis
+32 -3
View File
@@ -1,11 +1,12 @@
import logging
import re
from collections import Counter
from collections.abc import Iterator
from collections.abc import Iterator, Mapping
from pathlib import Path
from typing import TypeGuard
import yaml
from jinja2 import Environment, StrictUndefined
from strix.utils.resource_paths import get_strix_resource_path
@@ -13,6 +14,15 @@ from strix.utils.resource_paths import get_strix_resource_path
logger = logging.getLogger(__name__)
_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n(?P<body>.*?)\n---\s*\n", re.DOTALL)
# Only skills with `template: jinja` in their frontmatter are rendered; the rest hold
# literal `{{ }}` payloads.
_SKILL_TEMPLATES = Environment(
autoescape=False, # noqa: S701 # nosec B701 - prompts, not HTML
undefined=StrictUndefined,
trim_blocks=True,
lstrip_blocks=True,
keep_trailing_newline=True,
)
_INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordination", "analysis"})
_ROOT_SKILL_CATEGORY = "root"
@@ -261,7 +271,22 @@ def _candidate_skill_files(skill_name: str) -> list[Path]:
return _bare_skill_files(skill_name)
def load_skills(skill_names: list[str]) -> dict[str, str]:
def skill_turn_limit(skill_names: list[str], turn_limits: Mapping[str, int]) -> int | None:
"""Largest of ``turn_limits`` (keyed by skill name) among these skills, if any."""
files = {path for name in skill_names for path in _candidate_skill_files(name)}
return max(
(
limit
for name, limit in turn_limits.items()
if files.intersection(_candidate_skill_files(name))
),
default=None,
)
def load_skills(
skill_names: list[str], *, supports_images: bool = True, coverage_tools: bool = True
) -> dict[str, str]:
"""Load skill markdown bodies (frontmatter stripped) by name.
Skill files live at ``strix/skills/<category>/<name>.md`` (or any
@@ -291,7 +316,11 @@ def load_skills(skill_names: list[str]) -> dict[str, str]:
continue
var_name = skill_name.split("/")[-1]
_, skill_body = _parse_skill_content(content, file_path)
metadata, skill_body = _parse_skill_content(content, file_path)
if metadata.get("template") == "jinja":
skill_body = _SKILL_TEMPLATES.from_string(skill_body).render(
supports_images=supports_images, coverage_tools=coverage_tools
)
skill_content[var_name] = skill_body
logger.debug("Loaded skill: %s -> %s", skill_name, var_name)
_track_skill_loaded(var_name, file_path)
+9 -3
View File
@@ -1,6 +1,7 @@
---
name: counterevidence
description: Closure discipline for security findings — what counts as proof of safety, what does not, and how to record an unresolved candidate instead of silently dropping it
template: jinja
---
# Counterevidence and Closure Discipline
@@ -20,7 +21,7 @@ state, and "I moved on" is not one of them.
**1. `confirmed`** — you have a working PoC or, in white-box, a complete
source → control → sink → impact trace plus evidence the path is
reachable. File it with `create_vulnerability_report`.
reachable. File it as a vulnerability report.
**2. `ruled_out`** — you can name the **specific control** that makes the
code safe, at a specific location, and you have checked that the control
@@ -32,8 +33,12 @@ that sentence, you are not in `ruled_out`.
**3. `open_proof_gap`** — the candidate is plausible, you could not
confirm it, and you also could not name a control that rules it out. This
{% if coverage_tools %}
is a legitimate, expected outcome. Record it with
`record_coverage(outcome="needs_follow_up")`, carry it up in
{% else %}
is a legitimate, expected outcome. Carry it up in
{% endif %}
`agent_finish(open_items=[...])`, and reflect it in `counterevidence` /
`confidence_rationale` if you file a related report. Do **not** convert
it to `ruled_out` to tidy up your worklist.
@@ -97,6 +102,7 @@ What ships and what is reachable is what matters.
reduces severity — it does not make the finding unreal. Downgrade it;
do not delete it.
{% if coverage_tools %}
## Recording Closure
Closure is only useful if it is written down. Every surface you assess
@@ -125,6 +131,7 @@ named control does not cover the path you just found goes back to
previous state is kept as history, so correcting the record costs
nothing and leaving it wrong costs a finding.
{% endif %}
## What DOES Rule Out a Candidate
- You executed the attack and it demonstrably failed, and you understand
@@ -144,8 +151,7 @@ endpoint is broken/unreachable for unrelated reasons".
## Before You File a Report
Run this pass on every finding before calling
`create_vulnerability_report`:
Run this pass on every finding before filing it:
1. **Argue the other side.** Spend real effort building the strongest
case that this is *not* exploitable, or not as severe as you think.
@@ -1,6 +1,7 @@
---
name: source_aware_discovery
description: Enumeration discipline for reading code — which locations to keep as separate candidates, which safe siblings prove nothing, and the per-family sweeps that are routinely missed
template: jinja
---
# Source-Aware Discovery
@@ -206,6 +207,10 @@ Not worth it: "this could use more validation" with no path, style and
maintainability complaints, and cosmetic variants of a candidate you
already opened.
{% if coverage_tools %}
Keep reading until no distinct plausible candidate remains — then record
what you swept with `record_coverage`, including the families that came
back clean.
{% else %}
Keep reading until no distinct plausible candidate remains.
{% endif %}
+10
View File
@@ -1,6 +1,7 @@
---
name: root-agent
description: Orchestration layer that coordinates specialized subagents for security assessments
template: jinja
---
# Root Agent
@@ -35,10 +36,12 @@ Every scan needs one shared answer to "who is the attacker here, and what are th
Either way you write it with the least information anyone on this scan will ever have, so expect it to be wrong somewhere. Subagents correct it with `amend_threat_model`, which appends an attributed addendum instead of overwriting — expect many of these on a black-box run, as authenticating, pivoting between roles, and reaching internal surfaces is exactly what turns inference into fact. Read the amendments back before you write the final report: an agent telling you a boundary you called trusted is attacker-reachable is a finding about your model, not a note. Only call `save_threat_model` again to fold accumulated amendments into the body; it replaces the document and clears them.
{% if coverage_tools %}
## Reconcile Coverage Before Finishing
Coverage entries are shared and mutable. Before `finish_scan`, list the `needs_follow_up` rows: each one is either work you still owe or a row somebody already resolved without updating. Assign the former to a subagent and have it call `update_coverage` on the existing entry rather than recording a second one — a stale open item sitting next to its own resolution is worse than either alone.
{% endif %}
## Agent Architecture
Structure agents by function:
@@ -55,6 +58,13 @@ Structure agents by function:
- Business logic flaws
- Infrastructure vulnerabilities
**Sensitive Browser Confirmations**
- For every browser workflow containing a security-sensitive or consequential user action—whether authenticated or unauthenticated—assign `browser_security` to the agent testing that workflow or create a dedicated UI-redressing agent. This includes actions that change protected state, grant authority, submit credentials, expose sensitive data, or cross a trust boundary.
- Its task must cover both iframe and cross-window/DoubleClickjacking paths; framing protections close only iframe coverage.
- A workflow crossing origins or ports remains eligible when every involved origin is an explicit scan target. Same-origin policy blocks DOM reads, not top-level navigation, popups, opener relationships, focus changes, or window closing; cross-origin separation alone does not close UI-redressing coverage.
- Never expand authorization from a link alone. If a required origin is not an explicit target, leave the path `needs_follow_up` and do not claim clickjacking or the related workflow is cleared; scope expansion must come from scan configuration.
- Before accepting `no_issue_found` or `ruled_out` for cross-window UI redressing, reject closure evidence that cites popup-as-tab behavior, ignored geometry, unavailable OS-level input, or only a manual control click; keep the exact two-click path `needs_follow_up`. A successfully executed protected consent, approval, or grant must be impact-calibrated, not dismissed because it is idempotent, pre-seeded, or issues no immediate token or data.
**Exploitation and Validation**
- Proof-of-concept development
- Impact demonstration
@@ -14,22 +14,22 @@ Increase white-box coverage by combining source-aware triage with dynamic valida
## Recommended Workflow
1. Build a quick source map before deep exploitation, including at least one AST-structural pass (`sg` or `tree-sitter`) scoped to relevant paths.
- For `sg` baseline, derive `sg-targets.txt` from `semgrep.json` scope first (`paths.scanned`, fallback to unique `results[].path`) and run `xargs ... sg run` on that list.
- Only fall back to path heuristics when semgrep scope is unavailable.
- For `sg` baseline, derive `sg-targets.txt` from `opengrep.json` scope first (`paths.scanned`, fallback to unique `results[].path`) and run `xargs ... sg run` on that list.
- Only fall back to path heuristics when opengrep scope is unavailable.
2. Run first-pass static triage to rank high-risk paths.
3. Use triage outputs to prioritize dynamic PoC validation.
4. Keep findings evidence-driven: no report without validation.
## Source-Aware Triage Stack
- `semgrep`: fast security-first triage and custom pattern scans
- `opengrep`: fast security-first triage and custom pattern scans
- `ast-grep` (`sg`): structural pattern hunting and targeted repo mapping
- `tree-sitter`: syntax-aware parsing support for symbol and route extraction
- `gitleaks` + `trufflehog`: complementary secret detection (working tree and history coverage)
- `trivy fs`: dependency, misconfiguration, license, and secret checks
Coverage target per repository:
- one `semgrep` pass
- one `opengrep` pass
- one AST structural pass (`sg` and/or `tree-sitter`)
- one secrets pass (`gitleaks` and/or `trufflehog`)
- one `trivy fs` pass
+14 -3
View File
@@ -115,7 +115,8 @@ not just "upgrade <vulnerable pkg> to <fixed>".
### Usage / reachability analysis (required for every dependency CVE)
For every CVE you are about to report, run a static usage analysis and record
the result in the structured `reachability` + `reachability_evidence` fields.
the result in the structured `reachability` + `reachability_evidence` fields,
and report every usage site you find as a `code_locations` entry (see Reporting).
The level is an **evidence ladder, never an exploitability verdict** — claim
only what you proved, and cite the proof. It never changes severity (that is
`advisory_cvss` alone); it exists so the reader can prioritize.
@@ -220,8 +221,18 @@ findings and rejects empty PoC fields):
GHSA, look up the mapped CVE; if there is genuinely no CVE, do not report it
with this tool.
- There are no PoC fields — `create_dependency_report` does not take
`poc_description` / `poc_script_code` / `code_locations`. The proof lives in
`description` and `technical_analysis` (scanner output + advisory).
`poc_description` / `poc_script_code`. The proof lives in `description` and
`technical_analysis` (scanner output + advisory).
- **You MUST report usage sites in `code_locations`.** Add one entry for each
import of the package and each call of an affected symbol that your usage
analysis found. Every `file:line` you cite in `reachability_evidence` must
also be a `code_locations` entry. Also add the line in `manifest_path` that
pins the vulnerable version. Each entry needs the repo-relative `file`, the
exact `start_line` / `end_line`, the verbatim `snippet` at those lines, and a
short `label` (e.g. "imports js-yaml", "calls yaml.load on CLI input"). Read
the real file to confirm the lines; never guess. When the package is
`not_imported`, the manifest line is the only entry. If an entry carries
`fix_before` / `fix_after`, also fill `fix_verification`.
- **Always fill the structured dependency fields** (they power the dedicated
dependency-report card; do not leave them only in free-text):
- `package_name` — `PkgName` (required).
+14 -13
View File
@@ -1,6 +1,6 @@
---
name: source-aware-sast
description: Practical source-aware SAST and AST playbook for semgrep, ast-grep, gitleaks, and trivy fs
description: Practical source-aware SAST and AST playbook for opengrep, ast-grep, gitleaks, and trivy fs
---
# Source-Aware SAST Playbook
@@ -23,19 +23,19 @@ Run this baseline once per repository before deep narrowing:
ART=/workspace/.source-aware
mkdir -p "$ART"
semgrep scan --config p/default --config p/golang --config p/secrets \
--metrics=off --json --output "$ART/semgrep.json" .
# Build deterministic AST targets from semgrep scope (no hardcoded path guessing)
opengrep scan --config /opt/opengrep-rules --taint-intrafile \
--json --output "$ART/opengrep.json" .
# Build deterministic AST targets from opengrep scope (no hardcoded path guessing)
python3 - <<'PY'
import json
from pathlib import Path
art = Path("/workspace/.source-aware")
semgrep_json = art / "semgrep.json"
opengrep_json = art / "opengrep.json"
targets_file = art / "sg-targets.txt"
try:
data = json.loads(semgrep_json.read_text(encoding="utf-8"))
data = json.loads(opengrep_json.read_text(encoding="utf-8"))
except Exception:
targets_file.write_text("", encoding="utf-8")
raise
@@ -63,17 +63,18 @@ trivy fs --scanners vuln,misconfig --timeout 30m --offline-scan \
--format json --output "$ART/trivy-fs.json" . || true
```
## Semgrep First Pass
## Opengrep First Pass
Use Semgrep as the default static triage pass:
Use Opengrep as the default static triage pass. Rules are bundled offline in `/opt/opengrep-rules` (one directory per language, plus `generic/secrets`):
```bash
# Preferred deterministic profile set (works with --metrics=off)
semgrep scan --config p/default --config p/golang --config p/secrets \
--metrics=off --json --output /workspace/.source-aware/semgrep.json .
# Full bundled ruleset
opengrep scan --config /opt/opengrep-rules --taint-intrafile \
--json --output /workspace/.source-aware/opengrep.json .
# If you choose auto config, do not combine it with --metrics=off
semgrep scan --config auto --json --output /workspace/.source-aware/semgrep-auto.json .
# Narrow to the target stack when the full set is slow or noisy
opengrep scan --config /opt/opengrep-rules/python --config /opt/opengrep-rules/generic/secrets \
--taint-intrafile --json --output /workspace/.source-aware/opengrep-python.json .
```
If diff scope is active, restrict to changed files first, then expand only when needed.
+2 -2
View File
@@ -205,10 +205,10 @@ Jinja2 backend without autoescape: `{{7*7}}`, RCE gadgets if sandbox misconfigur
## Tooling
Static analysis is the fastest way to reach the sinks above in white-box scope. The sandbox ships `python`/`pipx`, `semgrep`, `bandit`, `ast-grep`, and `ripgrep`.
Static analysis is the fastest way to reach the sinks above in white-box scope. The sandbox ships `python`/`pipx`, `opengrep`, `bandit`, `ast-grep`, and `ripgrep`.
- **bandit** (preinstalled) — Python security linter; flags `mark_safe`, `extra()`, `RawSQL`, `subprocess`, weak crypto, hardcoded secrets: `bandit -r . -ll`
- **semgrep** (preinstalled) with the Django ruleset — higher-signal than bandit for framework-specific bugs (`.extra()`, `RawSQL`, `|safe`, `csrf_exempt`, `ALLOWED_HOSTS=['*']`): `semgrep --config p/django .`
- **opengrep** (preinstalled) with the bundled Django rules — higher-signal than bandit for framework-specific bugs (`.extra()`, `RawSQL`, `|safe`, `csrf_exempt`, `ALLOWED_HOSTS=['*']`): `opengrep scan --config /opt/opengrep-rules/python/django .`
- **pip-audit** (PyPA) — dependency CVE scanner for known-vuln Django/DRF/simplejwt versions: `pipx install pip-audit && pip-audit -r requirements.txt`
- **ast-grep** (preinstalled) — quick structural grep for risky calls without a full SAST run: `ast-grep run -p 'mark_safe($X)' -l python`
+1 -1
View File
@@ -15,7 +15,7 @@ Thorough understanding before exploitation. Test every parameter, every endpoint
**Whitebox (source available)**
- Map every file, module, and code path in the repository
- Start with broad source-aware triage (`semgrep`, `ast-grep`, `gitleaks`, `trufflehog`, `trivy fs`) and use outputs to drive deep review
- Start with broad source-aware triage (`opengrep`, `ast-grep`, `gitleaks`, `trufflehog`, `trivy fs`) and use outputs to drive deep review
- Execute at least one structural AST pass (`sg` and/or Tree-sitter) per repository and store artifacts for reuse
- Keep AST artifacts bounded and query-driven (target relevant paths/sinks first; avoid whole-repo generic function dumps)
- Use syntax-aware parsing (Tree-sitter tooling) to improve symbol, route, and sink extraction quality
+8
View File
@@ -1,6 +1,7 @@
---
name: diff
description: Methodology for diff-scoped review of a pull request, commit, or branch — what counts as in scope, how far to follow a change, and what not to report
template: jinja
---
# Diff-Scoped Review
@@ -69,9 +70,14 @@ When it is not, the closure rules apply unchanged: a complete
source → control → sink → impact trace through the changed code is
reportable at reduced confidence, with the missing runtime proof named in
`confidence_rationale`. A candidate you can neither confirm nor rule out
{% if coverage_tools %}
with a named control is an `open_proof_gap` — record it as
`needs_follow_up` coverage rather than dropping it because the
environment was inconvenient.
{% else %}
with a named control is an `open_proof_gap` — carry it up as an open
item rather than dropping it because the environment was inconvenient.
{% endif %}
## Reporting
@@ -80,7 +86,9 @@ plainly which part of the diff introduced or exposed it. A reviewer
reading your report next to the diff should be able to see the connection
without re-deriving your analysis.
{% if coverage_tools %}
Record coverage per changed component, not per changed file — a
formatting-only file and a rewritten auth module are not equal rows.
{% endif %}
State which changed areas you reviewed and cleared, so the author knows
what a clean result actually covered.
+1 -1
View File
@@ -15,7 +15,7 @@ Optimize for fast feedback on critical security issues. Skip exhaustive enumerat
**Whitebox (source available)**
- Focus on recent changes: git diffs, new commits, modified files—these are most likely to contain fresh bugs
- Run a fast static triage on changed files first (`semgrep`, then targeted `sg` queries)
- Run a fast static triage on changed files first (`opengrep`, then targeted `sg` queries)
- Run at least one lightweight AST pass (`sg` or Tree-sitter) so structural mapping is not skipped
- Keep AST commands tightly scoped to changed or high-risk paths; avoid broad repository-wide pattern dumps
- Run quick secret and dependency checks (`gitleaks`, `trufflehog`, `trivy fs`) scoped to changed areas when possible
+1 -1
View File
@@ -15,7 +15,7 @@ Systematic testing across the full attack surface. Understand the application be
**Whitebox (source available)**
- Map codebase structure: modules, entry points, routing
- Run `semgrep` first-pass triage to prioritize risky flows before deep manual review
- Run `opengrep` first-pass triage to prioritize risky flows before deep manual review
- Run at least one AST-structural mapping pass (`sg` and/or Tree-sitter), then use outputs for route, sink, and trust-boundary mapping
- Keep AST output bounded to relevant paths and hypotheses; avoid whole-repo generic function dumps
- Identify architecture pattern (MVC, microservices, monolith)
+15 -6
View File
@@ -1,6 +1,7 @@
---
name: agent_browser
description: agent-browser CLI for headless Chrome via shell. Snapshot-and-ref workflow, click/fill/extract, screenshots, multi-tab, multi-session, network mocking. Pre-installed in the sandbox; invoke via exec_command.
template: jinja
---
@@ -260,8 +261,9 @@ only for simple expressions.
### Screenshot
`agent-browser screenshot` writes a PNG to disk in the sandbox. The
shell command alone does **not** put the image into your context —
`agent-browser screenshot` writes a PNG to disk in the sandbox.
{% if supports_images %}
The shell command alone does **not** put the image into your context —
chain it with the SDK ``view_image`` tool to actually see it:
```bash
@@ -275,6 +277,12 @@ the full path on stdout — pass that to ``view_image``). If you need a
specific filename, keep it inside that directory or a sibling hidden
dir under ``/workspace``. Never write screenshots to ``/tmp`` —
``view_image`` rejects anything outside the workspace root.
{% else %}
You are running on a text-only model and cannot view images. Screenshots
are only useful to you as a saved file (for example, login evidence).
Drive the page from `snapshot -i` refs, `eval` for any DOM/JS state you
need to read, and `text @ref` / `get text` for content extraction.
{% endif %}
```bash
agent-browser screenshot # path printed on stdout
@@ -283,6 +291,7 @@ agent-browser screenshot --full # full scroll height
agent-browser screenshot --annotate # numbered labels + legend keyed to snapshot refs
```
{% if supports_images %}
`--annotate` is designed for multimodal models: each label `[N]` maps
to ref `@eN`. Take the annotated screenshot, then ``view_image`` it,
and you can correlate visual layout with snapshot refs.
@@ -294,10 +303,10 @@ layout questions, captchas, custom widgets where the accessibility
tree is incomplete).
If ``view_image`` errors back at you (rejected image, "vision not
supported", or similar), you are running on a text-only model — stop
calling it and stop taking screenshots. Drive the page entirely from
`snapshot -i` refs, `eval` for any DOM/JS state you need to read, and
`text @ref` / `get text` for content extraction.
supported", or similar), you can drive the page from `snapshot -i`
refs, `eval` for any DOM/JS state you need to read, and `text @ref`
or `get text` for content extraction.
{% endif %}
### Handle multiple pages via tabs
+69
View File
@@ -0,0 +1,69 @@
---
name: opengrep
description: Exact Opengrep CLI structure, bundled offline rules, scoped rule selection, taint mode, and automation-safe output patterns.
---
# Opengrep CLI Playbook
Official docs:
- https://github.com/opengrep/opengrep
- https://github.com/opengrep/opengrep-rules
Canonical syntax:
`opengrep scan [flags] [targets]`
Bundled rules:
- `/opt/opengrep-rules` holds the full ruleset, one directory per language (`python`, `javascript`, `typescript`, `go`, `java`, `php`, `ruby`, …) plus `generic/secrets`, `dockerfile`, `terraform`, `yaml`, and `strix` (Strix-authored rules, e.g. GitHub-owned actions not pinned to a commit SHA)
- Framework rules sit under the language (`python/django`, `python/flask`, `javascript/express`, …)
High-signal flags:
- `--config <path>` rule file or directory; repeat for several
- `--taint-intrafile` intra-file inter-procedural taint (follows data across functions in one file)
- `--json` / `--sarif` output format
- `--output <file>` write findings to file
- `--severity <level>` filter by severity (`INFO`, `WARNING`, `ERROR`)
- `--error` return non-zero exit when findings exist
- `--quiet` suppress progress noise
- `--jobs <n>` parallel workers
- `--timeout <seconds>` per-rule, per-file timeout
- `--exclude <pattern>` / `--include <pattern>` path filters
- `--exclude-rule <rule_id>` suppress specific rule
- `--baseline-commit <sha>` only report findings introduced after baseline
Agent-safe baseline for automation:
`opengrep scan --config /opt/opengrep-rules --taint-intrafile --json --output opengrep.json --quiet --jobs 4 --timeout 20 /workspace`
Common patterns:
- Full bundled scan:
`opengrep scan --config /opt/opengrep-rules --json --output opengrep.json --quiet /workspace`
- High-severity focused pass:
`opengrep scan --config /opt/opengrep-rules --severity ERROR --json --output opengrep_high.json --quiet /workspace`
- SARIF output:
`opengrep scan --config /opt/opengrep-rules --sarif --output opengrep.sarif --quiet /workspace`
- Language- or framework-specific rules:
`opengrep scan --config /opt/opengrep-rules/python --config /opt/opengrep-rules/generic/secrets --json --output opengrep_python.json --quiet /workspace`
- Scoped directory scan:
`opengrep scan --config /opt/opengrep-rules --json --output opengrep_api.json --quiet /workspace/services/api`
- Custom rule:
`opengrep scan --config ./rules/my-rule.yaml --json --output opengrep_custom.json --quiet /workspace`
Critical correctness rules:
- Always provide an explicit `--config` from `/opt/opengrep-rules` or a local rule file. Registry shorthands (`p/...`, `auto`) fetch rules from the network; do not use them.
- Opengrep has no telemetry; there is no `--metrics` flag, and passing one fails the scan.
- Prefer `--json --output <file>` or `--sarif --output <file>` for machine-readable downstream processing. JSON keeps `paths.scanned` and `results[].path`.
- Keep the target path explicit; use an absolute or clearly scoped workspace path instead of `.` when possible.
Usage rules:
- Start with the full `/opt/opengrep-rules` set unless the task clearly calls for a narrower language or framework directory.
- Add `--taint-intrafile` for injection-class hunting (SQLi, command injection, SSRF, path traversal) in supported languages.
- Use `--quiet` in automation to reduce noisy logs.
- Use `--jobs` and `--timeout` explicitly for reproducible runtime behavior.
- `PartialParsing` errors on templates are normal; they do not invalidate the run.
Failure recovery:
- If scans are too slow, narrow the target path and the rule directories before changing engine settings.
- If scans time out, increase `--timeout` modestly or lower `--jobs`.
- If output is too broad, scope `--config`, add `--severity`, or exclude known irrelevant paths.
If uncertain, query web_search with:
`site:github.com/opengrep/opengrep <flag>`
-72
View File
@@ -1,72 +0,0 @@
---
name: semgrep
description: Exact Semgrep CLI structure, metrics-off scanning, scoped ruleset selection, and automation-safe output patterns.
---
# Semgrep CLI Playbook
Official docs:
- https://semgrep.dev/docs/cli-reference
- https://semgrep.dev/docs/getting-started/cli
- https://semgrep.dev/docs/semgrep-code/semgrep-pro-engine-intro
Canonical syntax:
`semgrep scan [flags]`
High-signal flags:
- `--config <rule_or_ruleset>` ruleset, registry pack, local rule file, or directory
- `--metrics=off` disable telemetry and metrics reporting
- `--json` JSON output
- `--sarif` SARIF output
- `--output <file>` write findings to file
- `--severity <level>` filter by severity
- `--error` return non-zero exit when findings exist
- `--quiet` suppress progress noise
- `--jobs <n>` parallel workers
- `--timeout <seconds>` per-file timeout
- `--exclude <pattern>` exclude path pattern
- `--include <pattern>` include path pattern
- `--exclude-rule <rule_id>` suppress specific rule
- `--baseline-commit <sha>` only report findings introduced after baseline
- `--pro` enable Pro engine if available
- `--oss-only` force OSS engine only
Agent-safe baseline for automation:
`semgrep scan --config p/default --metrics=off --json --output semgrep.json --quiet --jobs 4 --timeout 20 /workspace`
Common patterns:
- Default security scan:
`semgrep scan --config p/default --metrics=off --json --output semgrep.json --quiet /workspace`
- High-severity focused pass:
`semgrep scan --config p/default --severity ERROR --metrics=off --json --output semgrep_high.json --quiet /workspace`
- OWASP-oriented scan:
`semgrep scan --config p/owasp-top-ten --metrics=off --sarif --output semgrep.sarif --quiet /workspace`
- Language- or framework-specific rules:
`semgrep scan --config p/python --config p/secrets --metrics=off --json --output semgrep_python.json --quiet /workspace`
- Scoped directory scan:
`semgrep scan --config p/default --metrics=off --json --output semgrep_api.json --quiet /workspace/services/api`
- Pro engine check or run:
`semgrep scan --config p/default --pro --metrics=off --json --output semgrep_pro.json --quiet /workspace`
Critical correctness rules:
- Always include `--metrics=off`; Semgrep sends telemetry by default.
- Always provide an explicit `--config`; do not rely on vague or implied defaults.
- Prefer `--json --output <file>` or `--sarif --output <file>` for machine-readable downstream processing.
- Keep the target path explicit; use an absolute or clearly scoped workspace path instead of `.` when possible.
- If Pro availability matters, check it explicitly with a bounded command before assuming cross-file analysis exists.
Usage rules:
- Start with `p/default` unless the task clearly calls for a narrower pack.
- Add focused packs such as `p/secrets`, `p/python`, or `p/javascript` only when they match the target stack.
- Use `--quiet` in automation to reduce noisy logs.
- Use `--jobs` and `--timeout` explicitly for reproducible runtime behavior.
- Do not use `-h`/`--help` for routine operation unless absolutely necessary.
Failure recovery:
- If scans are too slow, narrow the target path and reduce the active rulesets before changing engine settings.
- If scans time out, increase `--timeout` modestly or lower `--jobs`.
- If output is too broad, scope `--config`, add `--severity`, or exclude known irrelevant paths.
- If Pro mode fails, rerun with `--oss-only` or without `--pro` and note the loss of cross-file coverage.
If uncertain, query web_search with:
`site:semgrep.dev semgrep <flag> cli`
@@ -1,6 +1,6 @@
---
name: browser-security
description: Browser-internals security testing for browsing-context relationships, postMessage, client-side path traversal, XS-Leaks, service workers, Web Workers, navigation behavior, CSP interactions, caches, and cross-origin state machines
description: Browser-internals security testing for clickjacking, UI redressing, DoubleClickjacking, browsing-context relationships, postMessage, client-side path traversal, XS-Leaks, service workers, Web Workers, navigation behavior, CSP interactions, caches, and cross-origin state machines
---
# Browser Security
@@ -43,6 +43,15 @@ Draw the context graph. Security checks on `event.origin`, `event.source`, or a
- Check whether a blocked intermediate frame leaves a useful browsing-context relationship intact.
- Use random per-flow names or `_blank` with `noopener` where an opener relationship is unnecessary.
### Clickjacking and UI Redressing
- **UI-redressing closure gate:** Never mark clickjacking as ruled out from `X-Frame-Options` or CSP `frame-ancestors` alone. Before closure, test both frame embedding and cross-window/popup retargeting.
- Do not treat a different origin or port as protection against UI redressing. Same-origin policy blocks DOM access, not top-level navigation, `window.open()`, opener relationships, focus changes, or window closing. Test only explicitly authorized targets; if a required origin is outside scope, keep the path `needs_follow_up` and do not claim the vulnerability class is cleared.
- For framed cases, prove the overlay and sensitive click. For DoubleClickjacking, use two real browser clicks and trace click 1 → window/focus/navigation/positioning change → click 2 → sensitive control. Never substitute scripted `.click()` calls.
- Reproduce in the victim state required by the workflow—authenticated when necessary—with realistic user activation and stable geometry; capture window bounds and timing.
- **Evidence-limited closure:** Verify a repeatable sensitive server-side state change with a recording or deterministic event trace and a negative control. If the exact sequence cannot be tested, record `needs_follow_up`, not `ruled_out` or `no_issue_found`. Popup-as-tab behavior, ignored geometry, unavailable OS-level input, or a manual control click are not counterevidence. Close the path only when a named mitigation blocks it, or when geometry is honored and both physical clicks are delivered without a sensitive action.
- **Impact floor:** Successful execution of a protected consent, approval, or grant is an integrity impact even when idempotent, pre-seeded, or followed by no immediate token or data access. Use downstream capability to calibrate severity, not erase the demonstrated protected action.
### Client-Side Path Traversal
Trace the complete source-to-request pipeline:
@@ -146,7 +155,7 @@ Also inspect DevTools network initiators, service workers, storage, CSP violatio
### Source Review
- Search for `postMessage`, message listeners, `window.open`, named targets, opener/parent access, frame creation, and sandbox attributes.
- Search for `postMessage`, message listeners, `window.open`, named targets, opener/parent access, frame creation, `dblclick`, `mousedown`/`mouseup`, focus/blur, move/resize, navigation handlers, and sandbox attributes.
- Search for router parameter APIs flowing into `fetch`, Axios, navigation, or HTML rendering.
- Search for service-worker registration, Cache API writes, worker constructors, Blob URLs, and dynamic imports.
- Search for raw HTML sinks and trust escape hatches in every supported frontend framework.
@@ -161,7 +170,7 @@ Also inspect DevTools network initiators, service workers, storage, CSP violatio
5. **Prove the primitive** - Data transfer, path change, state oracle, cache modification, or context capture.
6. **Escalate deliberately** - Chain to a privileged action, sensitive disclosure, SSRF, or executable DOM sink.
7. **Cross-browser check** - At minimum record Chromium/Firefox/Safari applicability when the primitive is browser-specific.
8. **State interaction requirements** - Click, drag, popup permission, timing window, login state, and visual deception.
8. **State interaction requirements** - Record each physical click or drag, popup permission state, active context, focus, geometry, navigation, timing window, login state, and visual deception.
## Validation
@@ -171,7 +180,7 @@ Also inspect DevTools network initiators, service workers, storage, CSP violatio
4. For XS-Leaks, repeat randomized success/failure trials and quantify separation and noise.
5. For workers/caches, show which later context consumes the modified data.
6. For client-side traversal, capture the final network request and the security-relevant response/action.
7. For interaction-dependent chains, provide a screen recording or deterministic event trace.
7. For interaction-dependent chains, show both click targets, intermediate context changes, and the resulting sensitive state change in a recording or deterministic event trace.
## False Positives
+30 -6
View File
@@ -1,6 +1,6 @@
---
name: xss
description: XSS testing covering reflected, stored, and DOM-based vectors with CSP bypass techniques
description: XSS and HTML injection testing covering reflected, stored, DOM-based, and normalization-dependent vectors
---
# XSS
@@ -11,6 +11,7 @@ Cross-site scripting persists because context, parser, and framework edges are c
**Types**
- Reflected, stored, and DOM-based XSS across web/mobile/desktop shells
- Scriptless HTML injection and second-order render paths
**Contexts**
- HTML, attribute, URL, JS, CSS, SVG/MathML, Markdown, PDF
@@ -20,6 +21,7 @@ Cross-site scripting persists because context, parser, and framework edges are c
**Defenses to Bypass**
- CSP/Trusted Types, DOMPurify, framework auto-escaping
- Decode, normalization, sanitization, and encoding order
## Injection Points
@@ -57,6 +59,24 @@ Cross-site scripting persists because context, parser, and framework edges are c
Distinguish HTML-string input from live DOM objects sanitized with `IN_PLACE`. Resolve the deployed DOMPurify build and check relevant upstream advisories; a controlled observable `nodeName` is one live-node bypass mechanism. Trace same-origin foreign nodes or `adoptNode()` into this mode before selecting the bypass ([example advisory](https://github.com/cure53/DOMPurify/security/advisories/GHSA-x4vx-rjvf-j5p4)).
### Scriptless HTML Injection
Do not require JavaScript execution before you report HTML injection. Prove that the browser creates attacker-controlled markup and a concrete security effect.
Test base or form target changes, external requests, deceptive content, DOM clobbering through `id` or `name`, and dangling markup. Report XSS only when you prove a script-capable execution path. Do not dismiss HTML injection only because CSP blocks scripts.
### Unicode Normalization
Test normalization only when the target performs it. NFKC can fold fullwidth delimiters such as `<` and `>` into HTML syntax.
Map validation, normalization, sanitization, encoding, and parsing in order. Start with the harmless canary `<h1 id="strix-nfkc">marker</h1>`.
Report a bypass only when normalization creates dangerous syntax that reaches an unprotected sink. Load `semantic_confusion` when components interpret the value differently.
### Second-Order Rendering
Follow stored values into previews, administrative pages, email, exports, and hydration. Record which stage transforms and parses the value.
### DOM XSS
**Sources**
@@ -166,7 +186,7 @@ Keep a compact set tuned per context:
## Testing Methodology
1. **Identify sources** - URL/query/hash/referrer, postMessage, storage, WebSocket, server JSON
2. **Trace to sinks** - Map data flow from source to sink
2. **Trace transformations and sinks** - Record decoding, normalization, sanitization, reparsing, and the final sink
3. **Classify context** - HTML node, attribute, URL, script block, event handler, JS eval-like, CSS, SVG
4. **Assess defenses** - Output encoding, sanitizer, CSP, Trusted Types, DOMPurify config
5. **Craft payloads** - Minimal payloads per context with encoding/whitespace/casing variants
@@ -175,9 +195,11 @@ Keep a compact set tuned per context:
## Validation
1. Provide minimal payload and context (sink type) with before/after DOM or network evidence
2. Demonstrate cross-browser execution where relevant or explain parser-specific behavior
3. Show bypass of stated defenses (sanitizer settings, CSP/Trusted Types) with proof
4. Quantify impact beyond alert: data accessed, action performed, persistence achieved
2. Record the original and transformed value when decoding or normalization affects the result
3. Demonstrate cross-browser execution where relevant or explain parser-specific behavior
4. Show bypass of stated defenses (sanitizer settings, CSP/Trusted Types) with proof
5. Separate HTML injection from XSS and state the proven capability
6. Quantify impact beyond alert: data accessed, action performed, persistence achieved
## False Positives
@@ -185,6 +207,8 @@ Keep a compact set tuned per context:
- CSP with nonces/hashes and no inline/event handlers
- Trusted Types enforced on sinks; DOMPurify in strict mode with URI allowlists
- Scriptable contexts disabled (no HTML pass-through, safe URL schemes enforced)
- The target does not perform the claimed normalization or decode
- A normalized value is safely encoded before the final sink
## Impact
@@ -207,4 +231,4 @@ Keep a compact set tuned per context:
## Summary
Context + sink decide execution. Encode for the exact context, verify at runtime with CSP/Trusted Types, and validate every alternative render path. Small payloads with strong evidence beat payload catalogs.
Context, transformation order, and sink decide whether input becomes text, markup, or executable code. Prove HTML injection separately from XSS, and validate every render path. Encode for the exact context and verify at runtime with CSP/Trusted Types. Small payloads with strong evidence beat payload catalogs.
+12
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import contextlib
import functools
import logging
import os
import sys
@@ -78,6 +79,16 @@ class _StdoutQuietFilter(logging.Filter):
)
class _PtyThresholdFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
return "PTY process count reached warning threshold" not in record.getMessage()
@functools.cache # install once; tests reset via ``cache_clear()``
def _silence_pty_threshold_warning() -> None:
logging.getLogger("agents.sandbox.sandboxes.docker").addFilter(_PtyThresholdFilter())
def configure_dependency_logging() -> None:
"""Quiet dependency logging/warnings that obscure Strix scan logs."""
litellm = sys.modules.get("litellm")
@@ -89,6 +100,7 @@ def configure_dependency_logging() -> None:
logging.getLogger("asyncio").propagate = False
warnings.filterwarnings("ignore", category=RuntimeWarning, module="asyncio")
_route_unraisable_to_log()
_silence_pty_threshold_warning()
def _route_unraisable_to_log() -> None:
+6 -7
View File
@@ -535,9 +535,8 @@ async def create_agent(
success looks like, any constraints. Name the target the
child should call ``get_threat_model`` on, and any shared
state it should build on rather than rediscover — what
recon already mapped, which surfaces are already covered,
which coverage entry it is picking up. A child that is not
told what is already known repeats it.
recon already mapped, which surfaces are already covered.
A child that is not told what is already known repeats it.
inherit_context: Default ``True``. The child receives the
parent's input history as background; only set ``False``
when starting a clean-slate task.
@@ -637,9 +636,8 @@ async def agent_finish(
doing: what did you test, what did you find/confirm/rule out,
what's still open.
**Close out honestly.** Before calling this, every surface you
assessed should have a ``record_coverage`` entry, and anything you
could neither confirm nor rule out belongs in ``open_items`` — an
**Close out honestly.** Before calling this, anything you could
neither confirm nor rule out belongs in ``open_items`` — an
unresolved candidate handed up to the parent is useful, a silently
dropped one is a missed vulnerability. Reporting nothing and
listing no open items asserts the area is clean; only say that if
@@ -683,7 +681,8 @@ async def agent_finish(
{
"success": False,
"error": (
"agent_finish is for subagents. Root/main agents must call finish_scan instead"
"agent_finish is for subagents. Root/main agents must call "
f"{coordinator.root_finish_tool} instead"
),
},
ensure_ascii=False,
+6 -2
View File
@@ -24,12 +24,16 @@ async def load_skill(ctx: RunContextWrapper, skills: list[str]) -> str:
Max 5. Names match the bare files under
``strix/skills/<category>/<name>.md``.
"""
del ctx
requested = list(skills or [])
err = validate_requested_skills(requested)
if err:
return f"load_skill: {err}"
contents = load_skills(requested)
context = ctx.context if isinstance(ctx.context, dict) else {}
contents = load_skills(
requested,
supports_images=context.get("supports_images", True),
coverage_tools=context.get("coverage_tools", True),
)
if not contents:
return "load_skill: no content loaded for requested skills."
sections = [f"## Skill: {name}\n\n{body}" for name, body in contents.items()]
+26 -3
View File
@@ -24,6 +24,7 @@ automatically.
from __future__ import annotations
import asyncio
import json
from typing import TYPE_CHECKING, Any
@@ -46,6 +47,7 @@ def _registry_from_ctx(ctx: RunContextWrapper) -> McpRegistry | None:
_NO_CONNECTIONS = "No MCP connections are configured for this run."
_MCP_DISCOVERY_WAIT_SECONDS = 30.0
def _unknown_connection(connection: str, registry: McpRegistry) -> str:
@@ -61,11 +63,13 @@ def _format_tool(tool: MCPTool) -> str:
@function_tool(timeout=60)
async def list_mcps(ctx: RunContextWrapper) -> dict[str, Any]:
"""List the MCP connections available this run, so you can discover them.
"""List available MCP connections, loading each connection's catalog on first use.
Read-only. Returns one entry per connection with its ``id`` (the exact name
you pass to the other MCP tools), ``name``, ``description``, and
``tool_count``. The response does not include tool schemas. Call
``tool_count`` (null while the catalog is unknown). Waits at most 30 seconds
for catalogs; unfinished catalogs keep loading in the background.
The response does not include tool schemas. Call
``search_mcp_tools`` next. Then call ``get_mcp_tool_schema`` for one selected
tool before you call ``call_mcp``. Returns an empty ``connections`` list when
the run has no MCP connections.
@@ -73,6 +77,23 @@ async def list_mcps(ctx: RunContextWrapper) -> dict[str, Any]:
registry = _registry_from_ctx(ctx)
if registry is None or not registry:
return {"connections": []}
pending = [
entry
for name in registry.names()
if (entry := registry.get(name)) is not None
and entry.state != "catalog_ready"
and (entry.session is None or not entry.session.is_dead)
]
if pending:
await asyncio.wait(
{
asyncio.gather(
*(entry.ensure_catalog() for entry in pending),
return_exceptions=True,
)
},
timeout=_MCP_DISCOVERY_WAIT_SECONDS,
)
dead_by_name = {status.name: status.dead for status in registry.statuses()}
return {
"connections": [
@@ -80,7 +101,7 @@ async def list_mcps(ctx: RunContextWrapper) -> dict[str, Any]:
"id": summary.name,
"name": summary.name,
"description": summary.purpose,
"tool_count": summary.tool_count,
"tool_count": summary.tool_count if summary.state == "catalog_ready" else None,
"dead": dead_by_name.get(summary.name, False),
"state": summary.state,
}
@@ -288,9 +309,11 @@ async def call_mcp(
"Call search_mcp_tools, then get_mcp_tool_schema."
)
session = await entry.ensure_connected()
context = ctx.context if isinstance(ctx.context, dict) else {}
return await session.dispatch(
tool,
arguments or {},
label=namespaced_tool_name(connection, tool),
result_transform=entry.result_transform,
supports_images=context.get("supports_images", True),
)
+16
View File
@@ -32,6 +32,7 @@ from agents.mcp import (
)
from mcp.client.stdio import stdio_client
from mcp.shared._httpx_utils import create_mcp_http_client
from mcp.types import TextContent
from strix.tools.mcp.failures import HttpStatusRecorder
from strix.tools.mcp.session import McpConnectionUnavailableError, SupervisedMcpSession
@@ -183,6 +184,18 @@ def _build_server(config: McpConnectionConfig) -> BuiltMcpServer:
)
def _without_images(result: Any) -> Any:
content = [
TextContent(
type="text", text=f"[{item.mimeType} image omitted: this model cannot view images]"
)
if item.type == "image"
else item
for item in result.content
]
return result.model_copy(update={"content": content})
def _mcp_result_to_tool_output(server: MCPServer, result: Any) -> Any:
"""Serialize a ``CallToolResult`` to a tool output, mirroring the agents SDK.
@@ -217,6 +230,7 @@ async def dispatch_mcp_call(
*,
label: str,
result_transform: ResultTransform | None = None,
supports_images: bool = True,
) -> Any:
"""Run one MCP tool call and convert its result to a tool output.
@@ -233,6 +247,8 @@ async def dispatch_mcp_call(
corrupt the content the agent receives).
"""
result = await server.call_tool(tool_name, arguments)
if not supports_images:
result = _without_images(result)
if result_transform is not None:
return result_transform(label, result.model_dump(mode="json"))
tool_output = _mcp_result_to_tool_output(server, result)
+5 -5
View File
@@ -432,19 +432,19 @@ class McpRegistry:
entry.set_status_sink(sink)
def start_warmup(self, *, max_concurrency: int = 6) -> asyncio.Task[None]:
"""Connect every configured entry in the background with a fixed bound."""
"""Connect and list every configured entry with a fixed concurrency bound."""
if self._warmup_task is not None:
return self._warmup_task
async def warm() -> None:
semaphore = asyncio.Semaphore(max(1, max_concurrency))
async def connect(entry: McpConnectionEntry) -> None:
async def load_catalog(entry: McpConnectionEntry) -> None:
async with semaphore:
with contextlib.suppress(McpConnectionUnavailableError):
await entry.ensure_connected()
with contextlib.suppress(Exception):
await entry.ensure_catalog()
await asyncio.gather(*(connect(entry) for entry in self._entries.values()))
await asyncio.gather(*(load_catalog(entry) for entry in self._entries.values()))
self._warmup_task = asyncio.create_task(warm(), name="mcp-warmup")
return self._warmup_task
+2
View File
@@ -365,6 +365,7 @@ class SupervisedMcpSession:
*,
label: str,
result_transform: ResultTransform | None = None,
supports_images: bool = True,
) -> Any:
"""Run one tool call with bounded retries for transient session failures.
@@ -382,6 +383,7 @@ class SupervisedMcpSession:
arguments,
label=label,
result_transform=result_transform,
supports_images=supports_images,
)
outcome = await self._run_job(job, phase="call")
+31 -9
View File
@@ -13,6 +13,7 @@ from datetime import datetime
from typing import TYPE_CHECKING, Any, Literal
from agents import RunContextWrapper, function_tool
from caido_sdk_client import BaseError
from strix.runtime.caido_handle import CaidoBootstrapHandle
from strix.tools.nullish import clean_optional
@@ -68,6 +69,15 @@ async def _call[T](client: Client, fn: Callable[[Client], Awaitable[T]]) -> T:
return await fn(client)
_REQUEST_EXISTS_QUERY = "query RequestExists($id: ID!) { request(id: $id) { id } }"
_REQUEST_EXISTS_ATTEMPTS = 2
async def _request_exists(client: Client, request_id: str) -> bool:
data = await client.graphql.query(_REQUEST_EXISTS_QUERY, {"id": request_id})
return bool(data) and data.get("request") is not None
async def existing_request_ids(
ctx: RunContextWrapper,
request_ids: list[str],
@@ -80,16 +90,24 @@ async def existing_request_ids(
if client is None:
raise RuntimeError("Caido client is not available")
# Request IDs are not an HTTPQL field. Resolve each ID through the same
# project-bound lookup as view_request rather than constructing a filter.
# Request IDs are not an HTTPQL field, so each one is resolved by id. Only
# the id is selected: pulling the raw request and response bodies just to
# learn whether a row exists is what made this time out on busy projects.
existing: set[str] = set()
for request_id in request_ids:
result = await _call(
client,
functools.partial(caido_api.get_request_with_client, request_id=request_id),
)
if result is not None:
existing.add(str(result.request.id))
for attempt in range(1, _REQUEST_EXISTS_ATTEMPTS + 1):
try:
exists = await _call(
client,
functools.partial(_request_exists, request_id=request_id),
)
except Exception:
if attempt == _REQUEST_EXISTS_ATTEMPTS:
raise
continue
if exists:
existing.add(request_id)
break
return existing
@@ -124,7 +142,11 @@ def _no_client() -> str:
def _err(name: str, exc: Exception) -> str:
logger.exception("%s failed", name)
# Caido-side failures (timeouts, bad filters, ...) are returned to the model, not our bugs.
if isinstance(exc, BaseError):
logger.warning("%s failed", name, exc_info=True)
else:
logger.exception("%s failed", name)
return json.dumps(
{"success": False, "error": f"{name} failed: {exc}"},
ensure_ascii=False,
+43 -12
View File
@@ -222,9 +222,9 @@ def _normalize_http_exchange_ids(raw: Any) -> tuple[list[str] | None, list[str]]
return normalized, errors
_HTTP_EXCHANGE_DROPPED_WARNING = (
"http_exchange_ids were not stored: the proxy project could not be reached to verify "
"them. Attach them with update_vulnerability_report when the proxy responds again."
_HTTP_EXCHANGE_UNVERIFIED_WARNING = (
"http_exchange_ids were stored without verification because the proxy project did "
"not respond in time. Ids that do not exist in the project will not render as evidence."
)
@@ -235,9 +235,9 @@ async def _verify_http_exchange_ids(
"""Verify proxy exchange IDs against the current Caido project.
IDs the project does not know are rejected. When the proxy itself cannot be
queried the IDs are dropped and a warning is returned instead, so a proxy
outage never blocks a finding and unverified IDs are never recorded as
evidence.
queried the IDs are kept and a warning is returned instead: a slow or
unreachable proxy must never strip evidence from a finding, and consumers
resolve each id against the recorded traffic before rendering it.
"""
request_ids, errors = _normalize_http_exchange_ids(raw)
if request_ids is None or errors or not request_ids:
@@ -250,7 +250,7 @@ async def _verify_http_exchange_ids(
"Could not verify HTTP exchange IDs against the current Caido project",
exc_info=True,
)
return None, [], _HTTP_EXCHANGE_DROPPED_WARNING
return request_ids, [], _HTTP_EXCHANGE_UNVERIFIED_WARNING
missing_ids = [request_id for request_id in request_ids if request_id not in existing_ids]
if missing_ids:
@@ -1876,14 +1876,10 @@ def _build_dependency_evidence(
evidence += f"\n\n**Usage analysis:** {label}."
if reachability_evidence and reachability_evidence.strip():
evidence += f" {reachability_evidence.strip()}"
evidence += (
" This is a prioritization signal from static analysis, not a"
" proof of exploitability or of safety."
)
return evidence
async def _do_create_dependency( # noqa: PLR0912
async def _do_create_dependency( # noqa: PLR0912, PLR0915
*,
title: str,
description: str,
@@ -1905,6 +1901,8 @@ async def _do_create_dependency( # noqa: PLR0912
manifest_path: str | None = None,
reachability: str = "unknown",
reachability_evidence: str | None = None,
code_locations: list[dict[str, Any]] | None = None,
fix_verification: str | None = None,
contextual_cvss_breakdown: dict[str, str] | None = None,
contextual_cvss_reasoning: str | None = None,
agent_id: str | None = None,
@@ -1947,6 +1945,16 @@ async def _do_create_dependency( # noqa: PLR0912
if manifest_err:
errors.append(manifest_err)
parsed_locations = _normalize_code_locations(code_locations)
if parsed_locations:
errors.extend(_validate_code_locations(parsed_locations))
elif code_locations:
errors.append(
"code_locations were dropped as unusable - every location needs a relative "
"'file' and an integer 'start_line'"
)
errors.extend(_validate_fix_verification(parsed_locations, fix_verification))
reachability = (reachability or "unknown").strip().lower()
if reachability not in _VALID_REACHABILITY:
errors.append(
@@ -2053,6 +2061,8 @@ async def _do_create_dependency( # noqa: PLR0912
cvss=cvss_score if advisory_cvss is not None else None,
cve=parsed_cve,
cwe=cwe,
code_locations=parsed_locations,
fix_verification=fix_verification,
finding_class="dependency_cve",
dependency_metadata=dependency_metadata,
agent_id=agent_id if isinstance(agent_id, str) else None,
@@ -2107,6 +2117,8 @@ async def create_dependency_report(
dependency_path: str | None = None,
reachability: str = "unknown",
reachability_evidence: str | None = None,
code_locations: list[dict[str, Any]] | None = None,
fix_verification: str | None = None,
contextual_cvss_breakdown: dict[str, str] | None = None,
contextual_cvss_reasoning: str | None = None,
) -> str:
@@ -2217,6 +2229,23 @@ async def create_dependency_report(
is off in production), and say who controls the input. State
it plainly when no entry point reaches the sink — that is the
most useful result a reader can get.
code_locations: Where application code imports or calls the package —
the evidence behind your ``reachability`` claim. List of dicts,
same shape as ``create_vulnerability_report``::
{"file": "src/api/client.ts", "start_line": 14, "end_line": 14,
"snippet": "const x = require('pkg')", "label": "imports it"}
Cite the repo-relative ``file`` and the exact 1-based
``start_line`` where ``snippet`` actually sits — verify against
the real file, never guess. A location may also propose
the fix itself: ``fix_before`` (verbatim current code, e.g. the
manifest pin line) + ``fix_after`` (same lines with the fixed
version) — that combination requires ``fix_verification``.
fix_verification: Required whenever any ``code_locations`` entry
carries ``fix_after``. Same bar as a normal finding: security
closure, bypass review, preserved behavior, and how each was
checked.
contextual_cvss_breakdown: **Required.** Full CVSS v3.1 rating of this
CVE **in this codebase** — the same 8-metric object as
``create_vulnerability_report``'s ``cvss_breakdown``:
@@ -2277,6 +2306,8 @@ async def create_dependency_report(
manifest_path=manifest_path,
reachability=reachability,
reachability_evidence=reachability_evidence,
code_locations=code_locations,
fix_verification=fix_verification,
contextual_cvss_breakdown=contextual_cvss_breakdown,
contextual_cvss_reasoning=contextual_cvss_reasoning,
agent_id=agent_id,
+100
View File
@@ -80,6 +80,94 @@ async def test_wrap_exec_command_preserves_explicit_shell(shell: str) -> None:
assert json.loads(captured["raw_input"])["shell"] == shell
def _running_exec_tool(result: str) -> FunctionTool:
async def invoke(_ctx: Any, _raw_input: str) -> str:
return result
return FunctionTool(
name="exec_command",
description="Runs a command.",
params_json_schema={"type": "object", "properties": {}},
on_invoke_tool=invoke,
)
_RUNNING_RESULT = (
"Chunk ID: abc123\nWall time: 30.0012 seconds\nProcess running with session ID 7\nOutput:\n"
)
def test_wrap_exec_command_describes_the_yield_cap() -> None:
wrapped = factory._wrap_exec_command(_running_exec_tool("ok"))
assert wrapped.description.startswith("Runs a command. The wait for output is capped at 30 s")
assert "cannot be read later" in wrapped.description
assert "tty=true" in wrapped.description
@pytest.mark.asyncio
async def test_wrap_exec_command_flags_a_running_non_tty_process() -> None:
wrapped = factory._wrap_exec_command(_running_exec_tool(_RUNNING_RESULT))
result = await wrapped.on_invoke_tool(
cast("Any", None), json.dumps({"cmd": "sleep 60; curl x", "yield_time_ms": 90000})
)
assert result == (
"Chunk ID: abc123\nWall time: 30.0012 seconds\n"
"Process running with session ID 7\nOutput:\n"
"Note: yield_time_ms=90000 was capped at 30000 (30 s). this non-TTY process cannot be "
"polled and its later output is lost. Rerun it in the background "
"(`cmd > /tmp/out.log 2>&1 &`, then read the log) or with tty=true.\n"
)
@pytest.mark.asyncio
async def test_wrap_exec_command_flags_a_running_non_tty_process_within_the_cap() -> None:
wrapped = factory._wrap_exec_command(_running_exec_tool(_RUNNING_RESULT))
result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "sleep 60"}))
assert "Note: this non-TTY process cannot be polled" in result
assert "was capped" not in result
@pytest.mark.asyncio
async def test_wrap_exec_command_leaves_a_running_tty_process_alone() -> None:
wrapped = factory._wrap_exec_command(_running_exec_tool(_RUNNING_RESULT))
result = await wrapped.on_invoke_tool(
cast("Any", None), json.dumps({"cmd": "python3 -m http.server", "tty": True})
)
assert result == _RUNNING_RESULT
@pytest.mark.asyncio
async def test_wrap_exec_command_leaves_an_exited_process_alone() -> None:
exited = "Chunk ID: abc123\nWall time: 0.1 seconds\nProcess exited with code 0\nOutput:\nhi"
wrapped = factory._wrap_exec_command(_running_exec_tool(exited))
result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "echo hi"}))
assert result == exited
@pytest.mark.asyncio
async def test_wrap_exec_command_ignores_the_running_marker_inside_command_output() -> None:
exited = (
"Chunk ID: abc123\nWall time: 0.2 seconds\nProcess exited with code 0\nOutput:\n"
"Process running with session ID 7\nOutput:\n"
)
wrapped = factory._wrap_exec_command(_running_exec_tool(exited))
result = await wrapped.on_invoke_tool(
cast("Any", None), json.dumps({"cmd": "cat transcript.txt"})
)
assert result == exited
@pytest.mark.asyncio
async def test_responses_filesystem_custom_tool_output_is_bounded() -> None:
async def invoke(_ctx: Any, _inp: str) -> str:
@@ -115,3 +203,15 @@ def test_function_tools_are_result_bounded() -> None:
by_name = {t.name: t for t in agent.tools}
assert getattr(by_name["think"], "_strix_bounded", False) is True
@pytest.mark.parametrize(("supports_images", "has_view_image"), [(True, True), (False, False)])
def test_view_image_is_left_out_for_text_only_models(
supports_images: bool, has_view_image: bool
) -> None:
agent = factory.build_strix_agent(is_root=True, supports_images=supports_images)
filesystem = agent.capabilities[0]
filesystem.bind(cast("Any", object()))
view_image = next(tool for tool in filesystem.tools() if tool.name == "view_image")
assert cast("FunctionTool", view_image).is_enabled is has_view_image
+21 -3
View File
@@ -38,11 +38,26 @@ async def _roundtrip(
_STRING = {"todos": {"type": "string"}}
@pytest.mark.asyncio
async def test_truncated_arguments_short_circuit_with_a_reissue_message() -> None:
captured: dict[str, str] = {}
wrapped = factory._with_coerced_arguments(_capturing_tool(captured, _STRING))
result = await wrapped.on_invoke_tool(cast("Any", None), '{"todos": "a, b')
assert "not executed" in result
assert "Re-issue the call" in result
assert "raw_input" not in captured
_ARRAY = {"tags": {"type": "array", "items": {"type": "string"}}}
_NULLABLE_ARRAY = {
"tags": {"anyOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}]}
}
_OBJECT = {"modifications": {"type": "object"}}
OVER_NESTED = '{"a":' + "[" * 100_000
@pytest.mark.asyncio
@@ -136,12 +151,15 @@ async def test_unknown_and_null_arguments_are_untouched() -> None:
@pytest.mark.asyncio
async def test_non_object_payloads_pass_through_unchanged() -> None:
@pytest.mark.parametrize("arguments", ["not json", '{"n": NaN}', OVER_NESTED])
async def test_non_object_payloads_are_reported_instead_of_invoked(arguments: str) -> None:
captured: dict[str, str] = {}
wrapped = factory._with_coerced_arguments(_capturing_tool(captured, _ARRAY))
assert await wrapped.on_invoke_tool(cast("Any", None), "not json") == "ok"
assert captured["raw_input"] == "not json"
result = await wrapped.on_invoke_tool(cast("Any", None), arguments)
assert result.startswith("probe: the tool call was not executed")
assert "raw_input" not in captured
@pytest.mark.asyncio
+54
View File
@@ -59,6 +59,15 @@ def test_registered_tools_appear_before_lifecycle_tool() -> None:
assert child_names[-2:] == ["extra", "agent_finish"]
def test_root_ends_with_the_given_finish_tool() -> None:
root = factory.build_strix_agent(is_root=True, finish_tool=_tool("finish_pr_review"))
root_names = [t.name for t in root.tools]
assert root_names[-1] == "finish_pr_review"
assert "finish_scan" not in root_names
def test_per_call_extra_tools_stack_with_registry() -> None:
factory.register_agent_tools(_tool("registered"))
@@ -84,6 +93,51 @@ def test_per_call_extra_tools_reject_duplicate_registered_names() -> None:
factory.build_strix_agent(is_root=True, extra_tools=[_tool("same_name")])
def test_report_tool_replaces_create_vulnerability_report() -> None:
for is_root in (True, False):
agent = factory.build_strix_agent(is_root=is_root, report_tool=_tool("file_pr_finding"))
names = [t.name for t in agent.tools]
assert "file_pr_finding" in names
assert "create_vulnerability_report" not in names
assert isinstance(agent.instructions, str)
assert "file_pr_finding" in agent.instructions
assert "create_vulnerability_report" not in agent.instructions
def _descriptions(tools: list[Any]) -> list[str]:
return [t.description for t in tools if isinstance(t, FunctionTool)]
def test_coverage_tools_can_be_turned_off() -> None:
coverage = {"record_coverage", "update_coverage", "list_coverage"}
for is_root in (True, False):
on = factory.build_strix_agent(is_root=is_root, is_whitebox=True)
off = factory.build_strix_agent(is_root=is_root, is_whitebox=True, coverage_tools=False)
assert coverage <= {t.name for t in on.tools}
assert not coverage & {t.name for t in off.tools}
assert isinstance(on.instructions, str)
assert isinstance(off.instructions, str)
assert "- COVERAGE" in on.instructions
assert "## Recording Closure" in on.instructions
assert not any(name in off.instructions for name in coverage)
on_descriptions = "\n".join(_descriptions(on.tools))
off_descriptions = "\n".join(_descriptions(off.tools))
if is_root:
assert "Coverage reconciliation" in on_descriptions
assert "Coverage reconciliation" not in off_descriptions
assert not any(name in off_descriptions for name in coverage)
diff_on = factory.build_strix_agent(is_root=True, is_diff_scoped=True)
diff_off = factory.build_strix_agent(is_root=True, is_diff_scoped=True, coverage_tools=False)
assert isinstance(diff_on.instructions, str)
assert isinstance(diff_off.instructions, str)
assert "Record coverage per changed component" in diff_on.instructions
assert "`needs_follow_up` coverage" in diff_on.instructions
assert "Record coverage per changed component" not in diff_off.instructions
assert "`needs_follow_up` coverage" not in diff_off.instructions
assert not any(name in diff_off.instructions for name in coverage)
def test_instructions_override_is_used_verbatim() -> None:
custom = "You are a scan agent. Follow the provided scope."
+146 -3
View File
@@ -21,7 +21,7 @@ from strix.core.sessions import open_agent_session
if TYPE_CHECKING:
from collections.abc import AsyncIterator, Callable
from collections.abc import AsyncIterator, Awaitable, Callable
from pathlib import Path
@@ -31,6 +31,7 @@ _CALL_LATENCY_S = 0.005
class _FakeLedger:
def __init__(self) -> None:
self.on_pause: Callable[[], Awaitable[None]] | None = None
self.cost = 0.0
self.calls: list[str] = []
self.remaining: dict[str, int] = {}
@@ -80,6 +81,8 @@ class _FakeStream:
await self._hooks.on_llm_start(ctx_wrapper, self._agent, None, input_items)
except BudgetPausedError as exc:
self.run_loop_exception = exc
if self._ledger.on_pause is not None:
await self._ledger.on_pause()
return
self._ledger.warned_inputs.append(input_items)
if self._ledger.gate is not None:
@@ -150,13 +153,16 @@ class _Scan:
monkeypatch: pytest.MonkeyPatch,
*,
max_budget_usd: float | None,
budget_policy: str = "pause",
) -> None:
self.ledger = _FakeLedger()
self.hooks = ReportUsageHooks(
model="test-model", max_budget_usd=max_budget_usd, budget_policy="pause"
model="test-model",
max_budget_usd=max_budget_usd,
budget_policy=budget_policy, # type: ignore[arg-type]
)
self.coordinator = AgentCoordinator()
self.coordinator.set_budget_policy("pause")
self.coordinator.set_budget_policy(budget_policy) # type: ignore[arg-type]
self.coordinator.set_budget_limit_setter(self.hooks.set_max_budget_usd)
monkeypatch.setattr(execution, "Runner", _fake_runner(self.ledger, self.coordinator))
monkeypatch.setattr(execution, "_compact_session", _noop_compact)
@@ -538,3 +544,140 @@ async def test_stop_policy_is_unchanged() -> None:
def test_budget_policy_is_validated() -> None:
with pytest.raises(ValueError, match="budget_policy"):
ReportUsageHooks(model="m", budget_policy="later") # type: ignore[arg-type]
def _pause_flags(coordinator: AgentCoordinator) -> tuple[bool, bool]:
return coordinator.budget_paused, coordinator.operator_paused
@pytest.mark.asyncio
async def test_operator_pause_holds_under_the_stop_policy(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
scan = _Scan(tmp_path, monkeypatch, max_budget_usd=None, budget_policy="stop")
agents = ("root", "child-a", "child-b")
with patch("strix.core.hooks.get_global_report_state", return_value=scan.ledger):
await scan.start_root(calls=6)
await scan.start_child("child-a", calls=6)
await scan.start_child("child-b", calls=6)
await _wait_until(lambda: scan.ledger.cost >= 3.0)
await scan.coordinator.pause_budget()
spent_at_pause = scan.ledger.cost
await _wait_until(lambda: _all_parked(scan.coordinator, *agents))
assert _pause_flags(scan.coordinator) == (True, True)
assert all(not task.done() for task in scan.tasks())
await scan.coordinator.send("child-a", {"from": "root", "content": "status?"})
await scan.coordinator.send("root", {"from": "user", "content": "keep going"})
await asyncio.sleep(0.05)
assert _pause_flags(scan.coordinator) == (True, True)
assert scan.ledger.cost == pytest.approx(spent_at_pause)
assert _all_parked(scan.coordinator, *agents)
woken = await scan.coordinator.resume_budget()
assert sorted(woken) == sorted(agents)
assert _pause_flags(scan.coordinator) == (False, False)
await asyncio.wait_for(asyncio.gather(*scan.tasks(), return_exceptions=True), timeout=5.0)
assert scan.ledger.cost == pytest.approx(18.0)
assert {aid: str(s) for aid, s in scan.coordinator.statuses.items()} == {
"root": "completed",
"child-a": "completed",
"child-b": "completed",
}
root_items = await scan.sessions[0].get_items()
assert any("keep going" in str(item) for item in root_items)
for session in scan.sessions:
session.close()
@pytest.mark.asyncio
async def test_a_resume_that_lands_before_the_park_still_continues(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
scan = _Scan(tmp_path, monkeypatch, max_budget_usd=None, budget_policy="stop")
resumed = 0
async def resume_at_once() -> None:
nonlocal resumed
resumed += 1
await scan.coordinator.resume_budget()
scan.ledger.on_pause = resume_at_once
with patch("strix.core.hooks.get_global_report_state", return_value=scan.ledger):
await scan.start_root(calls=6)
await scan.start_child("child-a", calls=6)
await _wait_until(lambda: scan.ledger.cost >= 2.0)
await scan.coordinator.pause_budget()
await asyncio.wait_for(asyncio.gather(*scan.tasks(), return_exceptions=True), timeout=5.0)
assert resumed >= 1
assert _pause_flags(scan.coordinator) == (False, False)
assert scan.ledger.cost == pytest.approx(12.0)
assert {aid: str(s) for aid, s in scan.coordinator.statuses.items()} == {
"root": "completed",
"child-a": "completed",
}
for session in scan.sessions:
session.close()
@pytest.mark.asyncio
async def test_operator_pause_gates_the_stop_policy_hooks() -> None:
coordinator = AgentCoordinator()
hooks = ReportUsageHooks(model="m", max_budget_usd=None)
assert hooks.budget_policy == "stop"
await coordinator.pause_budget()
with pytest.raises(BudgetPausedError):
await hooks.on_llm_start(_ctx(coordinator), MagicMock(), None, [])
await coordinator.resume_budget()
await hooks.on_llm_start(_ctx(coordinator), MagicMock(), None, [])
@pytest.mark.asyncio
async def test_a_user_message_lifts_only_the_limit_pause(tmp_path: Path) -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
session = open_agent_session("root", tmp_path / "agents.db")
await coordinator.attach_runtime("root", session=session)
extensions: list[int] = []
coordinator.set_budget_extender(lambda: extensions.append(1))
await coordinator.pause_budget()
await coordinator.park_for_budget("root")
assert coordinator.statuses["root"] == "budget_paused"
await coordinator.send("root", {"from": "user", "content": "keep going"})
assert _pause_flags(coordinator) == (True, True)
assert extensions == []
assert coordinator.pending_counts["root"] == 1
await coordinator.resume_budget()
await coordinator.pause_for_budget("root")
await coordinator.send("root", {"from": "user", "content": "keep going"})
assert _pause_flags(coordinator) == (False, False)
assert extensions == [1]
session.close()
@pytest.mark.asyncio
async def test_snapshot_and_reset_keep_an_operator_pause() -> None:
coordinator = AgentCoordinator()
await coordinator.register("root", "strix", parent_id=None)
await coordinator.pause_budget()
snap = await coordinator.snapshot()
restored = AgentCoordinator()
await restored.restore(snap)
assert _pause_flags(restored) == (True, True)
await restored.reset_budget_stops(budget_stopped=False, reserve_stopped=False)
assert _pause_flags(restored) == (True, True)
await restored.resume_budget()
await restored.pause_for_budget("root")
await restored.reset_budget_stops(budget_stopped=False, reserve_stopped=False)
assert _pause_flags(restored) == (False, False)
assert restored.statuses["root"] == "waiting"
+9 -2
View File
@@ -147,7 +147,9 @@ def _patch_budget(monkeypatch: pytest.MonkeyPatch, *, keep_tokens: int, window:
context.auto_compact = True
settings = SimpleNamespace(
context=context,
llm=SimpleNamespace(api_key=None, api_base=None, timeout=1, extra_headers=None),
llm=SimpleNamespace(
api_key=None, api_base=None, timeout=1, extra_headers=None, extra_body=None
),
)
monkeypatch.setattr(compaction, "load_settings", lambda: settings)
@@ -228,7 +230,11 @@ async def test_summarize_routes_through_provider_with_settings(
"load_settings",
lambda: SimpleNamespace(
llm=SimpleNamespace(
api_key=None, api_base=None, timeout=1, extra_headers={"X-Feature-Key": "svc"}
api_key=None,
api_base=None,
timeout=1,
extra_headers={"X-Feature-Key": "svc"},
extra_body={"service_tier": "flex"},
)
),
)
@@ -239,6 +245,7 @@ async def test_summarize_routes_through_provider_with_settings(
assert captured["model"] == "litellm/openai/some-model"
settings = captured["model_settings"]
assert settings.extra_headers == {"X-Feature-Key": "svc"}
assert settings.extra_body == {"service_tier": "flex"}
assert settings.max_tokens == 64
+57
View File
@@ -37,6 +37,12 @@ _LLM_ENV_KEYS = [
"STRIX_RUNTIME_BACKEND",
# TelemetrySettings
"STRIX_TELEMETRY",
# JSON-object settings
"LLM_EXTRA_HEADERS",
"LLM_EXTRA_BODY",
"STRIX_DEDUPE_MODEL",
"DEDUPE_LLM_EXTRA_HEADERS",
"DEDUPE_LLM_EXTRA_BODY",
]
@@ -406,6 +412,57 @@ def test_persist_current_replaces_corrupt_file(
assert json.loads(target.read_text(encoding="utf-8")) == {"env": {"STRIX_LLM": "env-model"}}
@pytest.mark.parametrize(
("env_var", "section", "field"),
[
("LLM_EXTRA_HEADERS", "llm", "extra_headers"),
("LLM_EXTRA_BODY", "llm", "extra_body"),
("DEDUPE_LLM_EXTRA_HEADERS", "dedupe", "extra_headers"),
("DEDUPE_LLM_EXTRA_BODY", "dedupe", "extra_body"),
],
)
def test_persisted_json_object_setting_reloads_without_env(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, env_var: str, section: str, field: str
) -> None:
loader.apply_config_override(tmp_path / "cli-config.json")
monkeypatch.setenv(env_var, '{"X-Tenant": "acme"}')
loader.load_settings()
loader.persist_current()
monkeypatch.delenv(env_var)
loader.apply_config_override(tmp_path / "cli-config.json")
settings = loader.load_settings()
assert getattr(getattr(settings, section), field) == {"X-Tenant": "acme"}
def test_json_object_setting_saved_as_object_still_loads(tmp_path: Path) -> None:
target = tmp_path / "cli-config.json"
target.write_text(
json.dumps({"env": {"LLM_EXTRA_BODY": {"service_tier": "flex"}}}), encoding="utf-8"
)
loader.apply_config_override(target)
assert loader.load_settings().llm.extra_body == {"service_tier": "flex"}
def test_invalid_saved_json_object_setting_is_rejected(tmp_path: Path) -> None:
target = tmp_path / "cli-config.json"
target.write_text(json.dumps({"env": {"LLM_EXTRA_BODY": "not json"}}), encoding="utf-8")
loader.apply_config_override(target)
with pytest.raises(ValidationError):
loader.load_settings()
def test_saved_string_setting_is_not_json_decoded(tmp_path: Path) -> None:
target = tmp_path / "cli-config.json"
target.write_text(json.dumps({"env": {"STRIX_LLM": '{"not": "decoded"}'}}), encoding="utf-8")
loader.apply_config_override(target)
assert loader.load_settings().llm.model == '{"not": "decoded"}'
def test_persist_current_sets_0600_mode(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("STRIX_LLM", "persisted-model")
target = tmp_path / "cli-config.json"
+139
View File
@@ -2,7 +2,9 @@
from __future__ import annotations
import json
import uuid
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock, call, patch
@@ -17,11 +19,13 @@ from strix.config.models import (
_configure_litellm_compatibility,
_install_openrouter_stream_cost_capture,
)
from strix.core.hooks import ReportUsageHooks
from strix.llm import request_log
from strix.report.state import (
ReportState,
litellm_cost_callback,
openrouter_stream_cost,
record_llm_request,
set_global_report_state,
streamed_openrouter_costs,
)
@@ -135,6 +139,7 @@ def test_cost_callback_estimates_cost_with_provider_prefixed_model() -> None:
with (
patch("strix.report.state.get_global_report_state", return_value=report_state),
patch("strix.report.state.resolve_litellm_model", side_effect=lambda model: model),
patch("litellm.completion_cost", side_effect=fake_completion_cost),
):
litellm_cost_callback(kwargs, response)
@@ -389,3 +394,137 @@ def test_openrouter_request_carries_agent_session_id() -> None:
assert body()["session_id"] == session_id
finally:
request_log.reset_call_context(token)
def _openrouter_config() -> Any:
_install_openrouter_stream_cost_capture()
config = ProviderConfigManager.get_provider_chat_config(
model="z-ai/glm-5.3", provider=LlmProviders.OPENROUTER
)
assert config is not None
return config
def test_openrouter_records_upstream_of_stream_and_mid_stream_error() -> None:
handler = _openrouter_config().get_model_response_iterator(
streaming_response=iter([]), sync_stream=True
)
reply = request_log.HttpReply()
token = request_log._http_reply.set(reply)
try:
handler.chunk_parser(
{
"id": "gen-a",
"created": 1,
"model": "z-ai/glm-5.3",
"provider": "Relace",
"choices": [{"index": 0, "delta": {"content": "x"}}],
}
)
assert reply.upstream_provider == "Relace"
assert reply.upstream_error_type is None
with pytest.raises(Exception, match="incomplete tool call"):
handler.chunk_parser(
{
"id": "gen-a",
"provider": "Relace",
"error": {
"code": 400,
"message": "Generation stopped with an incomplete tool call.",
"metadata": {"error_type": "invalid_request"},
},
}
)
finally:
request_log._http_reply.reset(token)
assert reply.upstream_provider == "Relace"
assert reply.upstream_error_type == "invalid_request"
def test_openrouter_records_upstream_of_pre_stream_error() -> None:
body = {
"error": {
"message": "Provider returned error",
"code": 400,
"metadata": {"raw": "...", "provider_name": "InferenceNet"},
}
}
reply = request_log.HttpReply()
token = request_log._http_reply.set(reply)
try:
_openrouter_config().get_error_class(json.dumps(body), 400, {})
_openrouter_config().get_error_class("<html>502 Bad Gateway</html>", 502, {})
finally:
request_log._http_reply.reset(token)
assert reply.upstream_provider == "InferenceNet"
assert reply.upstream_error_type is None
def test_request_timings_land_in_run_record() -> None:
started = datetime(2026, 10, 19, 12, 0, tzinfo=UTC)
event = request_log.LlmRequestEvent(
call_id="call-1",
route="litellm",
provider="openrouter",
model="openrouter/deepseek/deepseek-v4.1-flash",
api_host="openrouter.ai",
streaming=True,
outcome="success",
status_code=200,
provider_request_id=None,
response_id="gen-1",
error_type=None,
error_message=None,
started_at=started,
finished_at=started + timedelta(milliseconds=4200),
duration_ms=4200,
input_tokens=90_000,
output_tokens=600,
cached_input_tokens=85_000,
agent_id="agent-1",
time_to_first_token_ms=900,
details={"upstream_provider": "Together"},
)
state = ReportState.__new__(ReportState)
state.run_record = {}
try:
set_global_report_state(state)
record_llm_request(event)
finally:
set_global_report_state(None)
assert state.run_record["llm_requests"] == [
{
"started_at": "2026-10-19T12:00:00+00:00",
"agent_id": "agent-1",
"model": "openrouter/deepseek/deepseek-v4.1-flash",
"provider": "Together",
"outcome": "success",
"status": 200,
"duration_ms": 4200,
"ttft_ms": 900,
"input_tokens": 90_000,
"cached_tokens": 85_000,
"output_tokens": 600,
"retry_attempt": 0,
}
]
async def test_tool_call_timings_land_in_run_record() -> None:
hooks = ReportUsageHooks(model="m", record_tool_timings=True)
context = SimpleNamespace(tool_call_id="call-1", context={"agent_id": "agent-1"})
tool = SimpleNamespace(name="terminal_execute")
state = ReportState.__new__(ReportState)
state.run_record = {}
try:
set_global_report_state(state)
await hooks.on_tool_start(context, MagicMock(), tool) # type: ignore[arg-type]
await hooks.on_tool_end(context, MagicMock(), tool, "ok") # type: ignore[arg-type]
finally:
set_global_report_state(None)
[entry] = state.run_record["tool_calls"]
assert entry["agent_id"] == "agent-1"
assert entry["tool"] == "terminal_execute"
assert entry["duration_ms"] >= 0
@@ -0,0 +1,22 @@
"""The root agent must follow up open coverage rows before finishing."""
from __future__ import annotations
import pytest
from strix.agents.prompt import render_system_prompt
_FOLLOW_UP = 'calls `list_coverage(outcome="needs_follow_up")` and follows up every row it returns'
_KEEP_GENUINE_GAPS = "stays `needs_follow_up` only when it is genuinely blocked"
@pytest.mark.parametrize("scan_mode", ["quick", "standard", "deep"])
@pytest.mark.parametrize("is_whitebox", [False, True])
@pytest.mark.parametrize("is_root", [False, True])
def test_prompt_follows_up_open_coverage(scan_mode: str, is_whitebox: bool, is_root: bool) -> None:
prompt = render_system_prompt(scan_mode=scan_mode, is_whitebox=is_whitebox, is_root=is_root)
assert prompt
assert _FOLLOW_UP in prompt
assert _KEEP_GENUINE_GAPS in prompt
+159
View File
@@ -0,0 +1,159 @@
"""Only validated model decisions may suppress a finding."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock
import pytest
from agents.items import ModelResponse
from agents.usage import Usage
from openai.types.responses import ResponseOutputMessage, ResponseOutputText
from strix.config import loader
from strix.report import dedupe
from strix.report.state import ReportState, get_global_report_state, set_global_report_state
from strix.tools.reporting.tool import _do_create
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
@pytest.fixture
def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[ReportState]:
monkeypatch.chdir(tmp_path)
previous = get_global_report_state()
state = ReportState(run_name="dedupe-decisions")
set_global_report_state(state)
state.add_vulnerability_report(title="Existing finding", severity="low")
try:
yield state
finally:
set_global_report_state(previous)
def _mock_judge(monkeypatch: pytest.MonkeyPatch, decision: dict[str, Any]) -> AsyncMock:
settings = loader.load_settings().model_copy(deep=True)
settings.llm.model = "openai/synthetic-model"
model = AsyncMock()
model.get_response.return_value = ModelResponse(
output=[
ResponseOutputMessage(
id="msg_synthetic",
type="message",
role="assistant",
status="completed",
content=[
ResponseOutputText(
type="output_text",
annotations=[],
text=json.dumps({"confidence": 0.95, "reason": "Synthetic", **decision}),
)
],
)
],
usage=Usage(),
response_id="synthetic",
)
monkeypatch.setattr(dedupe, "load_settings", lambda: settings)
monkeypatch.setattr(dedupe, "configure_sdk_model_defaults", lambda _: None)
monkeypatch.setattr(dedupe, "resolve_dedupe_model", lambda *_: model)
return model
async def _file_finding() -> dict[str, Any]:
return await _do_create(
title="New finding",
description="The endpoint reflects unencoded input.",
impact="The reflected script executes in a victim's browser.",
target="https://example.invalid",
technical_analysis="Input is interpolated into HTML without encoding.",
poc_description="Open the endpoint with the reflected parameter.",
poc_script_code="print('synthetic reproduction')",
remediation_steps="Encode output for its HTML context.",
evidence="The synthetic response contains the unencoded parameter.",
assumptions="A victim follows the crafted link.",
counterevidence="No encoding or content security policy is present.",
confidence="high",
severity_change_conditions="A strict content security policy would lower the impact.",
fix_effort="low",
cvss_breakdown={
"attack_vector": "N",
"attack_complexity": "L",
"privileges_required": "N",
"user_interaction": "R",
"scope": "U",
"confidentiality": "L",
"integrity": "L",
"availability": "N",
},
endpoint="/search",
method="GET",
cve=None,
cwe="CWE-79",
code_locations=None,
)
@pytest.mark.parametrize("value", ["false", "true", 0, 1, None, [], {}])
def test_dedupe_parser_requires_a_boolean(value: Any) -> None:
with pytest.raises(TypeError, match="is_duplicate"):
dedupe._parse_dedupe_response(json.dumps({"is_duplicate": value}))
@pytest.mark.parametrize(
"decision",
[
{"is_duplicate": "false", "duplicate_id": "vuln-0001"},
{"is_duplicate": "true", "duplicate_id": "vuln-0001"},
{"is_duplicate": 1, "duplicate_id": "vuln-0001"},
{"is_duplicate": True, "duplicate_id": "vuln-9999"},
{"is_duplicate": True, "duplicate_id": ""},
{"is_duplicate": True, "duplicate_id": None},
{"is_duplicate": True, "duplicate_id": ["vuln-0001"]},
{"is_duplicate": True},
],
)
async def test_invalid_dedupe_decision_does_not_suppress_finding(
report_state: ReportState, monkeypatch: pytest.MonkeyPatch, decision: dict[str, Any]
) -> None:
model = _mock_judge(monkeypatch, decision)
result = await _file_finding()
assert result["success"] is True
model.get_response.assert_awaited_once()
saved = json.loads((report_state.get_run_dir() / "vulnerabilities.json").read_text())
assert [report["title"] for report in saved] == ["Existing finding", "New finding"]
assert (report_state.get_run_dir() / "vulnerabilities" / "vuln-0002.md").is_file()
@pytest.mark.parametrize("is_duplicate", [False, True])
async def test_valid_dedupe_decisions_keep_their_meaning(
report_state: ReportState, monkeypatch: pytest.MonkeyPatch, is_duplicate: bool
) -> None:
# An extraneous ID must not turn a valid false decision into a duplicate.
_mock_judge(monkeypatch, {"is_duplicate": is_duplicate, "duplicate_id": "vuln-0001"})
result = await _file_finding()
assert result["success"] is not is_duplicate
saved = json.loads((report_state.get_run_dir() / "vulnerabilities.json").read_text())
assert len(saved) == (1 if is_duplicate else 2)
if is_duplicate:
assert result["duplicate_of"] == "vuln-0001"
async def test_dedupe_id_is_matched_without_truncation(monkeypatch: pytest.MonkeyPatch) -> None:
existing_id = "x" * 64
_mock_judge(monkeypatch, {"is_duplicate": True, "duplicate_id": existing_id + "-unknown"})
result = await dedupe.check_duplicate(
{"title": "New finding"}, [{"id": existing_id, "title": "Existing finding"}]
)
assert result["is_duplicate"] is False
assert "error" in result
+39
View File
@@ -148,3 +148,42 @@ def test_config_file_loads_dedupe_model(
assert settings.dedupe.reasoning_effort == "minimal"
# Main model stays independent of the dedupe override.
assert settings.llm.model == "openai/root"
def test_dedicated_dedupe_model_uses_own_body_fields_not_main(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"safety_identifier": "main"}))
loader._cached = None
try:
dedupe = DedupeSettings(
STRIX_DEDUPE_MODEL="deepseek/cheap",
DEDUPE_LLM_EXTRA_BODY={"service_tier": "flex"},
)
settings = _dedupe_model_settings(dedupe, "deepseek/cheap", 300)
assert settings.extra_body == {"service_tier": "flex"}
finally:
loader._cached = None
def test_dedicated_dedupe_model_gets_no_main_body_fields_by_default(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"safety_identifier": "main"}))
loader._cached = None
try:
dedupe = DedupeSettings(STRIX_DEDUPE_MODEL="deepseek/cheap")
settings = _dedupe_model_settings(dedupe, "deepseek/cheap", 300)
assert settings.extra_body is None
finally:
loader._cached = None
def test_fallback_dedupe_inherits_main_body_fields(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("LLM_EXTRA_BODY", json.dumps({"safety_identifier": "main"}))
loader._cached = None
try:
settings = _dedupe_model_settings(DedupeSettings(), "openai/main-model", 300)
assert settings.extra_body == {"safety_identifier": "main"}
finally:
loader._cached = None
+316
View File
@@ -0,0 +1,316 @@
from __future__ import annotations
import asyncio
import io
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any
import pytest
from agents.run_config import SandboxRunConfig
from agents.sandbox import SandboxAgent
from agents.sandbox.capabilities import Filesystem, Shell
from agents.sandbox.runtime_session_manager import SandboxRuntimeSessionManager
from strix.runtime import session_manager
from strix.runtime.backends import _BACKENDS, _BIND_MOUNT_BACKENDS, register_backend
from strix.runtime.deferred_session import DeferredSandboxClient, DeferredSandboxSession
if TYPE_CHECKING:
from collections.abc import Iterator
@dataclass
class _RuntimeSettings:
backend: str
@dataclass
class _Settings:
runtime: _RuntimeSettings
@dataclass
class _Endpoint:
host: str = "127.0.0.1"
port: int = 8080
tls: bool = False
@dataclass
class _ExecResult:
exit_code: int = 0
stdout: bytes = b""
stderr: bytes = b""
def ok(self) -> bool:
return self.exit_code == 0
class _State:
def __init__(self) -> None:
self.manifest: Any = None
self.session_id = "00000000-0000-4000-8000-000000000001"
def model_copy(self, *, update: dict[str, Any]) -> _State:
copy = _State()
copy.manifest = update.get("manifest", self.manifest)
return copy
class _Session:
def __init__(self) -> None:
self.state = _State()
self.execs: list[tuple[str, ...]] = []
self.writes: list[tuple[Path, bytes]] = []
self.limits: list[Any] = []
def _set_concurrency_limits(self, limits: Any) -> None:
self.limits.append(limits)
def _set_archive_limits(self, limits: Any) -> None:
self.limits.append(limits)
async def running(self) -> bool:
return True
async def resolve_exposed_port(self, _port: int) -> _Endpoint:
return _Endpoint()
async def exec(self, *argv: str, timeout: float | None = None, **_kw: Any) -> _ExecResult:
del timeout
self.execs.append(argv)
return _ExecResult()
async def write(self, path: Path, data: io.IOBase, **_kw: Any) -> None:
self.writes.append((path, data.read()))
class _Client:
def __init__(self) -> None:
self.deleted: list[Any] = []
async def delete(self, session: Any) -> None:
self.deleted.append(session)
async def _no_caido(*_args: Any, **_kwargs: Any) -> None:
return None
class _Backend:
def __init__(self, *, fail: bool = False) -> None:
self.gate = asyncio.Event()
self.calls = 0
self.fail = fail
self.client = _Client()
self.session = _Session()
async def __call__(self, **_kwargs: Any) -> tuple[Any, Any]:
self.calls += 1
await self.gate.wait()
if self.fail:
raise RuntimeError("bring-up exploded")
return self.client, self.session
@pytest.fixture
def backend(monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest) -> Iterator[_Backend]:
name = f"deferred-{request.node.name}"
fake = _Backend(fail="fail" in request.node.name)
register_backend(name, fake, supports_bind_mounts=True)
monkeypatch.setattr(session_manager, "bootstrap_caido", _no_caido)
monkeypatch.setattr(
session_manager, "load_settings", lambda: _Settings(runtime=_RuntimeSettings(backend=name))
)
yield fake
_BACKENDS.pop(name, None)
_BIND_MOUNT_BACKENDS.discard(name)
async def _settle() -> None:
for _ in range(10):
await asyncio.sleep(0)
async def _create(scan_id: str, **kwargs: Any) -> dict[str, Any]:
return await session_manager.create_or_reuse(
scan_id, image="img", local_sources=[], defer=True, **kwargs
)
@pytest.mark.asyncio
async def test_deferred_bundle_returns_before_the_backend_finishes(backend: _Backend) -> None:
scan_id = "deferred-returns"
try:
bundle = await asyncio.wait_for(_create(scan_id), timeout=1)
session = bundle["session"]
assert isinstance(session, DeferredSandboxSession)
assert isinstance(bundle["client"], DeferredSandboxClient)
assert session.pending
await _settle()
assert backend.calls == 1
assert await session.running()
assert session.state.manifest.environment.value["PYTHONUNBUFFERED"] == "1"
again = await session_manager.create_or_reuse(scan_id, image="img", local_sources=[])
assert again is bundle
assert backend.calls == 1
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_first_io_resolves_the_sandbox_then_delegates(backend: _Backend) -> None:
scan_id = "deferred-io"
try:
bundle = await _create(scan_id)
session = bundle["session"]
session._set_concurrency_limits("limits")
exec_task = asyncio.create_task(session.exec("echo", "hi"))
await asyncio.sleep(0.05)
assert not exec_task.done()
backend.gate.set()
result = await asyncio.wait_for(exec_task, timeout=1)
assert result.ok()
assert backend.session.execs == [("echo", "hi")]
assert not session.pending
assert backend.session.limits == ["limits"]
assert backend.session.state.manifest is session.state.manifest
await session.write(Path("/workspace/x"), io.BytesIO(b"data"))
assert backend.session.writes == [(Path("/workspace/x"), b"data")]
finally:
await session_manager.cleanup(scan_id)
assert backend.client.deleted == [backend.session]
@pytest.mark.asyncio
async def test_failed_bring_up_surfaces_on_first_io(backend: _Backend) -> None:
scan_id = "deferred-fail"
try:
bundle = await _create(scan_id)
session = bundle["session"]
backend.gate.set()
with pytest.raises(RuntimeError, match="bring-up exploded"):
await asyncio.wait_for(session.exec("true"), timeout=1)
assert not await session.running()
with pytest.raises(RuntimeError, match="bring-up exploded"):
await bundle["caido_client"].get()
finally:
await session_manager.cleanup(scan_id)
assert backend.client.deleted == []
@pytest.mark.asyncio
async def test_cleanup_while_pending_cancels_the_bring_up(backend: _Backend) -> None:
scan_id = "deferred-cancel"
bundle = await _create(scan_id)
session = bundle["session"]
await asyncio.wait_for(session_manager.cleanup(scan_id), timeout=1)
assert session.bring_up.cancelled()
assert backend.client.deleted == []
assert session_manager._SESSION_CACHE.get(scan_id) is None
backend.gate.set()
await _settle()
assert backend.client.deleted == [backend.session]
@pytest.mark.asyncio
async def test_cleanup_while_pending_ignores_a_failed_late_bring_up(backend: _Backend) -> None:
scan_id = "deferred-cancel-fail"
await _create(scan_id)
await asyncio.wait_for(session_manager.cleanup(scan_id), timeout=1)
backend.gate.set()
await _settle()
assert backend.client.deleted == []
@pytest.mark.asyncio
async def test_extra_files_are_staged_inside_the_deferred_bring_up(backend: _Backend) -> None:
scan_id = "deferred-extra"
try:
bundle = await _create(
scan_id,
extra_files=[{"workspace_path": "/workspace/.strix/notes.md", "content": "hi"}],
)
assert backend.session.writes == []
backend.gate.set()
await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
assert len(backend.session.writes) == 1
assert backend.session.execs
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_awaitable_extra_files_do_not_delay_the_backend_call(backend: _Backend) -> None:
scan_id = "deferred-extra-awaitable"
files: asyncio.Future[list[dict[str, Any]] | None] = asyncio.get_running_loop().create_future()
try:
bundle = await _create(scan_id, extra_files=files)
backend.gate.set()
await _settle()
assert backend.calls == 1
assert backend.session.writes == []
files.set_result([{"workspace_path": "/workspace/.strix/notes.md", "content": "hi"}])
await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
assert len(backend.session.writes) == 1
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_awaitable_extra_files_resolving_to_none_stage_nothing(backend: _Backend) -> None:
scan_id = "deferred-extra-none"
async def _nothing() -> list[dict[str, Any]] | None:
return None
try:
bundle = await _create(scan_id, extra_files=_nothing())
backend.gate.set()
await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
assert backend.session.writes == []
finally:
await session_manager.cleanup(scan_id)
@pytest.mark.asyncio
async def test_default_path_still_blocks_until_the_sandbox_is_up(backend: _Backend) -> None:
scan_id = "deferred-off"
backend.gate.set()
try:
bundle = await session_manager.create_or_reuse(scan_id, image="img", local_sources=[])
assert bundle["session"] is backend.session
assert bundle["client"] is backend.client
finally:
await session_manager.cleanup(scan_id)
assert backend.client.deleted == [backend.session]
@pytest.mark.asyncio
async def test_sdk_session_prep_does_not_wait_for_the_bring_up(backend: _Backend) -> None:
scan_id = "deferred-sdk-prep"
try:
bundle = await _create(scan_id)
agent = SandboxAgent(
name="root", instructions="x", tools=[], capabilities=[Filesystem(), Shell()]
)
manager = SandboxRuntimeSessionManager(
starting_agent=agent,
sandbox_config=SandboxRunConfig(client=bundle["client"], session=bundle["session"]),
run_state=None,
)
session = await asyncio.wait_for(
manager.ensure_session(
agent=agent, capabilities=list(agent.capabilities), is_resumed_state=False
),
timeout=1,
)
assert session is bundle["session"]
assert session.pending
backend.gate.set()
await asyncio.wait_for(session.resolve(), timeout=1)
assert not session.pending
finally:
await session_manager.cleanup(scan_id)
+32
View File
@@ -0,0 +1,32 @@
"""``STRIX_DOCKER_SANDBOX_DEVICES`` passes host devices through to the sandbox container."""
from __future__ import annotations
from typing import TYPE_CHECKING
from strix.runtime import docker_client
if TYPE_CHECKING:
import pytest
def test_unset_adds_no_devices(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv(docker_client._SANDBOX_DEVICES_ENV, raising=False)
kwargs: dict[str, object] = {}
docker_client._apply_sandbox_devices(kwargs)
assert kwargs == {}
def test_devices_are_mapped_to_the_same_path_rwm(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv(docker_client._SANDBOX_DEVICES_ENV, " /dev/net/tun, /dev/kvm ,,")
kwargs: dict[str, object] = {}
docker_client._apply_sandbox_devices(kwargs)
assert kwargs["devices"] == ["/dev/net/tun:/dev/net/tun:rwm", "/dev/kvm:/dev/kvm:rwm"]
def test_existing_devices_are_kept_and_not_duplicated(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv(docker_client._SANDBOX_DEVICES_ENV, "/dev/fuse,/dev/net/tun")
kwargs: dict[str, object] = {"devices": ["/dev/fuse"]}
docker_client._apply_sandbox_devices(kwargs)
assert kwargs["devices"] == ["/dev/fuse", "/dev/net/tun:/dev/net/tun:rwm"]
+67
View File
@@ -9,9 +9,11 @@ from typing import Any, cast
from unittest.mock import MagicMock
import pytest
from agents import RunConfig
from agents.exceptions import MaxTurnsExceeded
from agents.items import MessageOutputItem
from agents.memory import SQLiteSession
from agents.run_config import CallModelData, ModelInputData
from agents.tool_context import ToolContext
from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal, ResponseOutputText
@@ -130,6 +132,27 @@ async def test_reserve_stop_notifies_root_once(monkeypatch: pytest.MonkeyPatch)
assert "finish_scan" in str(message["content"])
@pytest.mark.asyncio
async def test_reserve_notice_names_the_root_finish_tool(monkeypatch: pytest.MonkeyPatch) -> None:
coordinator = AgentCoordinator()
coordinator.root_finish_tool = "finish_pr_review"
await coordinator.register("root", "strix", parent_id=None)
await coordinator.register("child-a", "recon", parent_id="root")
sent: list[dict[str, Any]] = []
async def _record(_target_agent_id: str, message: dict[str, Any]) -> bool:
sent.append(message)
return True
monkeypatch.setattr(coordinator, "send", _record)
await _notify_root_on_budget_reserve(coordinator)
assert "call finish_pr_review" in str(sent[0]["content"])
assert "finish_scan" not in str(sent[0]["content"])
@pytest.mark.asyncio
async def test_concurrent_reserve_claims_yield_single_root() -> None:
coordinator = AgentCoordinator()
@@ -1293,6 +1316,21 @@ async def test_interactive_nudge_offers_waiting_without_repeating() -> None:
assert "do not repeat it" in items[0]["content"]
@pytest.mark.asyncio
async def test_nudge_names_the_root_finish_tool() -> None:
items = await execution._append_tool_required_message(
session=None,
context={"parent_id": None},
attempt=1,
limit=3,
interactive=False,
root_finish_tool="finish_pr_review",
)
assert "call finish_pr_review" in items[0]["content"]
assert "finish_scan" not in items[0]["content"]
def _cycle_with_items(
coordinator: AgentCoordinator,
agent_id: str,
@@ -1531,3 +1569,32 @@ async def test_autonomous_nudge_does_not_offer_the_user() -> None:
)
assert "wait_for_user" not in items[0]["content"]
@pytest.mark.asyncio
async def test_text_only_filter_scrubs_images_then_chains_existing_filter() -> None:
run_config = RunConfig(model="m")
assert execution._with_image_scrub(run_config, {"supports_images": True}) is run_config
seen: list[ModelInputData] = []
def _inner(data: CallModelData[Any]) -> ModelInputData:
seen.append(data.model_data)
return data.model_data
run_config = RunConfig(model="m", call_model_input_filter=_inner)
scrub: Any = execution._with_image_scrub(run_config, {"supports_images": False})
image = {"type": "input_image", "image_url": "data:image/png;base64,aGk="}
output = {"type": "function_call_output", "call_id": "c1", "output": [image]}
data = CallModelData(
model_data=ModelInputData(input=[cast("Any", output)], instructions="sys"),
agent=MagicMock(),
context=None,
)
result = await scrub.call_model_input_filter(data)
assert result is seen[0]
assert result.input[0]["output"] == [
{"type": "input_text", "text": execution._TEXT_ONLY_IMAGE_TEXT}
]
assert data.model_data.input[0]["output"] == [image]
+80 -9
View File
@@ -79,12 +79,13 @@ def test_content_guardrail_is_not_retried() -> None:
assert execution._is_transient_model_error(guardrail) is False
def test_client_errors_are_not_transient() -> None:
def test_client_errors_are_transient() -> None:
bad_request = BadRequestError(
"bad", response=httpx.Response(400, request=_request()), body=None
)
assert execution._is_transient_model_error(bad_request) is False
assert execution._is_transient_model_error(_status_error(404)) is False
assert execution._is_transient_model_error(bad_request) is True
for status in (401, 402, 403, 404):
assert execution._is_transient_model_error(_status_error(status)) is False
assert execution._is_transient_model_error(ValueError("nope")) is False
@@ -109,6 +110,7 @@ def _patch_fast_backoff(monkeypatch: pytest.MonkeyPatch) -> None:
async def _run_once(
monkeypatch: pytest.MonkeyPatch,
streams: list[_FakeStream],
session: Any = None,
) -> Any:
_patch_fast_backoff(monkeypatch)
calls = {"n": 0}
@@ -131,7 +133,7 @@ async def _run_once(
run_config=cast("RunConfig", object()),
context={},
max_turns=5,
session=None,
session=session,
interactive=False,
event_sink=None,
hooks=None,
@@ -166,9 +168,78 @@ async def test_run_cycle_gives_up_after_max_retries(
async def test_run_cycle_does_not_retry_permanent_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
bad_request = BadRequestError(
"bad", response=httpx.Response(400, request=_request()), body=None
)
streams = [_FakeStream(exc=bad_request), _FakeStream()]
with pytest.raises(BadRequestError):
streams = [_FakeStream(exc=ValueError("nope")), _FakeStream()]
with pytest.raises(ValueError, match="nope"):
await _run_once(monkeypatch, streams)
@pytest.mark.parametrize(
("status", "message", "expected"),
[
(
404,
"litellm.NotFoundError: NotFoundError: OpenrouterException - "
'{"error":{"message":"No endpoints found that support image input","code":404,'
'"metadata":{"failed_routing_step":"Filter by Image Support"}}}',
True,
),
(400, "Invalid content type. image_url is only supported by certain models.", True),
(400, "my-model is not a multimodal model", True),
(400, "This model doesn't support the image field for user messages.", True),
(422, "messages[3]: unknown variant `image_url`, expected `text`", True),
(400, "'Image URLs in user messages' functionality not supported.", True),
(
400,
"OpenRouterException: Message: This model's maximum context length is 1048576 "
"tokens, but the request requires 1073484 tokens (942412 input including "
"image/vision expansion + 131072 for the completion). Reduce the input length "
"or max_tokens.",
False,
),
(400, "At most 5 image(s) may be provided in one prompt.", False),
(
400,
"Upstream error from Relace: Generation stopped with an incomplete tool call",
False,
),
(500, "No endpoints found that support image input", False),
],
)
def test_is_image_rejection(status: int, message: str, expected: bool) -> None:
exc = APIStatusError(message, response=httpx.Response(status, request=_request()), body=None)
assert execution._is_image_rejection(exc) is expected
async def _run_with_image_session(
monkeypatch: pytest.MonkeyPatch, streams: list[_FakeStream]
) -> tuple[Any, int, int]:
strips = {"n": 0}
async def _fake_strip(_session: Any) -> bool:
strips["n"] += 1
return True
async def _no_compact(*_args: Any, **_kwargs: Any) -> bool:
return False
monkeypatch.setattr(execution, "strip_all_images_from_session", _fake_strip)
monkeypatch.setattr(execution, "_compact_session", _no_compact)
result, attempts, _coordinator = await _run_once(monkeypatch, streams, session=object())
return result, attempts, strips["n"]
@pytest.mark.asyncio
async def test_run_cycle_strips_images_on_image_rejection(
monkeypatch: pytest.MonkeyPatch,
) -> None:
rejection = APIStatusError(
"No endpoints found that support image input",
response=httpx.Response(status_code=404, request=_request()),
body=None,
)
streams = [_FakeStream(exc=rejection), _FakeStream()]
result, attempts, strips = await _run_with_image_session(monkeypatch, streams)
assert result is streams[1]
assert attempts == 2
assert strips == 1
+84 -1
View File
@@ -2,11 +2,15 @@
from __future__ import annotations
from typing import Any
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any
from unittest.mock import MagicMock, patch
import pytest
from agents.usage import Usage
from strix.config import loader
from strix.core.agents import AgentCoordinator
from strix.core.hooks import (
BudgetExceededError,
BudgetPausedError,
@@ -14,6 +18,11 @@ from strix.core.hooks import (
SubagentBudgetReservedError,
recomputed_budget_flags,
)
from strix.report.usage import LLMUsageLedger
if TYPE_CHECKING:
from pathlib import Path
def _make_hooks(max_budget: float | None) -> ReportUsageHooks:
@@ -84,6 +93,38 @@ async def test_over_budget_raises() -> None:
await hooks.on_llm_end(_make_context(), MagicMock(), MagicMock())
@pytest.mark.asyncio
async def test_manual_rates_make_unpriced_model_trip_budget(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.setenv("STRIX_LLM", "test-model")
monkeypatch.setenv("STRIX_LLM_INPUT_COST_PER_MTOK", "10")
monkeypatch.delenv("STRIX_LLM_OUTPUT_COST_PER_MTOK", raising=False)
monkeypatch.delenv("STRIX_LLM_CACHED_INPUT_COST_PER_MTOK", raising=False)
monkeypatch.setattr(loader, "_override", tmp_path / "missing.json")
monkeypatch.setattr(loader, "_cached", None)
usage = Usage()
usage.requests = 1
usage.input_tokens = 1000
usage.total_tokens = 1000
ledger = LLMUsageLedger()
class ReportState:
def record_sdk_usage(self, **kwargs: Any) -> None:
ledger.record(**kwargs)
def get_total_llm_cost(self) -> float:
return ledger.total_cost
hooks = _make_hooks(0.005)
response: Any = SimpleNamespace(usage=usage)
with (
patch("strix.core.hooks.get_global_report_state", return_value=ReportState()),
pytest.raises(BudgetExceededError),
):
await hooks.on_llm_end(_make_context(), MagicMock(), response)
@pytest.mark.asyncio
async def test_budget_check_uses_live_cost_accessor() -> None:
# The check must read the live ledger, not the persisted run-record snapshot,
@@ -335,6 +376,21 @@ async def test_budget_warning_root_directive_distinct_from_subagent() -> None:
assert "confirmed" in sub
@pytest.mark.asyncio
async def test_root_directive_names_the_coordinators_finish_tool() -> None:
hooks = ReportUsageHooks(model="test-model", max_turns=100)
coordinator = AgentCoordinator()
coordinator.root_finish_tool = "finish_pr_review"
ctx = _make_warn_context(requests=85, parent_id=None)
ctx.context["coordinator"] = coordinator
items: list[Any] = []
await hooks.on_llm_start(ctx, MagicMock(), None, items)
assert "finish_pr_review" in items[0]["content"]
assert "finish_scan" not in items[0]["content"]
@pytest.mark.parametrize("parent_id", [None, "root-1"])
@pytest.mark.asyncio
async def test_turn_warning_directive_escalates_per_stage(parent_id: str | None) -> None:
@@ -491,3 +547,30 @@ def test_recomputed_budget_flags(
expected: tuple[bool, bool],
) -> None:
assert recomputed_budget_flags(cost, max_budget, interactive=interactive) == expected
@pytest.mark.asyncio
async def test_turn_limit_warns_ten_before_and_repeats_past_it() -> None:
hooks = ReportUsageHooks(model="test-model")
noticed = []
for requests in range(40):
ctx = _make_warn_context(requests=requests, parent_id="root-1")
ctx.context["turn_limit"] = 25
items: list[Any] = []
await hooks.on_llm_start(ctx, MagicMock(), None, items)
if items:
noticed.append((requests + 1, items[0]["content"]))
assert [turn for turn, _ in noticed] == [15, 25, 30, 35, 40]
assert "15 of your 25-turn limit" in noticed[0][1]
assert "past your 25-turn limit" in noticed[1][1]
@pytest.mark.asyncio
async def test_no_turn_limit_no_notice() -> None:
hooks = ReportUsageHooks(model="test-model")
items: list[Any] = []
await hooks.on_llm_start(
_make_warn_context(requests=14, parent_id="root-1"), MagicMock(), None, items
)
assert items == []
Loaded 100 of 120 files, more files were not shown because too many files have changed in this diff. Show more