Compare commits

...
Author SHA1 Message Date
Ahmed Allam a7c3b55d4f ci: drop the CodeQL workflow 2026-10-04 17:59:22 +00:00
Ahmed Allam a6aa4033ae ci: split CI into per-concern reusable workflows behind one ci-passed gate
Add uv lock --check, a wheel install smoke test, a PyInstaller dry run,
actionlint and zizmor on the workflows, CodeQL, and Dependabot for action
pins, uv, Go and npm dependencies. Release caches are disabled so zizmor's
cache-poisoning audit passes.
2026-10-04 17:57:26 +00:00
Ahmed Allam 24c05c1949 Merge remote-tracking branch 'origin/main' into chore/str-617-pull-request-checks 2026-10-04 17:39:08 +00:00
Ahmed Allam 7a31053348 test(pricing): accept any identically priced provider for bare grok-4.5 2026-10-04 20:38:45 +03:00
Ahmed Allam fe4902b840 ci: run ruff, mypy, bandit, pytest, Go TUI and viewer checks on every pull request 2026-10-04 17:27:06 +00:00
Ahmed Allam cd3bc8011e fix(resume-picker): keep --instruction on a picked run, tolerate blank instructions, read UTF-8 keys, fit narrow terminals, cap the window at 8 rows 2026-10-04 20:07:47 +03:00
Ahmed Allam e954301a09 feat(cli): pick a prior run interactively when --resume has no name
strix --resume with no run name now lists the runs in ./strix_runs inline
in the terminal (started, target, status, findings, run name), newest
first, with arrow-key selection, type-to-search and esc to cancel.
Picking a run continues through the same path as --resume <name>.
Headless or non-TTY launches error with the run list instead.
2026-10-04 20:07:47 +03:00
Ahmed Allam deb6d81d79 refactor(telemetry): keep only the unraisable hook, no docstrings 2026-10-04 18:35:07 +03:00
Ahmed Allam eaf16c6817 refactor(telemetry): collapse the exception hooks into one plain function
No install-once global, no hand-rolled traceback formatting, no SystemExit special case: one helper logs the report with exc_info on strix.telemetry when a handler exists, and both hooks call it.
2026-10-04 18:35:07 +03:00
Ahmed Allam 8ea2c99225 revert(runtime): drop the PTY teardown change in StrixDockerSandboxClient.delete()
Keep this PR to the logging change only: the finalizer reports are routed to strix.log by the unraisable hook, so the sandbox teardown stays as on main.
2026-10-04 18:35:07 +03:00
Ahmed Allam fa5f99c4e0 fix(telemetry): route every unraisable and thread exception to strix.log, never stderr
Completes the previous commit, which only carried the test rename. sys.unraisablehook (finalizers, __del__, GC and weakref callbacks, files closed at exit) and threading.excepthook are replaced for the life of the process with hooks that log the report at WARNING on strix.telemetry: strix.log, and stderr only when the stream handler runs at DEBUG (STRIX_DEBUG=1). Nothing is delegated to Python's default printers; a report arriving after the strix handlers are torn down is dropped instead of reaching logging.lastResort, and failures inside logging itself are swallowed.
2026-10-04 18:35:07 +03:00
Ahmed Allam 121216096c fix(telemetry): route every unraisable and thread exception to strix.log, never stderr
Replaces the pattern-matched urllib3/http.client filter with a generic rule: sys.unraisablehook (finalizers, __del__, GC and weakref callbacks, files closed at exit) and threading.excepthook are replaced for the life of the process with hooks that log the report at WARNING on strix.telemetry. That goes to strix.log, and to stderr only when the stream handler runs at DEBUG (STRIX_DEBUG=1). Nothing is delegated to Python's default printers; a report arriving after the strix handlers are torn down is dropped instead of reaching logging.lastResort, and failures inside logging itself are swallowed so nothing can print mid-shutdown.

tests/test_hook_exceptions_logged.py covers both hooks in-process and end to end in a fresh interpreter (a __del__ raising at shutdown, a thread traceback, a urllib3 response whose file was closed first): stderr stays empty.
2026-10-04 18:35:07 +03:00
Ahmed Allam d945eeee32 fix(telemetry): match only http.client responses in the finalizer filter; test PTY teardown through the real SDK session
The object-shape match accepted any http.* class; it now takes urllib3 and http.client only, so a closed-file error from another module still reaches the previous unraisable hook. New tests build a real DockerSandboxSession holding a PTY exec stream and run the real SDK delete() with the container gone: the SDK alone leaves the socket and its pinned response open, StrixDockerSandboxClient.delete() closes both.
2026-10-04 18:35:07 +03:00
Ahmed Allam b4be726a3b fix(runtime): terminate PTY exec streams on teardown and silence 3.14 response finalizer noise
StrixDockerSandboxClient.delete() killed the container and then let the SDK's delete() run shutdown(), which only terminates the agent's PTY exec streams while the container still exists. Once the container was gone the hijacked exec sockets were left to the garbage collector and their HTTP responses failed to close at interpreter exit, which Python 3.14 reports as "Exception ignored while finalizing file <urllib3.response.HTTPResponse>" after the scan summary. delete() now awaits pty_terminate_all() first, whatever the container's state.

The unraisable filter now also matches Python 3.14's finalizer shape (object=None, repr in err_msg) and http.client responses, and logs the match at DEBUG on strix.telemetry instead of letting it reach stderr.
2026-10-04 18:35:07 +03:00
Ahmed Allam 25f3978804 fix(cli): install the stderr log handler at startup and prepare the run state before the TUI launches 2026-10-04 05:44:06 +03:00
Ahmed Allam 5325009f65 fix(cli): verify the model before the TUI opens on a direct launch 2026-10-04 05:44:06 +03:00
Ahmed Allam d218c29ed1 fix(tui): blank row under panel headers; no panel controls when Agents is the only panel 2026-10-04 04:54:29 +03:00
Ahmed Allam 849671f288 fix(tui): drop the Model label from the stats panel 2026-10-04 04:54:29 +03:00
Ahmed Allam faca0d4d1d fix(tui): pin the Model panel to the sidebar bottom and enlarge the show-sidebar button 2026-10-04 04:54:29 +03:00
Ahmed Allam 1a680f4aa3 fix(tui): keep sidebar panels inside the terminal height 2026-10-04 04:54:29 +03:00
Ahmed Allam 701c22b8a9 fix(tui): collapsible, zoomable sidebar panels and a sidebar toggle
Every sidebar panel gets a header with a click-to-collapse chevron and a
click-to-zoom glyph; a one-row chip on the viewer line hides the whole
sidebar and brings it back from a narrow rail. Tab skips collapsed and
hidden panels, scrollbar hit zones follow the rendered bars, and
fillBackground also repaints after the bare ESC[m reset so no cell shows
the terminal background.
2026-10-04 04:54:29 +03:00
Ahmed Allam e1d7f5c433 fix(tui): keep blank lines inside fenced code when rendering report markdown 2026-10-04 02:11:45 +03:00
Ahmed Allam 64dae6f484 fix(tui): render report section bodies as markdown 2026-10-04 02:11:45 +03:00
Ahmed Allam b49ac2fd9e fix(preflight): check extra headers for subscription models and the dedupe key as sent 2026-10-04 01:48:35 +03:00
Ahmed Allam 08df05db61 fix(preflight): skip unused credentials for subscription models, check dedupe credentials too 2026-10-04 01:48:35 +03:00
Ahmed Allam 4ca15a6e60 style: drop docstrings from the header-value check 2026-10-04 01:48:35 +03:00
Ahmed Allam 539960d953 fix(preflight): name a non-ASCII character in the API key instead of raising UnicodeEncodeError
httpx encodes header values as ASCII, so a smart quote, non-breaking
space or byte-order mark pasted into LLM_API_KEY (or LLM_EXTRA_HEADERS)
surfaced as a bare UnicodeEncodeError from inside the client, wrapped in
ModelConnectionError. The preflight now checks these settings first and
fails with the setting name, the code point, its Unicode name and its
position. Nothing is trimmed or rewritten and the key itself is never
printed.
2026-10-04 01:48:35 +03:00
Ahmed Allam 7280c40caa style: drop docstrings from the UTF-8 stream helpers 2026-10-04 01:19:48 +03:00
Ahmed Allam 5c3476795a fix(cli): force UTF-8 stdout/stderr on Windows so Rich output never raises
Windows hands a redirected or legacy console stream the ANSI code page
(cp1252), which cannot encode Rich's panels or the model's text: one
check mark in a finding ended a headless run with UnicodeEncodeError, and
the error handler raised again rendering its own panel. Reconfigure both
streams to UTF-8 at startup on win32, and open the Go TUI's os.devnull
sink as UTF-8 so logging into it stops producing --- Logging error ---
reports under the same code page.
2026-10-04 01:19:48 +03:00
Ahmed Allam 45b775dcd5 fix(cli): apply the --fail-on threshold regardless of run status
A stopped run with no findings already exits 0, so skipping the threshold
only when low findings exist made the gate depend on the wrong thing.
The threshold now always applies; reporting an unfinished run is a separate
exit-code concern.
2026-10-04 00:03:23 +03:00
siundu254 b11228192d Resolve PR Comments 2026-10-04 00:03:23 +03:00
siundu254 3cd6c93fa0 Fail on Severity 2026-10-04 00:03:23 +03:00
Ahmed Allam 65172fecd8 fix(tui): link every wrapped line of the viewer URL to the full URL
The sidebar wraps the viewer URL, and terminals that linkify by text only
pick up the first line, so the click opened the viewer with a truncated
token. Emit each wrapped line as an OSC 8 hyperlink to the whole URL.
2026-10-03 23:24:05 +03:00
Ahmed Allam 0107a15295 test: reject a leading heading in any finish_scan example field 2026-10-03 23:07:56 +03:00
Ahmed Allam c0258b25fa fix(finish_scan): stop asking for a section heading in every report field
Every renderer of the final report already titles each section, so the
heading the docstring asked for printed twice. Describe the fields as
section bodies and drop the example headings.
2026-10-03 23:07:56 +03:00
Ahmed Allam 99c0711687 fix(config): use the Responses API whenever the model's catalog entry lists /v1/responses 2026-10-02 18:41:53 +03:00
Ahmed Allam 2635fb5307 chore(inputs): drop unused logger 2026-10-02 18:17:27 +03:00
Ahmed Allam 8ab5e39c06 fix(inputs): send reasoning_effort as configured; no route-specific handling 2026-10-02 18:17:27 +03:00
Ahmed Allam e1ec259ac2 fix(inputs): send reasoning_effort=none explicitly on chat completions; hint at the Responses API when tools+effort are rejected 2026-10-02 18:17:27 +03:00
Ahmed Allam 066bd60a03 fix(runner): pick the SDK route from the resolved model override
configure_sdk_model_defaults only sees STRIX_LLM; a model= override to
run_strix_scan re-applies the route for the model that actually runs.
2026-10-02 18:17:27 +03:00
Ahmed Allam 6ab123484d fix(config): choose Responses vs chat completions from the model, not the base URL
A base URL no longer forces chat completions. resolve_api_type() keeps an
explicit STRIX_API_TYPE, uses Responses without a base URL or for
api.openai.com, uses Responses for models whose LiteLLM catalog entry has
no /v1/chat/completions endpoint, and chat completions for other gateways.

On the chat completions route, reasoning_effort is omitted for models
whose LiteLLM parameter map does not list it there instead of failing the
request with function tools. STRIX_REASONING_EFFORT and STRIX_API_TYPE
are matched case-insensitively.
2026-10-02 18:17:27 +03:00
alex s 007ed1a94e fix(reporting): restore create_vulnerability_report parameter descrip… (#1391)
* fix(reporting): restore create_vulnerability_report parameter descriptions

A docstring line beginning with a backtick fence example opened a markdown
code block that griffe's Google-style parser never saw closed, so the Args
section was parsed as plain text and the generated tool schema carried no
per-parameter descriptions. Reword the example, move Args after the trailing
notes so nothing after it is dropped from the tool description, and add a
test asserting every scan-agent tool parameter has a description.

* test(reporting): cover respond_to_user and reject null parameter descriptions
2026-09-30 13:26:48 -04:00
61 changed files with 3592 additions and 501 deletions

No files matched your search

@@ -0,0 +1,21 @@
name: Set up Python environment
description: Install Python and uv, then sync the locked project environment.
inputs:
python-version:
description: Python version to install.
default: '3.12'
runs:
using: composite
steps:
- uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: ${{ inputs.python-version }}
- uses: astral-sh/setup-uv@d4b2f3b6ecc6e67c4457f6d3e41ec42d3d0fcb86 # v5.4.2
with:
enable-cache: true
- run: uv sync --frozen --python "$PYTHON_VERSION"
shell: bash
env:
PYTHON_VERSION: ${{ inputs.python-version }}
+49
View File
@@ -0,0 +1,49 @@
version: 2
updates:
- package-ecosystem: github-actions
directory: /
schedule:
interval: weekly
cooldown:
default-days: 7
groups:
actions:
patterns:
- '*'
- package-ecosystem: uv
directory: /
schedule:
interval: weekly
cooldown:
default-days: 7
groups:
python:
update-types:
- minor
- patch
- package-ecosystem: gomod
directory: /strix/interface/tui
schedule:
interval: weekly
cooldown:
default-days: 7
groups:
go:
update-types:
- minor
- patch
- package-ecosystem: npm
directory: /strix/interface/viewer/frontend
schedule:
interval: weekly
cooldown:
default-days: 7
groups:
viewer:
update-types:
- minor
- patch
+3 -1
View File
@@ -43,12 +43,14 @@ jobs:
python-version: '3.12'
- uses: astral-sh/setup-uv@d4b2f3b6ecc6e67c4457f6d3e41ec42d3d0fcb86 # v5.4.2
with:
enable-cache: false
- uses: actions/setup-go@44694675825211faa026b3c33043df3e48a5fa00 # v6.0.0
with:
go-version: '1.24.x'
check-latest: true
cache-dependency-path: strix/interface/tui/go.sum
cache: false
- name: Build
shell: bash
+55
View File
@@ -0,0 +1,55 @@
name: Package
on:
workflow_call:
permissions:
contents: read
jobs:
wheel:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- uses: actions/setup-go@44694675825211faa026b3c33043df3e48a5fa00 # v6.0.0
with:
go-version-file: strix/interface/tui/go.mod
cache-dependency-path: strix/interface/tui/go.sum
- uses: ./.github/actions/setup-python-env
- run: uv build --wheel
- name: Wheel ships the TUI sidecar and the viewer bundle
run: |
uv run --frozen python - <<'PY'
import glob, zipfile
(wheel,) = glob.glob("dist/*.whl")
names = zipfile.ZipFile(wheel).namelist()
assert "strix/bin/strix-tui" in names, names
assert "strix/interface/viewer/static/index.html" in names, names
assert not any(name.startswith("strix/interface/viewer/frontend/") for name in names)
assert not any(name.startswith("strix/interface/tui/cmd/") for name in names)
PY
- name: Install into a clean environment and run the CLI
run: |
uv venv --python 3.12 /tmp/strix-smoke
uv pip install --python /tmp/strix-smoke/bin/python dist/*.whl
/tmp/strix-smoke/bin/strix --version
/tmp/strix-smoke/bin/strix --help >/dev/null
binary:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- uses: actions/setup-go@44694675825211faa026b3c33043df3e48a5fa00 # v6.0.0
with:
go-version-file: strix/interface/tui/go.mod
cache-dependency-path: strix/interface/tui/go.sum
- uses: ./.github/actions/setup-python-env
- run: make tui-build
- run: uv run --frozen pyinstaller strix.spec --noconfirm
- run: dist/strix --version
- name: Binary bundles the TUI sidecar
run: uv run --frozen pyi-archive_viewer -l dist/strix | grep -E "strix/bin/strix-tui" >/dev/null
+61
View File
@@ -0,0 +1,61 @@
name: Python
on:
workflow_call:
permissions:
contents: read
jobs:
lock:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- uses: astral-sh/setup-uv@d4b2f3b6ecc6e67c4457f6d3e41ec42d3d0fcb86 # v5.4.2
- run: uv lock --check
ruff:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- uses: ./.github/actions/setup-python-env
- run: uv run --frozen ruff check .
- run: uv run --frozen ruff format --check .
mypy:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- uses: ./.github/actions/setup-python-env
- run: uv run --frozen mypy strix/
bandit:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- uses: ./.github/actions/setup-python-env
- run: uv run --frozen bandit -r strix/ -c pyproject.toml
pytest:
name: pytest (${{ matrix.python-version }})
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ['3.12', '3.13', '3.14']
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- uses: ./.github/actions/setup-python-env
with:
python-version: ${{ matrix.python-version }}
- run: uv run --frozen pytest -q
+26
View File
@@ -0,0 +1,26 @@
name: TUI
on:
workflow_call:
permissions:
contents: read
jobs:
go:
runs-on: ubuntu-latest
defaults:
run:
working-directory: strix/interface/tui
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- uses: actions/setup-go@44694675825211faa026b3c33043df3e48a5fa00 # v6.0.0
with:
go-version-file: strix/interface/tui/go.mod
cache-dependency-path: strix/interface/tui/go.sum
- run: test -z "$(gofmt -l .)"
- run: go vet ./...
- run: CGO_ENABLED=0 go build -trimpath -o /dev/null ./cmd/strix-tui
- run: go test -race ./...
+27
View File
@@ -0,0 +1,27 @@
name: Viewer
on:
workflow_call:
permissions:
contents: read
jobs:
build:
runs-on: ubuntu-latest
defaults:
run:
working-directory: strix/interface/viewer/frontend
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
with:
node-version: '24'
cache: npm
cache-dependency-path: strix/interface/viewer/frontend/package-lock.json
- run: npm ci --no-audit --no-fund
- run: npm run build
- name: Committed viewer bundle matches the build
run: git -C "$GITHUB_WORKSPACE" diff --exit-code --stat -- strix/interface/viewer/static && test -z "$(git -C "$GITHUB_WORKSPACE" status --porcelain -- strix/interface/viewer/static)"
+37
View File
@@ -0,0 +1,37 @@
name: Workflows
on:
workflow_call:
permissions:
contents: read
jobs:
actionlint:
runs-on: ubuntu-latest
env:
ACTIONLINT_VERSION: 1.7.12
ACTIONLINT_SCRIPT_REF: 914e7df21a07ef503a81201c76d2b11c789d3fca # v1.7.12
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- name: Install actionlint
run: |
mkdir -p "$RUNNER_TEMP/actionlint"
curl -fsSL --retry 5 -o "$RUNNER_TEMP/actionlint/download.bash" \
"https://raw.githubusercontent.com/rhysd/actionlint/$ACTIONLINT_SCRIPT_REF/scripts/download-actionlint.bash"
bash "$RUNNER_TEMP/actionlint/download.bash" "$ACTIONLINT_VERSION" "$RUNNER_TEMP/actionlint"
- run: |
"$RUNNER_TEMP/actionlint/actionlint" -color
zizmor:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
with:
persist-credentials: false
- uses: zizmorcore/zizmor-action@cc914d7f3750a2d13d75c7f184a1060aa0e9d482 # v0.6.4
with:
advanced-security: false
version: 1.22.0
+52
View File
@@ -0,0 +1,52 @@
name: CI
on:
pull_request:
push:
branches:
- main
workflow_dispatch:
permissions:
contents: read
concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: true
jobs:
python:
uses: ./.github/workflows/ci-python.yml
tui:
uses: ./.github/workflows/ci-tui.yml
viewer:
uses: ./.github/workflows/ci-viewer.yml
package:
uses: ./.github/workflows/ci-package.yml
workflows:
uses: ./.github/workflows/ci-workflows.yml
ci-passed:
name: ci-passed
if: always()
needs:
- python
- tui
- viewer
- package
- workflows
runs-on: ubuntu-latest
steps:
- name: Every job succeeded
env:
NEEDS: ${{ toJSON(needs) }}
run: |
failing=$(jq -r 'to_entries[] | select(.value.result != "success") | "\(.key): \(.value.result)"' <<< "$NEEDS")
if [ -n "$failing" ]; then
echo "$failing"
exit 1
fi
+1 -1
View File
@@ -4,7 +4,7 @@ exclude: ^strix/interface/viewer/static/assets/
repos:
# Ruff for fast linting and formatting
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.11.13
rev: v0.15.20
hooks:
- id: ruff
args: [--fix, --exit-non-zero-on-fix]
+8 -2
View File
@@ -23,9 +23,15 @@ strix -n --target ./app --scan-mode quick --scope-mode diff --diff-base origin/m
| Code | Meaning |
|------|---------|
| 0 | No vulnerabilities found |
| 0 | No vulnerabilities found (with `--fail-on`, none at or above the threshold) |
| 1 | Execution error |
| 2 | Vulnerabilities found |
| 2 | Vulnerabilities found (with `--fail-on`, at least one at or above the threshold) |
To fail only on serious findings, pass `--fail-on` with a minimum severity. Lower findings are still reported but exit `0`, so a passing job does not mean the report is empty.:
```bash
strix -n --target ./app --scan-mode quick --fail-on high
```
## GitLab CI
+3 -1
View File
@@ -46,9 +46,11 @@ The workflow fails when vulnerabilities are found:
| Code | Result |
|------|--------|
| 0 | Pass — No vulnerabilities |
| 0 | Pass — No vulnerabilities (with `--fail-on`, none at or above the threshold) |
| 2 | Fail — Vulnerabilities found |
Add `--fail-on high` (or `critical`, `medium`, `low`) to fail only on findings at or above that severity. Lower findings still appear in the report, so a passing run is not necessarily finding-free.
## Scan Modes for CI
| Mode | Duration | Use Case |
+13 -2
View File
@@ -60,6 +60,14 @@ strix (--target <target> | --target-list <path>) [options]
Run in headless mode without TUI. Ideal for CI/CD.
</ParamField>
<ParamField path="--fail-on" type="string">
Headless mode only (requires `-n`). Minimum severity that makes the run exit
`2`: `critical`, `high`, `medium`, `low`, or `info`. Findings below the
threshold are still written to every report artifact. A finding whose
severity Strix does not recognize always counts, so the gate never passes on a
value it cannot rank. Omit the flag to exit `2` on any finding.
</ParamField>
<ParamField path="--config" type="string">
Path to a custom config file (JSON) to use instead of `~/.strix/cli-config.json`.
</ParamField>
@@ -132,6 +140,9 @@ strix --target api.example.com --instruction "Focus on IDOR and auth bypass"
# CI/CD mode
strix -n --target ./ --scan-mode quick
# CI/CD mode, failing only on high or critical findings
strix -n --target ./ --scan-mode quick --fail-on high
# Cap cost and per-agent turns
strix --target https://example.com --max-budget 25 --max-turns 300
@@ -159,6 +170,6 @@ strix --target https://app.com --workspace-file ./openapi.yaml:specs/openapi.yam
| Code | Meaning |
|------|---------|
| 0 | Scan completed successfully (interactive mode always exits `0`; in headless mode, `0` means no vulnerabilities were found) |
| 0 | Scan completed successfully (interactive mode always exits `0`; in headless mode, `0` means no vulnerabilities were found, or with `--fail-on`, none at or above the threshold; lower findings are still in the report) |
| 1 | A fatal error occurred before or during the scan (e.g. missing environment variables, Docker unavailable, invalid config file, diff-scope resolution failure, or an unhandled error) |
| 2 | Vulnerabilities found (headless mode only) |
| 2 | Vulnerabilities found (headless mode only; with `--fail-on`, at least one finding at or above that severity) |
@@ -67,7 +67,7 @@ Then tell the user to add two repository secrets: `STRIX_LLM` (model id, for exa
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.
- Exit codes: `0` pass, `2` vulnerabilities found (fails the job), `1` setup error.
- Exit codes: `0` pass, `2` vulnerabilities found (fails the job), `1` setup error. Add `--fail-on high` to fail only on high/critical findings; lower ones are still reported, so a passing job is not a finding-free report.
- The runner needs Docker (default GitHub-hosted Ubuntu runners have it).
- **Size the budget so the scan completes — do not let it fail open.** A `0` exit means "no validated vulnerabilities in what was analyzed"; if `--max-budget` is hit before the diff is fully covered, the scan wraps up early and can still exit `0`. The "Fail unless the scan completed" step above narrows the gap: `strix_runs/<run>/run.json` is `"stopped"` when the scan was cut off at the hard budget limit without a final report. It is not a complete guard — the agents get graduated wrap-up warnings before that limit, and a run that wraps up on a warning still calls `finish_scan` and records `"completed"` with partial coverage. So keep that step in any pipeline that gates merges **and** give the scan real headroom (compare `run.json`'s `llm_usage.cost` against `--max-budget`; if it ran right up to the cap, raise it). For a `quick` diff-scoped PR scan `--max-budget 10` is usually ample, raise it for large diffs.
@@ -94,6 +94,7 @@ Key flags:
| `--workspace-file PATH[:DEST]` | Copy a file from this machine into `/workspace` before the scan, for a wordlist, a spec, or notes. Repeatable. |
| `--max-budget USD` | Hard LLM spend cap; scan wraps up cleanly at the limit. |
| `--max-turns N` | Per-agent turn cap (default 500). |
| `--fail-on SEVERITY` | Headless only: exit `2` only for findings at or above `critical`/`high`/`medium`/`low`/`info`. Default: any finding. |
| `--resume RUN_NAME` | Resume a prior run from `strix_runs/`, with its agent history and targets. Cannot be combined with `-t`. |
| `--scope-mode` | For code targets: `auto` (diff-scope in CI/headless), `diff` (force changed files only), `full` (whole tree). |
| `--diff-base REF` | Branch or commit that `diff` scope compares against. Defaults to the repo's default branch. |
@@ -102,9 +103,9 @@ Scans take minutes (`quick`) to hours (`deep`). Run them in the background and p
### Exit codes (headless)
- `0` — finished with no validated vulnerabilities **in what was analyzed**
- `0` — finished with no validated vulnerabilities **in what was analyzed** (with `--fail-on`, none at or above the threshold; lower ones are still in the artifacts)
- `1` — fatal error (missing env vars, Docker down, bad config)
- `2` — vulnerabilities found
- `2` — vulnerabilities found (with `--fail-on`, at least one at or above that severity)
A `0` is not proof of full coverage: if `--max-budget`/`--max-turns` is reached before the scan completes, it wraps up early and still exits `0`. When you need assurance the scan finished, give it enough budget and check `strix_runs/<run>/run.json`: a hard budget stop leaves `status: "stopped"`, but an agent that wrapped up early on a budget *warning* still calls `finish_scan` and records `"completed"` — so also sanity-check the run's cost against `--max-budget` and the report's stated coverage before treating a clean result as full coverage.
+47 -15
View File
@@ -58,7 +58,7 @@ if TYPE_CHECKING:
from openai import AsyncOpenAI
from openai.types.responses.response_prompt_param import ResponsePromptParam
from strix.config.settings import LlmSettings, ReasoningEffort, Settings
from strix.config.settings import ApiType, LlmSettings, ReasoningEffort, Settings
logger = logging.getLogger(__name__)
@@ -649,14 +649,40 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
if llm.api_base:
os.environ["OPENAI_BASE_URL"] = llm.api_base
_configure_litellm_default("api_base", llm.api_base)
api_type = llm.api_type
if api_type is None:
api_type = "chat_completions" if llm.api_base else "responses"
set_default_openai_api(api_type)
configure_sdk_api_route(llm.model or "", settings)
_configure_extra_headers(llm)
def configure_sdk_api_route(model_name: str, settings: Settings) -> None:
"""Point SDK-native OpenAI requests for ``model_name`` at Responses or chat completions."""
api_type = resolve_api_type(model_name, settings)
logger.info("OpenAI API route for %s: %s", model_name, api_type)
set_default_openai_api(api_type)
_RESPONSES_ENDPOINT = "/v1/responses"
def resolve_api_type(model_name: str, settings: Settings) -> ApiType:
"""The SDK-native OpenAI route for ``model_name``: Responses or chat completions.
An explicit ``STRIX_API_TYPE`` wins. Otherwise the model decides: Responses
when LiteLLM's catalog lists ``/v1/responses`` for it, chat completions for
everything else.
"""
if settings.llm.api_type is not None:
return settings.llm.api_type
if _RESPONSES_ENDPOINT in _catalog_supported_endpoints(model_name):
return "responses"
return "chat_completions"
def _catalog_supported_endpoints(model_name: str) -> list[str]:
entry = _catalog_entry(model_name)
endpoints = entry.get("supported_endpoints") if entry else None
return [str(e) for e in endpoints] if isinstance(endpoints, list) else []
def _mirror_api_key_to_provider_env(model_name: str | None, api_key: str) -> None:
if not model_name:
return
@@ -859,11 +885,7 @@ def uses_chat_completions_tool_schema(model_name: str, settings: Settings) -> bo
model = model_name.strip().lower()
if "/" in model and not model.startswith("openai/"):
return True
if settings.llm.api_type is not None:
return settings.llm.api_type == "chat_completions"
if settings.llm.api_base:
return True
return not model_supports_reasoning(model_name)
return resolve_api_type(model_name, settings) == "chat_completions"
def supports_strict_tool_schemas(model_name: str) -> bool:
@@ -879,17 +901,27 @@ def supports_strict_tool_schemas(model_name: str) -> bool:
def model_supports_reasoning(model_name: str) -> bool:
import litellm
entry = _catalog_entry(model_name)
return bool(entry and entry.get("supports_reasoning"))
def _bare_openai_name(model_name: str) -> str:
name = model_name.strip().lower()
for prefix in ("litellm/", "any-llm/", "openai/"):
if name.startswith(prefix):
name = name[len(prefix) :]
break
return name[len(prefix) :]
return name
def _catalog_entry(model_name: str) -> dict[str, Any] | None:
"""LiteLLM's cost-map entry for the model, looked up as it would route it."""
import litellm
name = _bare_openai_name(model_name)
entry = litellm.model_cost.get(name)
if entry is None and "/" in name:
entry = litellm.model_cost.get(name.rsplit("/", 1)[1])
return bool(entry and entry.get("supports_reasoning"))
return entry if isinstance(entry, dict) else None
def is_known_openai_bare_model(model_name: str) -> bool:
+17 -1
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
from typing import Literal
from pydantic import AliasChoices, Field
from pydantic import AliasChoices, Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -13,6 +13,12 @@ ApiType = Literal["responses", "chat_completions"]
DEFAULT_MAX_TURNS = 500
def _lowercase(value: object) -> object:
"""Enum-like env values are matched case-insensitively (``None`` is ``none``)."""
return value.strip().lower() if isinstance(value, str) else value
_BASE_CONFIG = SettingsConfigDict(
case_sensitive=False,
populate_by_name=True,
@@ -78,6 +84,11 @@ class LlmSettings(BaseSettings):
alias="LLM_MAX_TOOL_CALLS_PER_TURN",
)
@field_validator("api_type", "reasoning_effort", mode="before")
@classmethod
def _normalize_case(cls, value: object) -> object:
return _lowercase(value)
class DedupeSettings(BaseSettings):
model_config = _BASE_CONFIG
@@ -95,6 +106,11 @@ class DedupeSettings(BaseSettings):
repr=False,
)
@field_validator("reasoning_effort", mode="before")
@classmethod
def _normalize_case(cls, value: object) -> object:
return _lowercase(value)
class ContextSettings(BaseSettings):
"""Context-window management: per-tool-output caps and history compaction."""
+1 -3
View File
@@ -267,9 +267,7 @@ def make_model_settings(
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))
if force_required_tool_choice and _accepts_required_tool_choice(model_name):
model_settings = model_settings.resolve(ModelSettings(tool_choice="required"))
+6 -1
View File
@@ -18,9 +18,10 @@ from openai import RateLimitError
from strix.agents.factory import build_strix_agent, make_child_factory
from strix.agents.prompt import render_scope_prompt, render_system_prompt
from strix.config import load_settings
from strix.config import codex, load_settings
from strix.config.models import (
StrixProvider,
configure_sdk_api_route,
configure_sdk_model_defaults,
supports_strict_tool_schemas,
uses_chat_completions_tool_schema,
@@ -257,6 +258,10 @@ async def run_strix_scan(
raise RuntimeError(
"No LLM model configured. Set STRIX_LLM env or pass model= to run_strix_scan().",
)
if resolved_model != (settings.llm.model or "").strip() and not codex.subscription_model(
resolved_model
):
configure_sdk_api_route(resolved_model, settings)
logger.info("LLM model resolved: %s", resolved_model)
chat_completions_tools = uses_chat_completions_tool_schema(resolved_model, settings)
strict_tool_schemas = supports_strict_tool_schemas(resolved_model)
+87 -22
View File
@@ -9,7 +9,7 @@ from pathlib import Path
from strix.config import apply_config_override
from strix.config.settings import DEFAULT_MAX_TURNS
from strix.core.paths import run_dir_for, runtime_state_dir
from strix.core.paths import RUNS_DIR_NAME, run_dir_for, runtime_state_dir
from strix.interface.scan_setup import attach_workspace_mount, build_targets_info
from strix.interface.update_check import self_update
from strix.interface.utils import (
@@ -18,6 +18,11 @@ from strix.interface.utils import (
resolve_workspace_files,
validate_config_file,
)
from strix.report.runs import list_run_summaries
# Severities ``--fail-on`` accepts, most severe first.
FAIL_ON_SEVERITIES = ("critical", "high", "medium", "low", "info")
def get_version() -> str:
@@ -186,6 +191,21 @@ Strix Cloud:
),
)
parser.add_argument(
"--fail-on",
dest="fail_on",
type=str.lower,
choices=FAIL_ON_SEVERITIES,
default=None,
metavar="SEVERITY",
help=(
"Headless mode only: exit 2 only when a finding is at or above this severity "
"(critical, high, medium, low, info). Lower findings are still written to every "
"report artifact. A finding with an unrecognized severity always counts. "
"Default: any finding exits 2."
),
)
parser.add_argument(
"-m",
"--scan-mode",
@@ -281,11 +301,14 @@ Strix Cloud:
parser.add_argument(
"--resume",
type=str,
nargs="?",
const="",
metavar="RUN_NAME",
help=(
"Resume a prior scan by its run name (the dir under ./strix_runs/). "
"Picks up the root + every non-terminal subagent's full LLM history "
"and agent topology. Skips fresh run-name generation."
"and agent topology. Skips fresh run-name generation. Without a "
"name, opens a picker of the prior runs."
),
)
@@ -293,6 +316,7 @@ Strix Cloud:
# Startup-resolved state lives alongside the parsed flags. The full schema
# is established here so downstream code reads attributes directly.
args.needs_setup = False
args.resume_picker = False
args.targets_info = []
args.local_sources = []
args.diff_scope = {"active": False}
@@ -318,6 +342,9 @@ Strix Cloud:
if args.update:
sys.exit(0 if self_update() else 1)
if args.fail_on and not args.non_interactive:
parser.error("--fail-on only applies to headless runs; add -n/--non-interactive.")
if args.instruction and args.instruction_file:
parser.error(
"Cannot specify both --instruction and --instruction-file. Use one or the other."
@@ -338,28 +365,34 @@ Strix Cloud:
except ValueError as error:
parser.error(f"--workspace-file: {error}")
args.user_explicit_instruction = args.instruction if args.resume else None
args.user_explicit_instruction = args.instruction if args.resume is not None else None
# What the user actually asked for, kept apart from args.instruction because
# prepare_run prepends the diff-scope preamble to that. This is the text the
# transcript shows as their opening message.
args.user_instruction = args.instruction or None
if args.resume:
if args.resume is not None:
if args.target or args.target_list:
parser.error(
"Cannot combine --resume with --target/--target-list. "
"--resume picks up where the prior run left off, including the "
"original target list."
)
_load_resume_state(args, parser)
agents_path = runtime_state_dir(run_dir_for(args.resume)) / "agents.json"
if not agents_path.exists():
parser.error(
f"--resume {args.resume}: missing {agents_path}. The run was "
f"persisted but never reached its first agent snapshot — "
f"there's nothing to resume from. Pick a fresh --run-name "
f"or remove --resume to start over with the same targets."
)
if not args.resume.strip():
# A bare --resume: main() opens the inline picker of prior runs
# before anything launches; headless has nobody to pick, so it
# lists them.
args.resume = None
args.resume_picker = True
if args.non_interactive:
parser.error(resume_run_list_message("--resume needs a run name in headless mode."))
if not list_run_summaries():
parser.error(f"--resume: no runs in ./{RUNS_DIR_NAME} to resume")
return args
try:
load_resume_state(args)
except ResumeError as exc:
parser.error(str(exc))
else:
if not args.target and not args.target_list:
if args.non_interactive:
@@ -381,21 +414,45 @@ Strix Cloud:
return args
def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
"""Populate ``args.targets_info`` and friends from a prior run's run.json."""
class ResumeError(ValueError):
"""A prior run cannot be resumed as recorded; the message names why."""
def resume_run_list_message(lead: str) -> str:
"""``lead`` followed by the runs that ``--resume <name>`` would accept."""
runs = list_run_summaries()
if not runs:
return f"{lead} There are no runs in ./{RUNS_DIR_NAME}."
name_width = max(len(run.run_name) for run in runs)
status_width = max(len(run.status) for run in runs)
lines = [f"{lead} Runs in ./{RUNS_DIR_NAME}:"]
lines.extend(
f" {run.run_name:<{name_width}} {run.status:<{status_width}} "
f"{run.started_at[:19]:<19} {run.target}".rstrip()
for run in runs
)
return "\n".join(lines)
def load_resume_state(args: argparse.Namespace) -> None:
"""Populate ``args.targets_info`` and friends from a prior run's run.json.
Raises :class:`ResumeError` when the run is missing, unreadable, or its
recorded workspace is gone.
"""
from strix.report.writer import read_run_record
run_dir = run_dir_for(args.resume)
state_path = run_dir / "run.json"
if not state_path.exists():
parser.error(
raise ResumeError(
f"--resume {args.resume}: no such run "
f"(missing {state_path}; remove --resume for a fresh start)"
)
try:
state = read_run_record(run_dir)
except (RuntimeError, TypeError) as exc:
parser.error(f"--resume {args.resume}: run.json unreadable: {exc}")
raise ResumeError(f"--resume {args.resume}: run.json unreadable: {exc}") from exc
args.targets_info = state.get("targets_info") or []
# A target-less run has no targets_info at all. It is driven by its
@@ -403,7 +460,7 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
# mount was declined, so either of those is enough to resume it.
workspace_mount = state.get("workspace_mount") or None
if not args.targets_info and not workspace_mount and not state.get("user_instruction"):
parser.error(f"--resume {args.resume}: run.json has no targets_info")
raise ResumeError(f"--resume {args.resume}: run.json has no targets_info")
for target in args.targets_info:
if not isinstance(target, dict):
@@ -413,7 +470,7 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
try:
check_mountable_dir(Path(details["target_path"]).expanduser())
except ValueError as exc:
parser.error(f"--resume {args.resume}: {exc}")
raise ResumeError(f"--resume {args.resume}: {exc}") from exc
continue
if target.get("type") != "repository":
continue
@@ -421,7 +478,7 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
if not cloned:
continue
if not Path(cloned).expanduser().exists():
parser.error(
raise ResumeError(
f"--resume {args.resume}: cloned repo at {cloned} is missing. "
f"It was deleted between runs. Pick a fresh --run-name to "
f"re-clone, or restore the directory before resuming."
@@ -452,10 +509,10 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
try:
args.workspace_files = resolve_workspace_files(restored)
except ValueError as error:
parser.error(f"--resume {args.resume}: invalid workspace file: {error}")
raise ResumeError(f"--resume {args.resume}: invalid workspace file: {error}") from error
if workspace_mount:
if not Path(workspace_mount).expanduser().is_dir():
parser.error(
raise ResumeError(
f"--resume {args.resume}: the working directory {workspace_mount} "
f"is missing. Restore it before resuming, or start a fresh run."
)
@@ -465,3 +522,11 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
persisted_scan_mode = state.get("scan_mode")
if persisted_scan_mode and args.scan_mode == "deep":
args.scan_mode = persisted_scan_mode
agents_path = runtime_state_dir(run_dir) / "agents.json"
if not agents_path.exists():
raise ResumeError(
f"--resume {args.resume}: missing {agents_path}. The run was "
f"persisted but never reached its first agent snapshot — "
f"there's nothing to resume from. Pick a fresh --run-name "
f"or remove --resume to start over with the same targets."
)
+82 -12
View File
@@ -8,14 +8,21 @@ import asyncio
import contextlib
import sys
from pathlib import Path
from typing import Any
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from strix.config import codex, load_settings, persist_current
from strix.core.paths import run_dir_for
from strix.interface.cli_args import parse_arguments
from strix.core.paths import RUNS_DIR_NAME, run_dir_for
from strix.interface.cli_args import (
FAIL_ON_SEVERITIES,
ResumeError,
load_resume_state,
parse_arguments,
resume_run_list_message,
)
from strix.interface.environment import (
check_docker_installed,
pull_docker_image,
@@ -43,7 +50,7 @@ from strix.interface.utils import (
)
from strix.llm.warmup import start_import_warmup, wait_for_import_warmup
from strix.telemetry import posthog, report_error, scarf, set_scan_phase
from strix.telemetry.logging import configure_dependency_logging
from strix.telemetry.logging import setup_console_logging
BEDROCK_MODEL_PREFIX = "bedrock/"
@@ -315,6 +322,30 @@ def display_completion_message(args: argparse.Namespace, results_path: Path) ->
notify_update(console)
def findings_fail_build(reports: list[dict[str, Any]], fail_on: str | None) -> bool:
"""Whether headless findings should exit 2 under the ``--fail-on`` threshold.
With no threshold any finding fails. Otherwise a finding fails when its
severity is at or above the threshold. A severity outside the known scale
fails too, so a gate never passes on a value it cannot rank. ``none`` is a
known level below ``info`` and only fails without a threshold.
"""
if not reports:
return False
if fail_on is None:
return True
threshold = FAIL_ON_SEVERITIES.index(fail_on)
for report in reports:
severity = str(report.get("severity") or "").strip().lower()
if severity == "none":
continue
if severity not in FAIL_ON_SEVERITIES:
return True
if FAIL_ON_SEVERITIES.index(severity) <= threshold:
return True
return False
def _print_error_panel(title: str, message: str) -> None:
console = Console()
error_text = Text()
@@ -366,12 +397,38 @@ def _print_model_connection_error(exc: BaseException, model_name: str) -> None:
console.print()
def _bootstrap_scan(args: argparse.Namespace) -> None:
"""Warm up the model and prepare the run for a non-interactive scan.
def _print_cli_error(message: str) -> None:
Console(stderr=True, soft_wrap=True).print(
f"strix: error: {message}", markup=False, highlight=False
)
Interactive launches skip this: the model preflight and run preparation
happen inside the TUI so the interface paints immediately instead of
waiting on a model round trip.
def _pick_run_to_resume(args: argparse.Namespace) -> None:
"""A bare --resume: let the user pick a run, then load it like --resume <name>."""
from strix.interface.resume_picker import PickerUnavailableError, pick_run
from strix.report.runs import list_run_summaries
try:
chosen = pick_run(list_run_summaries(), runs_dir=RUNS_DIR_NAME)
except PickerUnavailableError as exc:
_print_cli_error(resume_run_list_message(f"{exc}."))
sys.exit(2)
if chosen is None:
Console().print("No run selected.", style="dim")
sys.exit(0)
args.resume = chosen.run_name
try:
load_resume_state(args)
except ResumeError as exc:
_print_cli_error(str(exc))
sys.exit(2)
def _bootstrap_scan(args: argparse.Namespace) -> None:
"""Warm up the model and prepare the run before the interface starts.
Start-screen launches skip this: they verify the model and prepare the
run once the user has entered a target.
"""
set_scan_phase("preflight")
try:
@@ -390,12 +447,22 @@ def _bootstrap_scan(args: argparse.Namespace) -> None:
telemetry_start(args)
def main() -> None:
configure_dependency_logging()
def _force_utf8_streams() -> None:
for stream in (sys.stdout, sys.stderr):
reconfigure = getattr(stream, "reconfigure", None)
if reconfigure is None:
continue
with contextlib.suppress(ValueError, OSError):
reconfigure(encoding="utf-8")
def main() -> None:
if sys.platform == "win32":
_force_utf8_streams()
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
setup_console_logging()
if len(sys.argv) == 2 and sys.argv[1] in ("-h", "--help"):
try:
parse_arguments()
@@ -441,6 +508,9 @@ def main() -> None:
restart_after_update()
sys.exit(0)
if args.resume_picker:
_pick_run_to_resume(args)
check_docker_installed()
pull_docker_image()
validate_environment()
@@ -448,7 +518,7 @@ def main() -> None:
# Everything below imports the scan engine; do not race the warm-up thread.
wait_for_import_warmup()
if args.non_interactive:
if args.non_interactive or not args.needs_setup:
_bootstrap_scan(args)
from strix.report.state import get_global_report_state
@@ -501,7 +571,7 @@ def main() -> None:
if args.non_interactive:
report_state = get_global_report_state()
if report_state and report_state.vulnerability_reports:
if report_state and findings_fail_build(report_state.vulnerability_reports, args.fail_on):
sys.exit(2)
+407
View File
@@ -0,0 +1,407 @@
"""Inline arrow-key picker for ``strix --resume`` with no run name.
Draws a short list in the normal terminal (no alternate screen), redraws it in
place on every key, and returns the chosen run. Keys: up/down, page up/down,
home/end, enter, esc, typing filters the rows, backspace edits the filter.
"""
from __future__ import annotations
import os
import sys
from datetime import UTC, datetime
from typing import TYPE_CHECKING, TextIO
from rich.console import Console
from rich.text import Text
if TYPE_CHECKING:
from collections.abc import Callable
from strix.report.runs import RunSummary
KEY_UP = "up"
KEY_DOWN = "down"
KEY_PAGE_UP = "pageup"
KEY_PAGE_DOWN = "pagedown"
KEY_HOME = "home"
KEY_END = "end"
KEY_ENTER = "enter"
KEY_ESCAPE = "escape"
KEY_BACKSPACE = "backspace"
KEY_INTERRUPT = "interrupt"
_CSI_KEYS = {
"A": KEY_UP,
"B": KEY_DOWN,
"H": KEY_HOME,
"F": KEY_END,
"1~": KEY_HOME,
"4~": KEY_END,
"5~": KEY_PAGE_UP,
"6~": KEY_PAGE_DOWN,
"7~": KEY_HOME,
"8~": KEY_END,
}
_WINDOWS_KEYS = {
"H": KEY_UP,
"P": KEY_DOWN,
"I": KEY_PAGE_UP,
"Q": KEY_PAGE_DOWN,
"G": KEY_HOME,
"O": KEY_END,
}
_CONTROL_KEYS = {
"\r": KEY_ENTER,
"\n": KEY_ENTER,
"\x1b": KEY_ESCAPE,
"\x7f": KEY_BACKSPACE,
"\x08": KEY_BACKSPACE,
"\x03": KEY_INTERRUPT,
}
_CURSOR_HIDE = "\x1b[?25l"
_CURSOR_SHOW = "\x1b[?25h"
_CLEAR_BELOW = "\x1b[J"
_GREEN = "#22c55e"
_AMBER = "#f59e0b"
_STATUS_STYLES = {
"completed": _GREEN,
"running": _GREEN,
"interrupted": _AMBER,
"stopped": _AMBER,
"failed": "#ef4444",
}
_NO_STATE = "no state"
_CURSOR = " \u276f "
_STARTED_WIDTH = 14
_FINDINGS_WIDTH = 8
_MIN_TARGET_WIDTH = 12
_MAX_RUN_WIDTH = 40
_CHROME_LINES = 8
_MAX_VISIBLE = 8
def _utf8_length(lead: bytes) -> int:
byte = lead[0] if lead else 0
if byte >= 0xF0:
return 4
if byte >= 0xE0:
return 3
if byte >= 0xC0:
return 2
return 1
class PickerUnavailableError(RuntimeError):
"""The picker needs an interactive terminal on stdin and stdout."""
if sys.platform == "win32":
import msvcrt
def read_raw_key(_stream: TextIO) -> str:
char = str(msvcrt.getwch())
if char in ("\x00", "\xe0"):
return _WINDOWS_KEYS.get(str(msvcrt.getwch()), "")
return char
else:
import select
import termios
import tty
def read_raw_key(stream: TextIO) -> str:
fd = stream.fileno()
saved = termios.tcgetattr(fd)
try:
tty.setcbreak(fd, termios.TCSANOW)
first = os.read(fd, 1)
char = (first + os.read(fd, _utf8_length(first) - 1)).decode("utf-8", "replace")
if char != "\x1b":
return char
sequence = ""
while select.select([fd], [], [], 0.05)[0]:
sequence += os.read(fd, 1).decode("utf-8", "replace")
if (sequence.startswith("[") and sequence[-1].isalpha()) or sequence.endswith("~"):
break
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, saved)
if not sequence:
return "\x1b"
return _CSI_KEYS.get(sequence[1:], "") if sequence[0] in "[O" else ""
def read_key(stream: TextIO) -> str:
return translate_key(read_raw_key(stream))
def translate_key(raw: str) -> str:
return _CONTROL_KEYS.get(raw, raw)
def relative_time(stamp: str, now: datetime | None = None) -> str:
try:
started = datetime.fromisoformat(stamp)
except ValueError:
return stamp[:16] if stamp else "unknown"
if started.tzinfo is None:
started = started.replace(tzinfo=UTC)
now = now or datetime.now(UTC)
seconds = (now - started).total_seconds()
for limit, unit, label in (
(60, 1, ""),
(3600, 60, "min"),
(86400, 3600, "h"),
(7 * 86400, 86400, "d"),
):
if seconds < limit:
return "just now" if not label else f"{int(seconds // unit)} {label} ago"
local = started.astimezone()
if local.year == now.astimezone().year:
return f"{local:%b} {local.day}, {local:%H:%M}"
return f"{local:%b} {local.day}, {local.year}"
def filter_runs(runs: list[RunSummary], needle: str) -> list[RunSummary]:
needle = needle.strip().lower()
if not needle:
return list(runs)
return [
run
for run in runs
if needle in run.run_name.lower()
or needle in run.target.lower()
or needle in run.status.lower()
]
def _fit(text: str, width: int) -> str:
if len(text) <= width:
return text.ljust(width)
return text[: width - 1] + "\u2026"
def _status_text(run: RunSummary) -> str:
return run.status if run.resumable else f"{run.status} \u00b7 {_NO_STATE}"
class ResumePicker:
def __init__(
self,
runs: list[RunSummary],
*,
console: Console,
runs_dir: str,
now: datetime | None = None,
) -> None:
self.runs = runs
self.console = console
self.runs_dir = runs_dir
self.now = now
self.filter = ""
self.cursor = 0
self.offset = 0
self.notice = ""
self._drawn = 0
@property
def rows(self) -> list[RunSummary]:
return filter_runs(self.runs, self.filter)
def _visible(self) -> int:
return max(3, min(len(self.rows), _MAX_VISIBLE, self.console.height - _CHROME_LINES))
def _columns(self) -> tuple[int, int, int]:
width = max(40, self.console.width - 1)
status_width = max(len("status"), *(len(_status_text(run)) for run in self.runs))
run_width = min(_MAX_RUN_WIDTH, max(len("run"), *(len(run.run_name) for run in self.runs)))
fixed = len(_CURSOR) + _STARTED_WIDTH + _FINDINGS_WIDTH + status_width + 4 * 2
target_width = width - fixed - run_width
if target_width < _MIN_TARGET_WIDTH:
run_width = max(8, run_width + target_width - _MIN_TARGET_WIDTH)
target_width = width - fixed - run_width
return max(_MIN_TARGET_WIDTH, target_width), status_width, run_width
def _scroll(self) -> range:
rows = self.rows
visible = self._visible()
self.cursor = max(0, min(self.cursor, len(rows) - 1))
if self.cursor < self.offset:
self.offset = self.cursor
elif self.cursor >= self.offset + visible:
self.offset = self.cursor - visible + 1
self.offset = max(0, min(self.offset, max(0, len(rows) - visible)))
return range(self.offset, min(len(rows), self.offset + visible))
def render(self) -> list[Text]:
rows = self.rows
widths = self._columns()
window = self._scroll()
title = Text()
title.append(" Resume a run", style="bold")
title.append(f" {len(self.runs)} runs in ./{self.runs_dir}", style="dim")
if self.filter:
title.append(" search: ", style="dim")
title.append(self.filter)
header = Text(
" " * len(_CURSOR)
+ self._cells("started", "target", "status", "findings", "run", widths),
style="dim",
)
lines = [title, Text(), header]
if not rows:
lines.append(Text(f" no runs match {self.filter!r}", style="dim"))
if window.start:
lines.append(Text(f" \u2026 {window.start} more above", style="dim"))
lines.extend(self._row(rows[index], index == self.cursor, widths) for index in window)
if window.stop < len(rows):
lines.append(Text(f" \u2026 {len(rows) - window.stop} more below", style="dim"))
footer = Text(" ")
if self.notice:
footer.append(self.notice, style=_AMBER)
else:
footer.append(
"\u2191\u2193 move enter resume type to search esc cancel", style="dim"
)
lines.extend([Text(), footer])
for line in lines:
line.truncate(self.console.width - 1)
return lines
@staticmethod
def _cells(
started: str,
target: str,
status: str,
findings: str,
run: str,
widths: tuple[int, int, int],
) -> str:
target_width, status_width, run_width = widths
return " ".join(
[
_fit(started, _STARTED_WIDTH),
_fit(target, target_width),
_fit(status, status_width),
_fit(findings, _FINDINGS_WIDTH),
_fit(run, run_width),
]
)
def _row(self, run: RunSummary, selected: bool, widths: tuple[int, int, int]) -> Text:
target_width, status_width, run_width = widths
primary = "bold" if selected else ""
muted = "" if selected else "dim"
status_style = _STATUS_STYLES.get(run.status, "")
if not run.resumable:
primary = muted = status_style = "dim"
line = Text()
line.append(_CURSOR if selected else " " * len(_CURSOR), style=_GREEN)
line.append(_fit(relative_time(run.started_at, self.now), _STARTED_WIDTH), style=muted)
line.append(" ")
line.append(_fit(run.target, target_width), style=primary)
line.append(" ")
line.append(_fit(_status_text(run), status_width), style=status_style)
line.append(" ")
line.append(_fit(str(run.findings), _FINDINGS_WIDTH), style=muted)
line.append(" ")
line.append(_fit(run.run_name, run_width), style=muted)
return line
def draw(self) -> None:
self.clear()
lines = self.render()
for line in lines:
self.console.print(line, soft_wrap=True, overflow="crop", end="\n")
self.console.file.flush()
self._drawn = len(lines)
def clear(self) -> None:
if self._drawn:
self.console.file.write(f"\x1b[{self._drawn}A\r{_CLEAR_BELOW}")
self.console.file.flush()
self._drawn = 0
def handle(self, key: str) -> tuple[bool, RunSummary | None]:
"""Apply one key: (done, run), where done with no run means cancelled."""
self.notice = ""
if key == KEY_INTERRUPT:
raise KeyboardInterrupt
if key == KEY_ESCAPE:
if not self.filter:
return True, None
self.filter = ""
self.cursor = 0
return False, None
if key == KEY_ENTER:
return self._choose()
if key == KEY_BACKSPACE:
self.filter = self.filter[:-1]
self.cursor = 0
elif len(key) == 1 and key.isprintable():
self.filter += key
self.cursor = 0
else:
self._move(key)
self.cursor = max(0, min(self.cursor, max(0, len(self.rows) - 1)))
return False, None
def _choose(self) -> tuple[bool, RunSummary | None]:
rows = self.rows
if not rows:
return False, None
run = rows[self.cursor]
if not run.resumable:
self.notice = f"{run.run_name} has no saved agent state to resume from"
return False, None
return True, run
def _move(self, key: str) -> None:
steps = {
KEY_UP: -1,
KEY_DOWN: 1,
KEY_PAGE_UP: -self._visible(),
KEY_PAGE_DOWN: self._visible(),
}
if key in steps:
self.cursor += steps[key]
elif key == KEY_HOME:
self.cursor = 0
elif key == KEY_END:
self.cursor = len(self.rows) - 1
def run(self, next_key: Callable[[], str]) -> RunSummary | None:
self.console.file.write(_CURSOR_HIDE)
try:
self.draw()
while True:
done, chosen = self.handle(next_key())
if done:
return chosen
self.draw()
finally:
self.clear()
self.console.file.write(_CURSOR_SHOW)
self.console.file.flush()
def pick_run(
runs: list[RunSummary],
*,
runs_dir: str,
console: Console | None = None,
stdin: TextIO | None = None,
) -> RunSummary | None:
stdin = stdin or sys.stdin
console = console or Console()
if not (hasattr(stdin, "isatty") and stdin.isatty() and console.is_terminal):
raise PickerUnavailableError(
"--resume needs a run name when there is no interactive terminal"
)
picker = ResumePicker(runs, console=console, runs_dir=runs_dir)
return picker.run(lambda: read_key(stdin))
+41
View File
@@ -11,6 +11,7 @@ from __future__ import annotations
import asyncio
import logging
import unicodedata
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
@@ -58,6 +59,45 @@ class ModelConnectionError(RuntimeError):
self.model_name = model_name
def _first_non_ascii(value: str) -> tuple[int, str] | None:
for position, char in enumerate(value):
if ord(char) > 0x7F:
return position, char
return None
def _header_candidates(
prefix: str, model: str | None, api_key: str | None, extra_headers: dict[str, str] | None
) -> list[tuple[str, str]]:
candidates: list[tuple[str, str]] = []
if api_key and not codex.subscription_model(model):
candidates.append((f"{prefix}LLM_API_KEY", api_key))
for header, value in (extra_headers or {}).items():
candidates.append((f"{prefix}LLM_EXTRA_HEADERS header name {header!r}", header))
candidates.append((f"{prefix}LLM_EXTRA_HEADERS value for {header!r}", value))
return candidates
def check_header_safe_credentials(settings: Settings) -> None:
llm = settings.llm
dedupe = settings.dedupe
candidates = _header_candidates("", llm.model, llm.api_key, llm.extra_headers)
if dedupe.model:
candidates += _header_candidates(
"DEDUPE_", dedupe.model, (dedupe.api_key or "").strip(), dedupe.extra_headers
)
for setting, value in candidates:
found = _first_non_ascii(value)
if found is None:
continue
position, char = found
raise ValueError(
f"{setting} contains a character that cannot be sent in an HTTP header: "
f"U+{ord(char):04X} ({unicodedata.name(char, 'unnamed character')}) "
f"at position {position + 1} of {len(value)}. Re-enter the value without it."
)
async def preflight_model_connection(
model_name: str,
*,
@@ -70,6 +110,7 @@ async def preflight_model_connection(
from strix.core.inputs import make_model_settings
resolved_settings = load_settings() if settings is None else settings
check_header_safe_credentials(resolved_settings)
configure_sdk_model_defaults(resolved_settings)
model = StrixProvider().get_model(model_name)
request_settings = make_model_settings(
@@ -152,11 +152,6 @@ class TuiController:
]
self.notify_changed()
def begin_preparation(self) -> None:
"""Mark a directly-launched run as preparing behind the live TUI."""
self.scan_state = "preparing"
self.notify_changed()
def fail_preparation(self, detail: str) -> None:
self.scan_state = "failed"
self.error = detail
+1 -2
View File
@@ -209,8 +209,7 @@ func (m *Model) ensureAgentVisible() {
m.agentOffset = 0
return
}
_, _, _, agentHeight := m.sidebarHeights()
rows := max(1, agentHeight-4)
rows := m.agentPageSize()
row := selectedAgentRow(entries, m.selectedAgent)
if row < m.agentOffset {
m.agentOffset = row
+9 -8
View File
@@ -27,11 +27,14 @@ func mcpModel(t *testing.T) Model {
func TestMcpPanelShowsHealthyAndOffline(t *testing.T) {
m := mcpModel(t)
out := ansi.Strip(m.mcpConnectionsView(40, 6))
for _, want := range []string{"MCP Connections (2)", "supabase", "3 tools", "vercel", "offline"} {
for _, want := range []string{"supabase", "3 tools", "vercel", "offline"} {
if !strings.Contains(out, want) {
t.Fatalf("panel missing %q:\n%s", want, out)
}
}
if sidebar := ansi.Strip(m.sidebarView(26, 40)); !strings.Contains(sidebar, "MCP (2)") {
t.Fatalf("sidebar header did not carry the connection count:\n%s", sidebar)
}
}
// A roster longer than the panel height shows a window of rows rather than every
@@ -46,9 +49,8 @@ func TestMcpPanelWindowsLargeRosterAndCountsAll(t *testing.T) {
}
m.snapshot.Connections = conns
// rows = 6 → one header line + five roster rows.
out := ansi.Strip(m.mcpConnectionsView(40, 6))
if !strings.Contains(out, "MCP Connections (12)") {
if !strings.Contains(ansi.Strip(m.sidebarView(26, 40)), "MCP (12)") {
t.Fatalf("header did not carry the full connection count:\n%s", out)
}
if !strings.Contains(out, "conn-00") {
@@ -58,14 +60,13 @@ func TestMcpPanelWindowsLargeRosterAndCountsAll(t *testing.T) {
t.Fatalf("a roster past the panel height should be windowed, not fully drawn:\n%s", out)
}
if got := strings.Count(out, "\n") + 1; got != 6 {
t.Fatalf("panel rendered %d lines, want 6 (header + five rows)", got)
t.Fatalf("panel rendered %d lines, want 6", got)
}
// Scrolling the roster brings the tail into view while the header count holds.
m.mcpOffset = 7
m.mcpOffset = 6
scrolled := ansi.Strip(m.mcpConnectionsView(40, 6))
if !strings.Contains(scrolled, "conn-11") || !strings.Contains(scrolled, "MCP Connections (12)") {
t.Fatalf("scrolled window did not reveal the tail with the count intact:\n%s", scrolled)
if !strings.Contains(scrolled, "conn-11") {
t.Fatalf("scrolled window did not reveal the tail:\n%s", scrolled)
}
}
+4 -1
View File
@@ -100,6 +100,9 @@ type Model struct {
filtered []string
cursor int
collapsedAgents map[string]bool
collapsedPanels map[sidebarPanel]bool
zoomedPanel sidebarPanel
sidebarHidden bool
expandedEvents map[string]bool
blockCache map[string]renderedBlock
eventSpans []eventSpan
@@ -278,7 +281,7 @@ func New(client *Client) Model {
input.Focus()
return Model{
client: client, input: input, viewport: viewport.New(80, 20), vulnViewport: viewport.New(80, 20),
collapsedAgents: map[string]bool{}, expandedEvents: map[string]bool{}, blockCache: map[string]renderedBlock{}, showSplash: true, splashStarted: time.Now(), followOutput: true,
collapsedAgents: map[string]bool{}, collapsedPanels: map[sidebarPanel]bool{}, zoomedPanel: panelNone, expandedEvents: map[string]bool{}, blockCache: map[string]renderedBlock{}, showSplash: true, splashStarted: time.Now(), followOutput: true,
collectionRevisions: map[string]int{}, collectionAssemblies: map[string]*collectionAssembly{}, resyncRequested: map[string]bool{}, resyncRequests: map[string]string{},
seenMessages: map[string]bool{},
}
@@ -890,6 +890,14 @@ func TestRunningViewerShowsCompleteWrappedURL(t *testing.T) {
if want := strings.Count(model.viewerView(model.viewerContentWidth()), "\n") + 3; model.viewerHeight() != want {
t.Fatalf("viewer height = %d, want %d", model.viewerHeight(), want)
}
raw := model.viewerView(18)
if got, want := strings.Count(raw, ansi.SetHyperlink(url)), len(urlLines); got != want {
t.Fatalf("every wrapped line should link to the full URL: got %d links for %d lines in %q", got, want, raw)
}
if got := strings.Count(raw, ansi.ResetHyperlink()); got != len(urlLines) {
t.Fatalf("hyperlinks not closed: %d resets for %d lines", got, len(urlLines))
}
}
func TestVerticalScrollbarThumbTracksScrollOffset(t *testing.T) {
@@ -951,6 +959,17 @@ func TestFillBackgroundRestoresBaseForegroundAfterReset(t *testing.T) {
}
}
func TestFillBackgroundRestoresBaseColorsAfterBareReset(t *testing.T) {
view := "\x1b[38;2;115;115;115mModel\x1b[m padding\x1b[0m"
filled := fillBackground(view)
if !strings.Contains(filled, "\x1b[m"+baseFrameColors+" padding") {
t.Fatalf("base colors not restored after bare reset: %q", filled)
}
if strings.Contains(filled, "\x1b[m ") {
t.Fatalf("cells after a bare reset show the terminal background: %q", filled)
}
}
func TestMainTraceTreeAndFindingsRenderScrollbars(t *testing.T) {
model := New(nil)
model.width, model.height = 150, 35
+311
View File
@@ -0,0 +1,311 @@
package app
import (
"fmt"
"strings"
"github.com/charmbracelet/lipgloss"
)
type sidebarPanel int
const (
panelNone sidebarPanel = iota - 1
panelAgents
panelFindings
panelMcp
panelStats
)
const (
panelOpenGlyph = "▾"
panelCollapsedGlyph = "▸"
panelZoomGlyph = "⤢"
panelUnzoomGlyph = "⤡"
sidebarHideGlyph = "»"
sidebarShowGlyph = "«"
panelGlyphZone = 3
)
type panelRect struct {
panel sidebarPanel
top int
height int
}
func (m Model) panelControls() bool {
return len(m.snapshot.Vulnerabilities) > 0 || len(m.snapshot.Connections) > 0
}
func (m Model) panelShrunk(panel sidebarPanel) bool {
if panel == panelStats || !m.panelControls() {
return false
}
return m.collapsedPanels[panel] || (m.zoomedPanel != panelNone && m.zoomedPanel != panel)
}
func (m Model) panelHeight(panel sidebarPanel, open int) int {
if m.panelShrunk(panel) {
return 1
}
return open
}
func panelFocus(panel sidebarPanel) (focusMode, bool) {
switch panel {
case panelAgents:
return focusAgents, true
case panelFindings:
return focusVulnerabilities, true
case panelMcp:
return focusMcp, true
default:
return focusInput, false
}
}
func (m Model) sidebarPanels() []panelRect {
statsHeight, vulnHeight, mcpHeight, agentHeight := m.sidebarHeights()
top := m.viewerHeight()
rects := []panelRect{{panelAgents, top, agentHeight}}
top += agentHeight
if vulnHeight > 0 {
rects = append(rects, panelRect{panelFindings, top, vulnHeight})
top += vulnHeight
}
if mcpHeight > 0 {
rects = append(rects, panelRect{panelMcp, top, mcpHeight})
top += mcpHeight
}
top += m.sidebarGap()
return append(rects, panelRect{panelStats, top, statsHeight})
}
func (m Model) sidebarGap() int {
statsHeight, vulnHeight, mcpHeight, agentHeight := m.sidebarHeights()
return max(0, m.height-m.viewerHeight()-agentHeight-vulnHeight-mcpHeight-statsHeight)
}
func (m Model) panelAt(y int) (panelRect, bool) {
for _, rect := range m.sidebarPanels() {
if y >= rect.top && y < rect.top+rect.height {
return rect, true
}
}
return panelRect{}, false
}
func (m Model) panelTop(panel sidebarPanel) int {
for _, rect := range m.sidebarPanels() {
if rect.panel == panel {
return rect.top
}
}
return 0
}
func (m Model) panelTitle(panel sidebarPanel) string {
switch panel {
case panelAgents:
return fmt.Sprintf("Agents (%d)", len(m.snapshot.Agents))
case panelFindings:
return fmt.Sprintf("Findings (%d)", len(m.snapshot.Vulnerabilities))
case panelMcp:
return fmt.Sprintf("MCP (%d)", len(m.snapshot.Connections))
default:
return m.snapshot.Model
}
}
func (m Model) panelHeader(panel sidebarPanel, width int) string {
style := lipgloss.NewStyle().Foreground(dim)
if !m.panelControls() {
return truncate(style.Render(m.panelTitle(panel)), max(1, width))
}
glyph := panelZoomGlyph
if m.zoomedPanel == panel {
glyph = panelUnzoomGlyph
}
label := truncate(style.Render(panelOpenGlyph+" "+m.panelTitle(panel)), max(1, width-2))
gap := max(1, width-lipgloss.Width(label)-1)
return label + strings.Repeat(" ", gap) + style.Render(glyph)
}
func (m Model) collapsedPanelRow(panel sidebarPanel, width int) string {
title := m.panelTitle(panel)
if panel != panelStats {
title = panelCollapsedGlyph + " " + title
}
label := lipgloss.NewStyle().Foreground(dim).Render(title)
return " " + truncate(label, max(1, width-1))
}
func (m Model) panelBox(panel sidebarPanel, body string, width, height int, focused bool) string {
if height <= 1 {
return m.collapsedPanelRow(panel, width)
}
border := dark
if focused {
border = green
}
content := body
if panel != panelStats {
content = m.panelHeader(panel, width-4)
if body != "" {
content += "\n\n" + body
}
}
return lipgloss.NewStyle().Width(width-2).Height(height-2).Border(lipgloss.RoundedBorder()).
BorderForeground(border).Padding(0, 1).Render(content)
}
const (
toggleButtonWidth = 3
railButtonWidth = 5
railButtonHeight = 3
sidebarRailWidth = railButtonWidth + 1
)
var toggleButtonFill = lipgloss.Color("#262626")
func sidebarToggleButton(glyph string) string {
return lipgloss.NewStyle().
Background(toggleButtonFill).
Foreground(brightWhite).
Bold(true).
Width(toggleButtonWidth).
Align(lipgloss.Center).
Render(glyph)
}
func (m Model) railVisible() bool {
return m.sidebarHidden && m.width >= 120
}
func (m Model) sidebarRail(height int) string {
button := lipgloss.NewStyle().
Background(toggleButtonFill).
Foreground(brightWhite).
Bold(true).
Width(railButtonWidth).
Height(railButtonHeight).
Align(lipgloss.Center, lipgloss.Center).
Render(sidebarShowGlyph)
return lipgloss.NewStyle().
Width(sidebarRailWidth).
Height(height).
Align(lipgloss.Right).
Render(button)
}
func (m Model) toggleButtonHit(x, y int) bool {
if m.railVisible() {
return y < railButtonHeight && x >= m.width-railButtonWidth
}
return y == 1 && x >= m.width-2-toggleButtonWidth && x < m.width-2
}
func (m Model) viewerBox(width int) string {
textWidth := max(1, width-toggleButtonWidth-1)
text := m.viewerView(textWidth)
rows := strings.Count(text, "\n") + 1
return lipgloss.JoinHorizontal(
lipgloss.Top,
fixedPanelBody(text, textWidth, rows),
" ",
sidebarToggleButton(sidebarHideGlyph),
)
}
func (m *Model) toggleSidebar() {
m.sidebarHidden = !m.sidebarHidden
m.resizeViewport()
m.panelsChanged()
}
func (m *Model) togglePanelCollapsed(panel sidebarPanel) {
if m.collapsedPanels[panel] {
delete(m.collapsedPanels, panel)
} else {
m.collapsedPanels[panel] = true
if m.zoomedPanel == panel {
m.zoomedPanel = panelNone
}
}
m.panelsChanged()
}
func (m *Model) togglePanelZoom(panel sidebarPanel) {
if m.zoomedPanel == panel {
m.zoomedPanel = panelNone
} else {
m.zoomedPanel = panel
delete(m.collapsedPanels, panel)
}
m.panelsChanged()
}
func (m *Model) revealPanel(panel sidebarPanel) {
delete(m.collapsedPanels, panel)
m.zoomedPanel = panelNone
m.panelsChanged()
}
func (m *Model) clickPanel(rect panelRect, x, y int) bool {
switch {
case rect.panel == panelStats || !m.panelControls():
return false
case rect.height <= 1 && !m.panelShrunk(rect.panel):
m.togglePanelZoom(rect.panel)
case rect.height <= 1:
m.revealPanel(rect.panel)
case y-rect.top != 1:
return false
case x >= m.width-2-panelGlyphZone:
m.togglePanelZoom(rect.panel)
default:
m.togglePanelCollapsed(rect.panel)
}
if focus, ok := panelFocus(rect.panel); ok && !m.panelShrunk(rect.panel) {
m.focus = focus
m.input.Blur()
}
return true
}
func (m *Model) panelsChanged() {
showSidebar, _, _, _ := m.layout()
for _, rect := range m.sidebarPanels() {
if focus, ok := panelFocus(rect.panel); ok && m.focus == focus && (!showSidebar || rect.height <= 1) {
m.focus = focusInput
m.input.Focus()
}
}
m.ensureAgentVisible()
totalRows, _ := m.vulnerabilityScrollRows()
m.vulnOffset = clampVulnerabilityOffset(m.vulnOffset, totalRows, m.vulnerabilityPageSize())
m.mcpOffset = m.clampMcpOffset(m.mcpOffset)
}
func (m *Model) scrollPanel(panel sidebarPanel, delta int) {
switch panel {
case panelAgents:
m.focus = focusAgents
m.input.Blur()
rows := m.agentPageSize()
total := len(agentTreeEntries(m.snapshot.Agents, m.collapsedAgents))
m.agentOffset = min(max(0, total-rows), max(0, m.agentOffset+delta))
m.keepAgentSelectionInWindow()
m.refreshViewport()
case panelFindings:
m.focus = focusVulnerabilities
m.input.Blur()
totalRows, _ := m.vulnerabilityScrollRows()
m.vulnOffset = min(max(0, totalRows-m.vulnerabilityPageSize()), max(0, m.vulnOffset+delta))
m.keepVulnerabilitySelectionInWindow()
case panelMcp:
m.focus = focusMcp
m.input.Blur()
m.mcpOffset = m.clampMcpOffset(m.mcpOffset + delta)
}
}
@@ -0,0 +1,357 @@
package app
import (
"fmt"
"strings"
"testing"
tea "github.com/charmbracelet/bubbletea"
"github.com/charmbracelet/lipgloss"
"github.com/charmbracelet/x/ansi"
"github.com/usestrix/strix/tui/internal/protocol"
)
func panelsModel(t *testing.T) Model {
t.Helper()
m := New(nil)
m.width, m.height = 130, 40
m.showSplash = false
m.ready = true
root := "root"
agents := []protocol.Agent{{ID: root, Name: "Root Agent", Status: "running"}}
for i := 0; i < 30; i++ {
name := fmt.Sprintf("Worker %02d", i)
agents = append(agents, protocol.Agent{ID: name, Name: name, ParentID: &root, Status: "running"})
}
conns := make([]protocol.Connection, 0, 12)
for i := 0; i < 12; i++ {
conns = append(conns, protocol.Connection{Name: fmt.Sprintf("conn-%02d", i), ToolCount: 2})
}
url := "http://127.0.0.1:57388/?token=abc"
m.snapshot = protocol.Snapshot{
ScanStarted: true, ScanState: "running", Agents: agents, Connections: conns,
ViewerStatus: "running", ViewerURL: &url, Model: "openrouter/z-ai/glm-5.3",
Vulnerabilities: []map[string]any{{"id": "v1", "title": "Finding one", "severity": "high"}},
}
m.resizeViewport()
return m
}
func click(t *testing.T, m Model, x, y int) Model {
t.Helper()
updated, _ := m.updateMouse(tea.MouseMsg{X: x, Y: y, Button: tea.MouseButtonLeft, Action: tea.MouseActionPress})
return updated.(Model)
}
func panelRectOf(t *testing.T, m Model, panel sidebarPanel) panelRect {
t.Helper()
for _, rect := range m.sidebarPanels() {
if rect.panel == panel {
return rect
}
}
t.Fatalf("panel %d not in sidebar", panel)
return panelRect{}
}
func TestSidebarPanelsRenderHeadersWithControls(t *testing.T) {
m := panelsModel(t)
_, sidebarWidth, _, _ := m.layout()
view := ansi.Strip(m.sidebarView(sidebarWidth, m.height))
for _, want := range []string{"▾ Agents (31)", "▾ Findings (1)", "▾ MCP (12)", "⤢"} {
if !strings.Contains(view, want) {
t.Fatalf("sidebar missing %q:\n%s", want, view)
}
}
if strings.Contains(view, "⤡") {
t.Fatalf("nothing is zoomed, yet a restore glyph is drawn:\n%s", view)
}
}
func TestClickingPanelHeaderCollapsesAndExpands(t *testing.T) {
m := panelsModel(t)
_, _, chatWidth, _ := m.layout()
rect := panelRectOf(t, m, panelFindings)
before := panelRectOf(t, m, panelAgents).height
m = click(t, m, chatWidth+4, rect.top+1)
collapsed := panelRectOf(t, m, panelFindings)
if collapsed.height != 1 || !m.collapsedPanels[panelFindings] {
t.Fatalf("header click did not collapse the panel: %+v", collapsed)
}
if got := panelRectOf(t, m, panelAgents).height; got <= before {
t.Fatalf("agent panel did not grow into the freed rows: %d -> %d", before, got)
}
_, sidebarWidth, _, _ := m.layout()
view := ansi.Strip(m.sidebarView(sidebarWidth, m.height))
if !strings.Contains(view, "▸ Findings (1)") || strings.Contains(view, "Finding one") {
t.Fatalf("collapsed panel should be a one-line header without its rows:\n%s", view)
}
m = click(t, m, chatWidth+4, collapsed.top)
if panelRectOf(t, m, panelFindings).height == 1 || m.collapsedPanels[panelFindings] {
t.Fatalf("clicking the collapsed row did not expand the panel")
}
if m.focus != focusVulnerabilities {
t.Fatalf("expanded panel did not take focus: %v", m.focus)
}
}
func TestClickingZoomGlyphGivesPanelTheSidebar(t *testing.T) {
m := panelsModel(t)
_, sidebarWidth, chatWidth, _ := m.layout()
rect := panelRectOf(t, m, panelMcp)
m = click(t, m, m.width-3, rect.top+1)
if m.zoomedPanel != panelMcp {
t.Fatalf("zoom glyph click did not zoom the panel: %d", m.zoomedPanel)
}
for _, other := range []sidebarPanel{panelAgents, panelFindings} {
if got := panelRectOf(t, m, other).height; got != 1 {
t.Fatalf("panel %d should shrink to its header while another is zoomed, got %d", other, got)
}
}
zoomed := panelRectOf(t, m, panelMcp)
rects := m.sidebarPanels()
if last := rects[len(rects)-1]; last.top+last.height != m.height || zoomed.height < m.height-12 {
t.Fatalf("zoomed panel does not fill the sidebar: %+v screen=%d", rects, m.height)
}
view := ansi.Strip(m.sidebarView(sidebarWidth, m.height))
if !strings.Contains(view, "⤡") || !strings.Contains(view, "conn-11") {
t.Fatalf("zoomed roster should show every connection and the restore glyph:\n%s", view)
}
if m.focus != focusMcp {
t.Fatalf("zoomed panel did not take focus: %v", m.focus)
}
m = click(t, m, m.width-3, zoomed.top+1)
if m.zoomedPanel != panelNone || panelRectOf(t, m, panelAgents).height == 1 {
t.Fatalf("restore glyph did not unzoom")
}
m = click(t, m, m.width-3, panelRectOf(t, m, panelAgents).top+1)
m = click(t, m, chatWidth+4, panelRectOf(t, m, panelFindings).top)
if m.zoomedPanel != panelNone || m.collapsedPanels[panelFindings] {
t.Fatalf("clicking a shrunk row should restore the whole sidebar")
}
}
func TestToggleButtonsHideAndShowSidebar(t *testing.T) {
m := panelsModel(t)
m.focus = focusAgents
showSidebar, sidebarWidth, chatWidth, _ := m.layout()
if !showSidebar {
t.Fatalf("precondition: sidebar visible")
}
rows := strings.Split(ansi.Strip(m.sidebarView(sidebarWidth, m.height)), "\n")
if !strings.HasSuffix(rows[1], " » │") || strings.Contains(rows[2], "»") {
t.Fatalf("viewer panel lacks the one-row hide button:\n%s", strings.Join(rows[:3], "\n"))
}
if strings.Contains(ansi.Strip(m.View()), "«") || strings.Contains(ansi.Strip(m.statusView(chatWidth)), "sidebar") {
t.Fatalf("show button or text hint drawn while the sidebar is visible")
}
m = click(t, m, m.width-4, 1)
if showSidebar, _, width, _ := m.layout(); showSidebar || width != m.width-sidebarRailWidth-1 {
t.Fatalf("hide button click did not hide the sidebar: show=%v chatWidth=%d", showSidebar, width)
}
if m.focus != focusInput {
t.Fatalf("focus stayed on a hidden panel: %v", m.focus)
}
frame := ansi.Strip(m.View())
rows = strings.Split(frame, "\n")
if !strings.HasSuffix(rows[1], " « ") || strings.Contains(rows[0], "«") || strings.Contains(rows[3], "«") || lipgloss.Width(rows[0]) != m.width {
t.Fatalf("hidden sidebar should leave a %d-row show button in the rail:\n%s", railButtonHeight, strings.Join(rows[:4], "\n"))
}
if !m.toggleButtonHit(m.width-railButtonWidth, railButtonHeight-1) || m.toggleButtonHit(m.width-railButtonWidth-1, 0) || m.toggleButtonHit(m.width-1, railButtonHeight) {
t.Fatalf("show button hit zone does not match the drawn button")
}
if strings.Contains(frame, "»") {
t.Fatalf("hide button drawn while the sidebar is hidden")
}
m = click(t, m, m.width-3, 2)
if showSidebar, _, _, _ := m.layout(); !showSidebar {
t.Fatalf("show button click did not bring the sidebar back")
}
}
func TestFocusCyclingSkipsShrunkPanels(t *testing.T) {
m := panelsModel(t)
m.collapsedPanels[panelFindings] = true
seen := map[focusMode]bool{}
for range 6 {
m.cycleFocus(1)
seen[m.focus] = true
}
if seen[focusVulnerabilities] || !seen[focusAgents] || !seen[focusMcp] {
t.Fatalf("tab order wrong with findings collapsed: %v", seen)
}
m.zoomedPanel = panelAgents
seen = map[focusMode]bool{}
for range 6 {
m.cycleFocus(1)
seen[m.focus] = true
}
if seen[focusMcp] || !seen[focusAgents] {
t.Fatalf("tab order wrong with agents zoomed: %v", seen)
}
m.sidebarHidden = true
seen = map[focusMode]bool{}
for range 4 {
m.cycleFocus(1)
seen[m.focus] = true
}
if seen[focusAgents] || len(seen) != 2 {
t.Fatalf("tab order should stay in the chat column with the sidebar hidden: %v", seen)
}
}
func TestScrollbarHitTestFollowsPanelState(t *testing.T) {
m := panelsModel(t)
showSidebar, _, chatWidth, chatHeight := m.layout()
agents := panelRectOf(t, m, panelAgents)
at := func(y int) scrollbarTarget {
return m.scrollbarAt(tea.MouseMsg{X: m.width - 3, Y: y}, showSidebar, chatWidth, chatHeight)
}
if at(agents.top+2) != scrollbarAgents {
t.Fatalf("overflowing agent tree should expose its scrollbar")
}
if at(agents.top+1) != scrollbarNone {
t.Fatalf("the header row is a control, not a scrollbar")
}
findings := panelRectOf(t, m, panelFindings)
if at(findings.top+2) != scrollbarNone {
t.Fatalf("a findings list that fits has no scrollbar to grab")
}
m.collapsedPanels[panelAgents] = true
agents = panelRectOf(t, m, panelAgents)
if agents.height != 1 || at(agents.top) != scrollbarNone {
t.Fatalf("collapsed panel still reports a scrollbar")
}
}
func TestSidebarFitsShortTerminal(t *testing.T) {
m := panelsModel(t)
m.height = 19
m.snapshot.Connections = m.snapshot.Connections[:1]
m.resizeViewport()
_, sidebarWidth, _, _ := m.layout()
rects := m.sidebarPanels()
last := rects[len(rects)-1]
if last.top+last.height > m.height {
t.Fatalf("panels run past the screen: %+v (height %d)", rects, m.height)
}
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)
}
squeezed := panelRectOf(t, m, panelMcp)
if squeezed.height != 1 {
t.Fatalf("expected the MCP panel squeezed to its header, got %+v", squeezed)
}
if last.panel != panelStats || last.height < 3 {
t.Fatalf("Model panel must keep its box on short terminals: %+v", last)
}
m = click(t, m, m.width-6, squeezed.top)
if m.zoomedPanel != panelMcp {
t.Fatalf("clicking a squeezed header should zoom it, zoomed=%v", m.zoomedPanel)
}
}
func TestModelPanelHasNoControlsAndNeverShrinks(t *testing.T) {
m := panelsModel(t)
_, sidebarWidth, _, _ := m.layout()
stats := panelRectOf(t, m, panelStats)
rows := strings.Split(ansi.Strip(m.sidebarView(sidebarWidth, m.height)), "\n")
if first := rows[stats.top+1]; strings.ContainsAny(first, "▾▸⤢⤡") || strings.Contains(first, "Model") || !strings.Contains(first, m.snapshot.Model[:8]) {
t.Fatalf("Model panel should start with the model name and carry no controls: %q", first)
}
before := m.sidebarPanels()
m = click(t, m, m.width-3, stats.top+1)
m = click(t, m, m.width-10, stats.top+1)
if m.zoomedPanel != panelNone || len(m.collapsedPanels) != 0 {
t.Fatalf("clicks on the Model header changed panel state: zoom=%v collapsed=%v", m.zoomedPanel, m.collapsedPanels)
}
m = click(t, m, m.width-3, panelRectOf(t, m, panelAgents).top+1)
if m.zoomedPanel != panelAgents {
t.Fatalf("precondition: agents zoomed")
}
after := panelRectOf(t, m, panelStats)
if after.height != before[len(before)-1].height {
t.Fatalf("Model panel shrank under zoom: %+v -> %+v", before[len(before)-1], after)
}
}
func TestModelPanelSticksToBottom(t *testing.T) {
m := panelsModel(t)
m = click(t, m, m.width-10, panelRectOf(t, m, panelAgents).top+1)
if !m.collapsedPanels[panelAgents] {
t.Fatalf("precondition: agents collapsed")
}
stats := panelRectOf(t, m, panelStats)
if stats.top+stats.height != m.height {
t.Fatalf("Model panel not at the bottom: %+v (height %d)", stats, m.height)
}
_, 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)
}
if !strings.Contains(rows[stats.top+1], m.snapshot.Model[:8]) {
t.Fatalf("Model panel not where its rect says:\n%s", strings.Join(rows[stats.top-2:], "\n"))
}
}
func TestStatsPanelKeepsAssignedHeight(t *testing.T) {
m := panelsModel(t)
m.snapshot.Model = strings.Repeat("openrouter/some-vendor/a-very-long-model-name ", 12)
m.resizeViewport()
_, sidebarWidth, _, _ := m.layout()
statsHeight, _, _, _ := m.sidebarHeights()
if statsHeight != 15 {
t.Fatalf("stats panel should hit its cap, got %d", statsHeight)
}
if got := lipgloss.Height(m.sidebarView(sidebarWidth, m.height)); got != m.height {
t.Fatalf("sidebar renders %d rows, want %d", got, m.height)
}
}
func TestHeadersAreFollowedByABlankRow(t *testing.T) {
m := panelsModel(t)
_, sidebarWidth, _, _ := m.layout()
rows := strings.Split(ansi.Strip(m.sidebarView(sidebarWidth, m.height)), "\n")
for _, panel := range []sidebarPanel{panelAgents, panelFindings, panelMcp} {
rect := panelRectOf(t, m, panel)
header, blank, first := rows[rect.top+1], rows[rect.top+2], rows[rect.top+3]
if !strings.Contains(header, m.panelTitle(panel)) || strings.TrimSpace(strings.Trim(blank, "│")) != "" || strings.TrimSpace(strings.Trim(first, "│")) == "" {
t.Fatalf("panel %d should render header, blank row, content:\n%s", panel, strings.Join(rows[rect.top:rect.top+4], "\n"))
}
}
}
func TestLonePanelHasNoControls(t *testing.T) {
m := panelsModel(t)
m.snapshot.Vulnerabilities = nil
m.snapshot.Connections = nil
_, sidebarWidth, _, _ := m.layout()
view := ansi.Strip(m.sidebarView(sidebarWidth, m.height))
if strings.ContainsAny(view, "▾▸⤢⤡") || !strings.Contains(view, "Agents (31)") {
t.Fatalf("a lone Agents panel should carry no collapse or zoom controls:\n%s", view)
}
rect := panelRectOf(t, m, panelAgents)
m = click(t, m, m.width-3, rect.top+1)
m = click(t, m, m.width-10, rect.top+1)
if m.zoomedPanel != panelNone || len(m.collapsedPanels) != 0 || panelRectOf(t, m, panelAgents) != rect {
t.Fatalf("header clicks changed a lone panel: zoom=%v collapsed=%v", m.zoomedPanel, m.collapsedPanels)
}
if m.sidebarGap() != 0 {
t.Fatalf("lone Agents panel should take the spare rows, gap=%d", m.sidebarGap())
}
m.snapshot.Connections = panelsModel(t).snapshot.Connections
if !strings.Contains(ansi.Strip(m.sidebarView(sidebarWidth, m.height)), "▾ Agents (31)") {
t.Fatalf("controls should return once a second panel exists")
}
}
+6 -3
View File
@@ -126,11 +126,14 @@ func (m Model) statusVisible() bool {
}
func (m Model) layout() (showSidebar bool, sidebarWidth, chatWidth, chatHeight int) {
showSidebar = m.width >= 120
if showSidebar {
showSidebar = m.width >= 120 && !m.sidebarHidden
switch {
case showSidebar:
sidebarWidth = max(24, m.width/5)
chatWidth = m.width - sidebarWidth - 1
} else {
case m.railVisible():
chatWidth = m.width - sidebarRailWidth - 1
default:
chatWidth = m.width
}
statusH := 0
+63 -102
View File
@@ -162,12 +162,8 @@ func (m Model) updateMouse(msg tea.MouseMsg) (tea.Model, tea.Cmd) {
return m.updateSetupMouse(msg)
}
showSidebar, _, chatWidth, chatHeight := m.layout()
viewerHeight := m.viewerHeight()
_, vulnHeight, mcpHeight, agentHeight := m.sidebarHeights()
x, y := msg.X, msg.Y
if m.updateMainScrollbarMouse(
msg, showSidebar, chatWidth, chatHeight, viewerHeight, agentHeight, vulnHeight, mcpHeight,
) {
if m.updateMainScrollbarMouse(msg, showSidebar, chatWidth, chatHeight) {
return m, nil
}
if m.selection.dragging {
@@ -198,24 +194,8 @@ func (m Model) updateMouse(msg tea.MouseMsg) (tea.Model, tea.Cmd) {
switch msg.Button {
case tea.MouseButtonWheelUp:
if showSidebar && x >= chatWidth+1 {
switch {
case y < viewerHeight:
return m, nil
case y < viewerHeight+agentHeight:
m.focus = focusAgents
m.input.Blur()
m.agentOffset = max(0, m.agentOffset-3)
m.keepAgentSelectionInWindow()
m.refreshViewport()
case vulnHeight > 0 && y < viewerHeight+agentHeight+vulnHeight:
m.focus = focusVulnerabilities
m.input.Blur()
m.vulnOffset = max(0, m.vulnOffset-3)
m.keepVulnerabilitySelectionInWindow()
case mcpHeight > 0 && y < viewerHeight+agentHeight+vulnHeight+mcpHeight:
m.focus = focusMcp
m.input.Blur()
m.mcpOffset = m.clampMcpOffset(m.mcpOffset - 3)
if rect, ok := m.panelAt(y); ok && rect.height > 1 {
m.scrollPanel(rect.panel, -3)
}
return m, nil
}
@@ -226,26 +206,8 @@ func (m Model) updateMouse(msg tea.MouseMsg) (tea.Model, tea.Cmd) {
return m, nil
case tea.MouseButtonWheelDown:
if showSidebar && x >= chatWidth+1 {
switch {
case y < viewerHeight:
return m, nil
case y < viewerHeight+agentHeight:
m.focus = focusAgents
m.input.Blur()
rows := m.agentPageSize()
m.agentOffset = min(max(0, len(agentTreeEntries(m.snapshot.Agents, m.collapsedAgents))-rows), m.agentOffset+3)
m.keepAgentSelectionInWindow()
m.refreshViewport()
case vulnHeight > 0 && y < viewerHeight+agentHeight+vulnHeight:
m.focus = focusVulnerabilities
m.input.Blur()
totalRows, _ := m.vulnerabilityScrollRows()
m.vulnOffset = min(max(0, totalRows-m.vulnerabilityPageSize()), m.vulnOffset+3)
m.keepVulnerabilitySelectionInWindow()
case mcpHeight > 0 && y < viewerHeight+agentHeight+vulnHeight+mcpHeight:
m.focus = focusMcp
m.input.Blur()
m.mcpOffset = m.clampMcpOffset(m.mcpOffset + 3)
if rect, ok := m.panelAt(y); ok && rect.height > 1 {
m.scrollPanel(rect.panel, 3)
}
return m, nil
}
@@ -288,21 +250,32 @@ func (m Model) updateMouse(msg tea.MouseMsg) (tea.Model, tea.Cmd) {
return m, nil
}
if !showSidebar || x < chatWidth+1 {
if x < chatWidth+1 {
return m, nil
}
// Sidebar: viewer, agents, vulnerabilities, then stats.
switch {
case y < viewerHeight:
if m.toggleButtonHit(x, y) {
m.selection.active = false
m.toggleSidebar()
return m, nil
}
if !showSidebar {
return m, nil
}
if y < m.viewerHeight() {
return m, send(m.client, "viewer.open", map[string]any{})
case y < viewerHeight+agentHeight:
}
rect, ok := m.panelAt(y)
if !ok || m.clickPanel(rect, x, y) {
return m, nil
}
localY := y - rect.top
switch rect.panel {
case panelAgents:
m.focus = focusAgents
m.input.Blur()
// Content starts after the top border (1) and vertical padding (1).
entries := agentTreeEntries(m.snapshot.Agents, m.collapsedAgents)
start := windowStart(m.agentOffset, len(entries), max(1, agentHeight-4))
localY := y - viewerHeight
if row := start + localY - 2; localY >= 2 && localY < agentHeight-2 && row < len(entries) {
start := windowStart(m.agentOffset, len(entries), m.agentPageSize())
if row := start + localY - 2; localY >= 2 && localY < rect.height-1 && row < len(entries) {
m.selectedAgent = entries[row].index
agentID := m.snapshot.Agents[m.selectedAgent].ID
if hasAgentChildren(agentID, m.snapshot.Agents) {
@@ -311,24 +284,22 @@ func (m Model) updateMouse(msg tea.MouseMsg) (tea.Model, tea.Cmd) {
}
m.refreshViewport()
}
case vulnHeight > 0 && y < viewerHeight+agentHeight+vulnHeight:
case panelFindings:
m.focus = focusVulnerabilities
m.input.Blur()
// Content starts after the top border (1); clicking a row opens its detail.
row := y - viewerHeight - agentHeight - 1
if idx := m.vulnerabilityIndexAtRow(row); row >= 0 && row < vulnHeight-2 && idx >= 0 {
row := localY - 2
if idx := m.vulnerabilityIndexAtRow(row); row >= 0 && row < rect.height-3 && idx >= 0 {
m.selectedVuln = idx
m.openModal(modalVulnerability)
}
case panelMcp:
m.focus = focusMcp
m.input.Blur()
}
return m, nil
}
func (m *Model) updateMainScrollbarMouse(
msg tea.MouseMsg,
showSidebar bool,
chatWidth, chatHeight, viewerHeight, agentHeight, vulnHeight, mcpHeight int,
) bool {
func (m *Model) updateMainScrollbarMouse(msg tea.MouseMsg, showSidebar bool, chatWidth, chatHeight int) bool {
if msg.Action == tea.MouseActionRelease {
if m.draggingScrollbar == scrollbarNone {
return false
@@ -337,18 +308,18 @@ func (m *Model) updateMainScrollbarMouse(
return true
}
if msg.Action == tea.MouseActionMotion && m.draggingScrollbar != scrollbarNone {
m.scrollFromMouse(m.draggingScrollbar, msg.Y, chatHeight, viewerHeight, agentHeight, vulnHeight)
m.scrollFromMouse(m.draggingScrollbar, msg.Y, chatHeight)
return true
}
if msg.Action != tea.MouseActionPress || msg.Button != tea.MouseButtonLeft {
return false
}
target := m.scrollbarAt(msg, showSidebar, chatWidth, chatHeight, viewerHeight, agentHeight, vulnHeight, mcpHeight)
target := m.scrollbarAt(msg, showSidebar, chatWidth, chatHeight)
if target == scrollbarNone {
return false
}
m.draggingScrollbar = target
m.scrollFromMouse(target, msg.Y, chatHeight, viewerHeight, agentHeight, vulnHeight)
m.scrollFromMouse(target, msg.Y, chatHeight)
return true
}
@@ -361,32 +332,28 @@ func nearColumn(x, column int) bool {
return x >= column-scrollbarGrab && x <= column+scrollbarGrab
}
// scrollbarAt reports which scrollbar, if any, the pointer is over.
func (m Model) scrollbarAt(
msg tea.MouseMsg,
showSidebar bool,
chatWidth, chatHeight, viewerHeight, agentHeight, vulnHeight, mcpHeight int,
) scrollbarTarget {
mcpTop := viewerHeight + agentHeight + vulnHeight
switch {
case nearColumn(msg.X, chatWidth-2) && msg.Y >= 1 && msg.Y < chatHeight-1 &&
m.viewport.TotalLineCount() > m.viewport.VisibleLineCount():
func (m Model) scrollbarAt(msg tea.MouseMsg, showSidebar bool, chatWidth, chatHeight int) scrollbarTarget {
if nearColumn(msg.X, chatWidth-2) && msg.Y >= 1 && msg.Y < chatHeight-1 &&
m.viewport.TotalLineCount() > m.viewport.VisibleLineCount() {
return scrollbarTrace
case showSidebar && nearColumn(msg.X, m.width-3) && msg.Y >= viewerHeight+2 &&
msg.Y < viewerHeight+agentHeight-2 &&
len(agentTreeEntries(m.snapshot.Agents, m.collapsedAgents)) > m.agentPageSize():
return scrollbarAgents
case showSidebar && vulnHeight > 0 && nearColumn(msg.X, m.width-3) &&
msg.Y >= viewerHeight+agentHeight+1 &&
msg.Y < viewerHeight+agentHeight+vulnHeight-1:
totalRows, _ := m.vulnerabilityScrollRows()
if totalRows > m.vulnerabilityPageSize() {
}
if !showSidebar || !nearColumn(msg.X, m.width-3) {
return scrollbarNone
}
rect, ok := m.panelAt(msg.Y)
if !ok || rect.height <= 1 || msg.Y < rect.top+2 || msg.Y >= rect.top+rect.height-1 {
return scrollbarNone
}
switch rect.panel {
case panelAgents:
if len(agentTreeEntries(m.snapshot.Agents, m.collapsedAgents)) > m.agentPageSize() {
return scrollbarAgents
}
case panelFindings:
if totalRows, _ := m.vulnerabilityScrollRows(); totalRows > m.vulnerabilityPageSize() {
return scrollbarFindings
}
// The roster scrolls below a fixed header, so its bar starts two rows into
// the panel (border then header) rather than one.
case showSidebar && mcpHeight > 0 && nearColumn(msg.X, m.width-3) &&
msg.Y >= mcpTop+2 && msg.Y < mcpTop+mcpHeight-1:
case panelMcp:
if len(m.snapshot.Connections) > m.mcpPageSize() {
return scrollbarMcp
}
@@ -394,10 +361,7 @@ func (m Model) scrollbarAt(
return scrollbarNone
}
func (m *Model) scrollFromMouse(
target scrollbarTarget,
y, chatHeight, viewerHeight, agentHeight, vulnHeight int,
) {
func (m *Model) scrollFromMouse(target scrollbarTarget, y, chatHeight int) {
switch target {
case scrollbarTrace:
height := max(1, chatHeight-2)
@@ -411,7 +375,7 @@ func (m *Model) scrollFromMouse(
total := len(agentTreeEntries(m.snapshot.Agents, m.collapsedAgents))
m.focus = focusAgents
m.input.Blur()
m.agentOffset = scrollbarOffset(y-viewerHeight-2, height, total, height)
m.agentOffset = scrollbarOffset(y-m.panelTop(panelAgents)-2, height, total, height)
m.keepAgentSelectionInWindow()
m.refreshViewport()
case scrollbarFindings:
@@ -420,15 +384,14 @@ func (m *Model) scrollFromMouse(
m.focus = focusVulnerabilities
m.input.Blur()
// The offset is a row, so dragging moves the list continuously.
m.vulnOffset = scrollbarOffset(y-viewerHeight-agentHeight-1, height, totalRows, height)
m.vulnOffset = scrollbarOffset(y-m.panelTop(panelFindings)-2, height, totalRows, height)
m.keepVulnerabilitySelectionInWindow()
case scrollbarMcp:
height := m.mcpPageSize()
total := len(m.snapshot.Connections)
m.focus = focusMcp
m.input.Blur()
// The bar starts two rows into the panel (border then the fixed header).
m.mcpOffset = scrollbarOffset(y-viewerHeight-agentHeight-vulnHeight-2, height, total, height)
m.mcpOffset = scrollbarOffset(y-m.panelTop(panelMcp)-2, height, total, height)
}
}
@@ -579,13 +542,11 @@ func labelHitAt(panel, label string, left, top, x, y int) bool {
func (m *Model) cycleFocus(delta int) {
available := []focusMode{focusInput, focusChat}
if m.width >= 120 {
available = append(available, focusAgents)
if len(m.snapshot.Vulnerabilities) > 0 {
available = append(available, focusVulnerabilities)
}
if len(m.snapshot.Connections) > 0 {
available = append(available, focusMcp)
if showSidebar, _, _, _ := m.layout(); showSidebar {
for _, rect := range m.sidebarPanels() {
if focus, ok := panelFocus(rect.panel); ok && rect.height > 1 {
available = append(available, focus)
}
}
}
idx := 0
+121 -72
View File
@@ -177,6 +177,18 @@ func wrapBlock(value string, width int) string {
return strings.Join(out, "\n")
}
// hyperlinkBlock wraps a URL to the column width and marks every wrapped line
// as an OSC 8 hyperlink to the whole URL. Terminals that linkify by text only
// see the first line of a wrapped URL, which for the viewer meant opening it
// with a truncated token.
func hyperlinkBlock(url string, width int, style lipgloss.Style) string {
lines := strings.Split(wrapBlock(url, width), "\n")
for i, line := range lines {
lines[i] = ansi.SetHyperlink(url) + style.Render(line) + ansi.ResetHyperlink()
}
return strings.Join(lines, "\n")
}
// scrollbarThumb brightens the bar being dragged so the grab reads as taking
// hold of it.
func (m Model) scrollbarThumb(target scrollbarTarget) lipgloss.Color {
@@ -375,10 +387,11 @@ const (
// fillBackground paints the whole frame black like Textual's Screen background.
// Bubble Tea has no screen compositor, so any cell the view does not explicitly
// color shows the terminal's default background. lipgloss emits a full reset
// (\x1b[0m) at the end of every styled span, which clears both foreground and
// background. Reasserting only black made uncolored and faint text inherit the
// terminal profile's foreground; light profiles therefore rendered that text
// color shows the terminal's default background. lipgloss emits a reset
// (\x1b[0m, or the bare \x1b[m that x/ansi uses) at the end of every styled
// span, which clears both foreground and background. Reasserting only black
// made uncolored and faint text inherit the terminal profile's foreground;
// light profiles therefore rendered that text
// black-on-black. Reapply both base colors after each reset (and at the start).
// Spans that set their own colors — inline code, selected rows, buttons — keep
// them, because their color is emitted after the base style.
@@ -386,9 +399,14 @@ func fillBackground(view string) string {
if view == "" {
return view
}
return baseFrameColors + strings.ReplaceAll(view, "\x1b[0m", "\x1b[0m"+baseFrameColors)
return baseFrameColors + baseColorRestorer.Replace(view)
}
var baseColorRestorer = strings.NewReplacer(
"\x1b[0m", "\x1b[0m"+baseFrameColors,
"\x1b[m", "\x1b[m"+baseFrameColors,
)
func (m Model) splashView() string {
shine := "Starting Strix Agent"
chars := []rune(shine)
@@ -504,6 +522,8 @@ func (m Model) mainView() string {
body := leftColumn
if showSidebar {
body = lipgloss.JoinHorizontal(lipgloss.Top, leftColumn, " ", m.sidebarView(sidebarWidth, m.height))
} else if m.railVisible() {
body = lipgloss.JoinHorizontal(lipgloss.Top, leftColumn, " ", m.sidebarRail(m.height))
}
return lipgloss.NewStyle().Background(black).Foreground(textColor).Render(body)
}
@@ -514,75 +534,108 @@ func (m Model) mainView() string {
// and so never applied - honoring it made the outline vanish on the one panel
// that had just become active.
func (m Model) sidebarView(width, height int) string {
// Stats box height fits its content (auto, max 15); vulns panel max-height 12.
statsBody := m.statsView()
statsHeight, vulnHeight, mcpHeight, agentHeight := m.sidebarHeights()
agentBorder := dark
if m.focus == focusAgents {
agentBorder = green
}
// #agents_tree padding: 1 (all sides); interior lines = box - border - v.padding.
agentRows := max(1, agentHeight-4)
agentEntries := agentTreeEntries(m.snapshot.Agents, m.collapsedAgents)
agents := withVerticalScrollbar(
m.agentsView(max(1, width-5), agentRows),
width-4,
agentRows,
len(agentEntries),
agentRows,
m.agentOffset,
m.scrollbarThumb(scrollbarAgents),
)
parts := []string{
lipgloss.NewStyle().Width(width-2).Height(m.viewerHeight()-2).Border(lipgloss.RoundedBorder()).BorderForeground(dark).Padding(0, 1).Render(m.viewerView(width - 4)),
lipgloss.NewStyle().Width(width-2).Height(agentHeight-2).Border(lipgloss.RoundedBorder()).BorderForeground(agentBorder).Padding(1, 1).Render(agents),
}
if vulnHeight > 0 {
vulnBorder := dark
if m.focus == focusVulnerabilities {
vulnBorder = green
}
vulnRows := max(1, vulnHeight-2)
totalRows, offsetRows := m.vulnerabilityScrollRows()
findings := withVerticalScrollbar(
m.vulnerabilitiesView(m.vulnerabilityListWidth(), vulnRows),
parts := []string{lipgloss.NewStyle().
Width(width-2).
Height(m.viewerHeight()-2).
Border(lipgloss.RoundedBorder()).
BorderForeground(dark).
Padding(0, 1).
Render(m.viewerBox(width - 4))}
agents := ""
if agentHeight > 1 {
agentRows := max(1, agentHeight-4)
agents = withVerticalScrollbar(
m.agentsView(max(1, width-5), agentRows),
width-4,
vulnRows,
totalRows,
vulnRows,
offsetRows,
m.scrollbarThumb(scrollbarFindings),
agentRows,
len(agentTreeEntries(m.snapshot.Agents, m.collapsedAgents)),
agentRows,
m.agentOffset,
m.scrollbarThumb(scrollbarAgents),
)
parts = append(parts, lipgloss.NewStyle().Width(width-2).Height(vulnRows).Border(lipgloss.RoundedBorder()).BorderForeground(vulnBorder).Padding(0, 1).Render(findings))
}
parts = append(parts, m.panelBox(panelAgents, agents, width, agentHeight, m.focus == focusAgents))
if vulnHeight > 0 {
findings := ""
if vulnHeight > 1 {
vulnRows := max(1, vulnHeight-4)
totalRows, offsetRows := m.vulnerabilityScrollRows()
findings = withVerticalScrollbar(
m.vulnerabilitiesView(m.vulnerabilityListWidth(), vulnRows),
width-4,
vulnRows,
totalRows,
vulnRows,
offsetRows,
m.scrollbarThumb(scrollbarFindings),
)
}
parts = append(parts, m.panelBox(panelFindings, findings, width, vulnHeight, m.focus == focusVulnerabilities))
}
if mcpHeight > 0 {
mcpBorder := dark
if m.focus == focusMcp {
mcpBorder = green
roster := ""
if mcpHeight > 1 {
roster = m.mcpConnectionsView(width-4, max(1, mcpHeight-4))
}
mcpRows := max(1, mcpHeight-2)
parts = append(parts, lipgloss.NewStyle().Width(width-2).Height(mcpRows).Border(lipgloss.RoundedBorder()).BorderForeground(mcpBorder).Padding(0, 1).Render(m.mcpConnectionsView(width-4, mcpRows)))
parts = append(parts, m.panelBox(panelMcp, roster, width, mcpHeight, m.focus == focusMcp))
}
parts = append(parts, lipgloss.NewStyle().Width(width-2).Height(statsHeight-2).Border(lipgloss.RoundedBorder()).BorderForeground(dark).Padding(0, 1).Render(statsBody))
return strings.Join(parts, "\n")
if gap := m.sidebarGap(); gap > 0 {
parts = append(parts, lipgloss.NewStyle().Width(width).Height(gap).Render(""))
}
stats := ""
if statsHeight > 1 {
stats = fixedPanelBody(lipgloss.NewStyle().Width(width-4).Render(m.statsView()), width-4, statsHeight-2)
}
parts = append(parts, m.panelBox(panelStats, stats, width, statsHeight, false))
return lipgloss.JoinVertical(lipgloss.Left, parts...)
}
func (m Model) sidebarHeights() (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.
statsRows := lipgloss.Height(lipgloss.NewStyle().Width(m.viewerContentWidth()).Render(m.statsView()))
statsHeight = min(15, statsRows+2)
statsRows := lipgloss.Height(lipgloss.NewStyle().Width(m.sidebarInnerWidth()).Render(m.statsView()))
statsHeight = m.panelHeight(panelStats, min(15, statsRows+2))
if len(m.snapshot.Vulnerabilities) > 0 {
vulnHeight = min(12, len(m.vulnerabilityRows(m.vulnerabilityListWidth()))+2)
vulnHeight = m.panelHeight(panelFindings, min(13, len(m.vulnerabilityRows(m.vulnerabilityListWidth()))+4))
}
// One header line + one line per connection + the box border (2). Capped so a
// Header line + one line per connection + the box border (2). Capped so a
// long roster cannot crowd out the agent tree; a roster past the cap scrolls
// inside the panel. Absent entirely when the run has no MCP connections.
if len(m.snapshot.Connections) > 0 {
mcpHeight = min(9, len(m.snapshot.Connections)+3)
mcpHeight = m.panelHeight(panelMcp, min(10, len(m.snapshot.Connections)+4))
}
agentHeight = m.panelHeight(panelAgents, 5)
for _, p := range []struct {
h *int
floor int
}{{&statsHeight, 4}, {&mcpHeight, 5}, {&vulnHeight, 5}} {
over := m.viewerHeight() + statsHeight + vulnHeight + mcpHeight + agentHeight - m.height
if over <= 0 {
break
}
if *p.h > p.floor {
*p.h -= min(over, *p.h-p.floor)
}
}
for _, h := range []*int{&mcpHeight, &vulnHeight, &statsHeight} {
if m.viewerHeight()+statsHeight+vulnHeight+mcpHeight+agentHeight <= m.height {
break
}
if *h > 1 {
*h = 1
}
}
spare := max(0, m.height-m.viewerHeight()-statsHeight-vulnHeight-mcpHeight-agentHeight)
switch {
case m.zoomedPanel == panelFindings && vulnHeight > 1:
vulnHeight += spare
case m.zoomedPanel == panelMcp && mcpHeight > 1:
mcpHeight += spare
case agentHeight > 1:
agentHeight += spare
}
agentHeight = max(3, m.height-m.viewerHeight()-statsHeight-vulnHeight-mcpHeight)
return
}
@@ -590,7 +643,7 @@ func (m Model) viewerHeight() int {
return strings.Count(m.viewerView(m.viewerContentWidth()), "\n") + 3
}
func (m Model) viewerContentWidth() int {
func (m Model) sidebarInnerWidth() int {
_, sidebarWidth, _, _ := m.layout()
if sidebarWidth == 0 {
sidebarWidth = 24
@@ -598,13 +651,16 @@ func (m Model) viewerContentWidth() int {
return max(1, sidebarWidth-4)
}
func (m Model) viewerContentWidth() int {
return max(1, m.sidebarInnerWidth()-toggleButtonWidth-1)
}
func (m Model) viewerView(width int) string {
switch m.snapshot.ViewerStatus {
case "running":
status := lipgloss.NewStyle().Foreground(green).Render("● Viewer running")
if m.snapshot.ViewerURL != nil && strings.TrimSpace(*m.snapshot.ViewerURL) != "" {
url := wrapBlock(strings.TrimSpace(*m.snapshot.ViewerURL), width)
return status + "\n" + lipgloss.NewStyle().Foreground(dim).Render(url)
return status + "\n" + hyperlinkBlock(strings.TrimSpace(*m.snapshot.ViewerURL), width, lipgloss.NewStyle().Foreground(dim))
}
return status
case "unavailable":
@@ -651,29 +707,22 @@ func (m Model) statsView() string {
return b.String()
}
// mcpConnectionsView renders the sidebar MCP panel: a header carrying the total
// connection count, then one row per connection with a status glyph and its tool
// count (or "offline").
// mcpConnectionsView renders the sidebar MCP roster: one row per connection with
// a status glyph and its tool count (or "offline").
// - a solid green dot marks an attached, idle connection;
// - a green cycling quarter-circle (◐ ◓ ◑ ◒) marks a call running against it;
// - a red dot plus "offline" marks a connection whose live session has died.
//
// The header stays fixed while the roster below it scrolls: when there are more
// connections than the panel can show, the visible window is chosen by
// m.mcpOffset and withVerticalScrollbar draws a thumb in the reserved last
// column, exactly as the agent tree and findings list scroll.
// When there are more connections than the panel can show, the visible window
// is chosen by m.mcpOffset and withVerticalScrollbar draws a thumb in the
// reserved last column, exactly as the agent tree and findings list scroll.
//
// "In use" is derived from the connection-tagged tool-call events in the stream,
// not carried on the connection roster, so a call in flight shows motion without
// any extra backend signal. The quarter-circle rides the shared sweepFrame tick.
func (m Model) mcpConnectionsView(width, rows int) string {
conns := m.snapshot.Connections
header := truncate(lipgloss.NewStyle().Foreground(dim).Render(
fmt.Sprintf("MCP Connections (%d)", len(conns))), width)
bodyRows := max(0, rows-1)
if bodyRows == 0 {
return header
}
bodyRows := max(1, rows)
inUse := m.mcpInUse()
frames := []rune{'◐', '◓', '◑', '◒'}
// Reserve the scrollbar column whether or not the bar is showing, so the
@@ -710,7 +759,7 @@ func (m Model) mcpConnectionsView(width, rows int) string {
m.mcpOffset,
m.scrollbarThumb(scrollbarMcp),
)
return header + "\n" + roster
return roster
}
// mcpPageSize is how many connection rows the roster shows at once, below its
@@ -135,7 +135,7 @@ func clampVulnerabilityOffset(offset, total, height int) int {
func (m Model) vulnerabilityPageSize() int {
_, vulnHeight, _, _ := m.sidebarHeights()
return max(1, vulnHeight-2)
return max(1, vulnHeight-4)
}
// vulnerabilityPageItems is how many findings a page step should move by: the
@@ -192,7 +192,7 @@ func (m Model) modalView() string {
switch m.modal {
case modalHelp:
title := lipgloss.NewStyle().Bold(true).Foreground(green).Width(34).Align(lipgloss.Center).Render("Strix Help")
body := lipgloss.NewStyle().Foreground(textColor).Render("F1 Help\nCtrl+O Open viewer\nCtrl+Q/C Quit\nESC Stop Agent\nEnter Send / expand node\nCtrl+J Newline in message\nTab Switch panels\n↑/↓ Navigate tree\nDrag Select & copy text\nClick Expand/collapse tool")
body := lipgloss.NewStyle().Foreground(textColor).Render("F1 Help\nCtrl+O Open viewer\nCtrl+Q/C Quit\nESC Stop Agent\nEnter Send / expand node\nCtrl+J Newline in message\nTab Switch panels\n↑/↓ Navigate tree\nDrag Select & copy text\nClick Expand/collapse tool\nClick ▾/⤢ Collapse / zoom panel\nClick »/« Hide / show sidebar")
content := title + "\n\n" + body
return lipgloss.NewStyle().Width(38).Border(lipgloss.RoundedBorder()).BorderForeground(green).Background(black).Padding(1, 2).Render(content)
case modalQuit:
@@ -11,8 +11,6 @@ import (
// Markdown (agent_message_renderer.py)
// ---------------------------------------------------------------------------
var blankLineRuns = regexp.MustCompile(`\n\s*\n`)
type mdHeader struct {
prefix string
strip int
@@ -33,13 +31,36 @@ func renderAssistantMarkdown(content string) string {
if content == "" {
return ""
}
cleaned := strings.TrimSpace(blankLineRuns.ReplaceAllString(content, "\n\n"))
cleaned := strings.TrimSpace(collapseBlankLines(content))
if cleaned == "" {
return ""
}
return applyMarkdownStyles(cleaned)
}
func collapseBlankLines(content string) string {
lines := strings.Split(content, "\n")
out := make([]string, 0, len(lines))
inCode := false
blank := false
for _, line := range lines {
if strings.HasPrefix(line, "```") {
inCode = !inCode
}
if !inCode && strings.TrimSpace(line) == "" {
if blank {
continue
}
blank = true
out = append(out, "")
continue
}
blank = false
out = append(out, line)
}
return strings.Join(out, "\n")
}
func applyMarkdownStyles(text string) string {
var out strings.Builder
lines := strings.Split(text, "\n")
@@ -51,7 +51,7 @@ func renderDependencyReport(args map[string]any, result any) string {
field("Target", StringValue(args["target"]))
section := func(label, value string) {
if value != "" {
b.WriteString("\n\n" + Bold(Field).Render(label) + "\n" + value)
b.WriteString("\n\n" + Bold(Field).Render(label) + "\n" + renderAssistantMarkdown(value))
}
}
section("Description", StringValue(args["description"]))
@@ -321,3 +321,36 @@ func TestCollapseToolOnlyOutputHeavyTools(t *testing.T) {
t.Fatal("respond_to_user must never collapse")
}
}
func TestReportSectionsRenderMarkdown(t *testing.T) {
body := "**Systemic themes:** zero auth on `/share`\n\n# Recommendations\n1. Merge the **ejs 3.1.10** fix"
cases := map[string]string{
"finish_scan": renderFinishScan(map[string]any{"executive_summary": body}),
"create_vulnerability_report": renderVulnerabilityReport(map[string]any{"title": "x", "description": body}, nil),
"create_dependency_report": renderDependencyReport(map[string]any{"title": "x", "description": body}, nil),
}
for name, out := range cases {
plain := ansi.Strip(out)
for _, raw := range []string{"**", "`", "# Recommendations"} {
if strings.Contains(plain, raw) {
t.Fatalf("%s left raw markdown %q in %q", name, raw, plain)
}
}
for _, want := range []string{"Systemic themes:", "/share", "Recommendations", "1. Merge the ejs 3.1.10 fix"} {
if !strings.Contains(plain, want) {
t.Fatalf("%s lost %q in %q", name, want, plain)
}
}
}
}
func TestReportSectionsKeepBlankLinesInFencedCode(t *testing.T) {
body := "Intro.\n\n\n\nRequest:\n```http\nGET /share HTTP/1.1\nHost: target\n\n\nbody\n```\n\n\n\nAfter."
plain := ansi.Strip(renderVulnerabilityReport(map[string]any{"title": "x", "technical_analysis": body}, nil))
if !strings.Contains(plain, "Host: target\n\n\nbody") {
t.Fatalf("fenced code lost its blank lines in %q", plain)
}
if strings.Contains(plain, "Intro.\n\n\n") || strings.Contains(plain, "\n\n\nAfter.") {
t.Fatalf("blank-line runs outside fenced code were not collapsed in %q", plain)
}
}
@@ -89,7 +89,7 @@ func renderReport(args map[string]any, result any, heading, pending string) stri
section := func(label, value string) {
if value != "" {
b.WriteString("\n\n" + Bold(Field).Render(label) + "\n" + value)
b.WriteString("\n\n" + Bold(Field).Render(label) + "\n" + renderAssistantMarkdown(value))
}
}
if confidence := StringValue(args["confidence"]); confidence != "" {
+1 -1
View File
@@ -13,7 +13,7 @@ func renderFinishScan(args map[string]any) string {
b.WriteString(Col(Green).Render("◆ ") + Bold(Green).Render("Penetration test completed"))
section := func(label, value string) {
if value != "" {
b.WriteString("\n\n" + Bold(Field).Render(label) + "\n" + value)
b.WriteString("\n\n" + Bold(Field).Render(label) + "\n" + renderAssistantMarkdown(value))
}
}
es := StringValue(args["executive_summary"])
+13 -35
View File
@@ -11,7 +11,7 @@ import shutil
import sys
from copy import deepcopy
from pathlib import Path
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, TextIO
from strix.config import load_settings, persist_current
from strix.core.agents import AgentCoordinator
@@ -58,6 +58,10 @@ class GoTuiPreActivationError(RuntimeError):
"""A sidecar failure raised before the Go TUI activates."""
def _open_output_sink() -> TextIO:
return Path(os.devnull).open("a", buffering=1, encoding="utf-8")
class GoTuiRuntime:
def __init__(self, args: argparse.Namespace) -> None:
self.args = args
@@ -123,8 +127,8 @@ class GoTuiRuntime:
async def check_setup_model(self) -> None:
"""Verify the model route as soon as the start screen is up.
The same round trip a direct launch makes in prepare_and_start, run in
the background so the screen paints first and the outcome lands in the
The same round trip main() makes before a direct launch, run in the
background so the screen paints first and the outcome lands in the
setup log before the user has finished typing.
"""
if not (load_settings().llm.model or "").strip():
@@ -158,13 +162,13 @@ class GoTuiRuntime:
await preflight_model_connection(model)
self.model_verified = True
def _start_preparation(self) -> asyncio.Task[None]:
def _start_preparation(self) -> asyncio.Task[None] | None:
"""Kick off the work that runs behind the freshly painted TUI."""
if self.controller.setup_mode:
self._setup_preflight = asyncio.create_task(self.check_setup_model())
return self._setup_preflight
self.controller.begin_preparation()
return asyncio.create_task(self.prepare_and_start())
self.start_scan()
return None
async def start_from_setup(self) -> None:
candidate = deepcopy(self.args)
@@ -204,34 +208,6 @@ class GoTuiRuntime:
self.init_run_state()
self.start_scan()
async def prepare_and_start(self) -> None:
"""Prepare a directly-launched scan once the TUI is on screen.
The model round trip and run preparation run here rather than before
launch so the interface appears immediately.
"""
model = (load_settings().llm.model or "").strip()
set_scan_phase("preflight")
try:
await preflight_model_connection(model)
except Exception as exc:
logger.exception("Go TUI scan preparation failed")
report_error("model_connection_failed", exc)
self.controller.fail_preparation(str(exc))
return
try:
persist_current()
prepare_run(self.args)
telemetry_start(self.args)
except Exception as exc:
logger.exception("Go TUI scan preparation failed")
report_error("scan_preparation_failed", exc)
self.controller.fail_preparation(str(exc))
return
self.controller.scan_state = "running"
self.init_run_state()
self.start_scan()
def start_scan(self) -> None:
if self.scan_task is None:
self.scan_task = asyncio.create_task(self._run_scan())
@@ -433,7 +409,7 @@ class GoTuiRuntime:
# only the Python-level bindings change.
original_stdout = sys.stdout
original_stderr = sys.stderr
output_sink = Path(os.devnull).open("a", buffering=1) # noqa: SIM115
output_sink = _open_output_sink()
sys.stdout = output_sink
sys.stderr = output_sink
backend_socket: socket.socket | None = None
@@ -441,6 +417,8 @@ class GoTuiRuntime:
prepare_task: asyncio.Task[None] | None = None
process: asyncio.subprocess.Process | subprocess.Popen[bytes] | None = None
try:
if not self.controller.setup_mode:
self.init_run_state()
env = child_environment()
env["STRIX_VERSION"] = package_version()
command = self.binary_command()
+86
View File
@@ -0,0 +1,86 @@
"""Summaries of the runs under ./strix_runs, for pickers and listings."""
from __future__ import annotations
import json
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from strix.core.paths import run_record_path, runs_base_dir, runtime_state_dir
if TYPE_CHECKING:
from pathlib import Path
@dataclass(frozen=True)
class RunSummary:
run_name: str
target: str
started_at: str
ended_at: str
status: str
findings: int
resumable: bool
def list_run_summaries(*, cwd: Path | None = None) -> list[RunSummary]:
"""Every run with a run.json, newest activity first (run.json mtime)."""
base = runs_base_dir(cwd=cwd)
if not base.is_dir():
return []
rows: list[tuple[float, RunSummary]] = []
for child in base.iterdir():
record_path = run_record_path(child)
try:
if not record_path.is_file():
continue
modified = record_path.stat().st_mtime
except OSError:
continue
rows.append((modified, _summarize(child, _load_json(record_path, default={}))))
rows.sort(key=lambda row: (row[0], row[1].run_name), reverse=True)
return [summary for _, summary in rows]
def _summarize(run_dir: Path, record: Any) -> RunSummary:
if not isinstance(record, dict):
record = {}
findings = _load_json(run_dir / "vulnerabilities.json", default=[])
return RunSummary(
run_name=run_dir.name,
target=_describe_target(record),
started_at=str(record.get("start_time") or ""),
ended_at=str(record.get("end_time") or ""),
status=str(record.get("status") or "unknown"),
findings=len(findings) if isinstance(findings, list) else 0,
resumable=(runtime_state_dir(run_dir) / "agents.json").is_file(),
)
def _describe_target(record: dict[str, Any]) -> str:
targets = record.get("targets_info")
originals = [
str(entry["original"])
for entry in targets or []
if isinstance(entry, dict) and entry.get("original")
]
if originals:
if len(originals) == 1:
return originals[0]
return f"{originals[0]} +{len(originals) - 1} more"
mount = record.get("workspace_mount")
if isinstance(mount, str) and mount:
return f"{mount} (workspace)"
instruction = str(record.get("user_instruction") or record.get("instruction") or "").strip()
return instruction.splitlines()[0] if instruction else ""
def _load_json(path: Path, *, default: Any) -> Any:
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return default
__all__ = ["RunSummary", "list_run_summaries"]
+60 -43
View File
@@ -59,6 +59,7 @@ _NOISY_LIBS: tuple[str, ...] = (
_HANDLER_TAG = "_strix_scan_handler"
_STREAM_TAG = "_strix_stream_handler"
# ``openai.agents`` is the openai-agents SDK's canonical logger root.
@@ -87,35 +88,64 @@ def configure_dependency_logging() -> None:
logging.getLogger("asyncio").setLevel(logging.CRITICAL)
logging.getLogger("asyncio").propagate = False
warnings.filterwarnings("ignore", category=RuntimeWarning, module="asyncio")
_silence_urllib3_finalizer_noise()
_route_unraisable_to_log()
_unraisable_hook_installed = False
def _is_urllib3_closed_file_noise(unraisable: sys.UnraisableHookArgs) -> bool:
return (
isinstance(unraisable.exc_value, ValueError)
and "I/O operation on closed file" in str(unraisable.exc_value)
and type(unraisable.object).__module__.split(".")[0] == "urllib3"
)
def _silence_urllib3_finalizer_noise() -> None:
global _unraisable_hook_installed # noqa: PLW0603
if _unraisable_hook_installed:
return
_unraisable_hook_installed = True
previous = sys.unraisablehook
def hook(unraisable: sys.UnraisableHookArgs) -> None:
if _is_urllib3_closed_file_noise(unraisable):
return
previous(unraisable)
def _route_unraisable_to_log() -> None:
def hook(u: sys.UnraisableHookArgs) -> None:
with contextlib.suppress(BaseException):
logger = logging.getLogger("strix.telemetry")
if logger.hasHandlers():
logger.warning(
u.err_msg or f"Exception ignored in {u.object!r}",
exc_info=(u.exc_type, u.exc_value, u.exc_traceback), # type: ignore[arg-type]
)
sys.unraisablehook = hook
class _CurrentStderrHandler(logging.StreamHandler): # type: ignore[type-arg]
def __init__(self, level: int) -> None:
logging.Handler.__init__(self, level)
@property
def stream(self) -> object:
return sys.stderr
@stream.setter
def stream(self, _value: object) -> None:
pass
def _debug_enabled(debug: bool | None) -> bool:
if debug is not None:
return debug
return (os.environ.get("STRIX_DEBUG") or "").strip().lower() in {"1", "true", "yes", "on"}
def _stream_handler(debug: bool) -> logging.Handler:
handler = _CurrentStderrHandler(logging.DEBUG if debug else logging.ERROR)
handler.setFormatter(logging.Formatter(_FORMAT, datefmt=_DATEFMT))
handler.addFilter(_StrixContextFilter())
handler.addFilter(_StdoutQuietFilter())
setattr(handler, _STREAM_TAG, True)
return handler
def _has_stream_handler(tracked: logging.Logger) -> bool:
return any(getattr(handler, _STREAM_TAG, False) for handler in tracked.handlers)
def setup_console_logging(*, debug: bool | None = None) -> None:
configure_dependency_logging()
for name in _TRACKED_ROOTS:
tracked = logging.getLogger(name)
tracked.setLevel(logging.DEBUG)
tracked.propagate = False
if not _has_stream_handler(tracked):
tracked.addHandler(_stream_handler(_debug_enabled(debug)))
def setup_scan_logging(run_dir: Path, *, debug: bool | None = None) -> Callable[[], None]:
"""Attach scan-scoped handlers; return a teardown callable.
@@ -134,38 +164,25 @@ def setup_scan_logging(run_dir: Path, *, debug: bool | None = None) -> Callable[
"""
configure_dependency_logging()
if debug is None:
debug = (os.environ.get("STRIX_DEBUG") or "").strip().lower() in {
"1",
"true",
"yes",
"on",
}
debug = _debug_enabled(debug)
run_dir.mkdir(parents=True, exist_ok=True)
log_path = run_dir / "strix.log"
formatter = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
context_filter = _StrixContextFilter()
file_handler = logging.FileHandler(log_path, mode="a", encoding="utf-8")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
file_handler.addFilter(context_filter)
file_handler.setFormatter(logging.Formatter(_FORMAT, datefmt=_DATEFMT))
file_handler.addFilter(_StrixContextFilter())
setattr(file_handler, _HANDLER_TAG, True)
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.DEBUG if debug else logging.ERROR)
stream_handler.setFormatter(formatter)
stream_handler.addFilter(context_filter)
stream_handler.addFilter(_StdoutQuietFilter())
setattr(stream_handler, _HANDLER_TAG, True)
tracked_loggers = [logging.getLogger(name) for name in _TRACKED_ROOTS]
for tracked in tracked_loggers:
tracked.setLevel(logging.DEBUG)
tracked.addHandler(file_handler)
tracked.addHandler(stream_handler)
if not _has_stream_handler(tracked):
stream_handler = _stream_handler(debug)
setattr(stream_handler, _HANDLER_TAG, True)
tracked.addHandler(stream_handler)
tracked.propagate = False
for name in _NOISY_LIBS:
+11 -14
View File
@@ -238,12 +238,17 @@ async def finish_scan(
remediation steps. End with retest/validation guidance.
- **Formatting — use markdown in every field.** These fields may be
rendered into generated reports, so structure them clearly: lead
each section with a short ``# Heading``, use ``**bold**`` for labels/emphasis,
``inline code`` for identifiers/paths/parameters, bullet or
numbered lists for enumerations, and fenced code blocks
(```` ```language ````) for any code/payload excerpts. Never emit
one flat wall of prose or leave code unformatted.
rendered into generated reports, so structure them clearly: use
``**bold**`` for labels/emphasis, ``inline code`` for
identifiers/paths/parameters, bullet or numbered lists for
enumerations, and fenced code blocks (```` ```language ````) for
any code/payload excerpts. Never emit one flat wall of prose or
leave code unformatted.
- **Each field is a section body, not a document.** The report adds
the section title ("Executive Summary", "Methodology", ...) itself,
so do NOT start a field with a heading such as ``# Executive
Summary`` — it would print twice. Sub-headings (``##``) inside a
field are fine.
- If **zero** vulnerabilities were found, say so plainly and
characterize the posture positively; ``technical_analysis`` should
summarize the areas tested and confirm no issues, and
@@ -252,8 +257,6 @@ async def finish_scan(
Example (abbreviated — mirror this structure, not the wording)::
executive_summary:
# Executive Summary
An external assessment of the **Acme Customer Portal**
identified multiple weaknesses that could lead to
unauthorized access to customer data.
@@ -269,8 +272,6 @@ async def finish_scan(
- Potential exposure of customer records across tenants.
methodology:
# Methodology
Conducted per the **OWASP WSTG**.
**Engagement type:** Gray-box external test.
@@ -280,8 +281,6 @@ async def finish_scan(
and tenant-isolation testing, input/SSRF testing.
technical_analysis:
# Technical Analysis
**Severity model** reflects exploitability x impact.
1. **SSRF in URL preview** (Critical) — insufficient
@@ -293,8 +292,6 @@ async def finish_scan(
no deny-by-default egress policy.
recommendations:
# Recommendations
**Immediate**
1. Remediate SSRF: enforce a destination allowlist,
deny-by-default, re-validate on every redirect hop.
+55 -55
View File
@@ -1075,11 +1075,11 @@ async def create_vulnerability_report(
"Techniques" that read like an engineering runbook rather than a
client deliverable.
- **Use markdown in every text field**: ``**bold**`` for emphasis,
``inline code`` for identifiers/values/parameters, and fenced
code blocks (```` ```language ````) for any code/payload/HTTP
``inline code`` for identifiers/values/parameters, and
language-tagged fenced code blocks for any code/payload/HTTP
excerpt. Never leave code bare/unformatted. When referencing a
file, annotate the fence, e.g.
```` ```python title=app.py startLineNumber=42 endLineNumber=50 ````.
file, annotate the opening fence with
``title=app.py startLineNumber=42 endLineNumber=50`` after the language.
- Field discipline: ``poc_description`` is steps only — NO code (all
code goes in ``poc_script_code``); ``remediation_steps`` is prose
only — NO code/diffs (code fixes go in ``code_locations``).
@@ -1191,6 +1191,57 @@ async def create_vulnerability_report(
Broken / Risky Crypto, CWE-311 Missing Encryption, CWE-916 Weak
Password Hashing.
Example (abbreviated — mirror this structure)::
title: "Reflected XSS in /search q parameter"
description:
The **`q`** parameter of `/search` reflects user input into
the HTML response without encoding, allowing script
injection.
technical_analysis:
The handler interpolates `q` directly into the page body:
```python title=views.py startLineNumber=42 endLineNumber=44
html = f"<h2>Results for {q}</h2>"
return HttpResponse(html)
```
No output encoding is applied, so `<script>` executes.
poc_description:
1. Navigate to `/search?q=<payload>`.
2. Observe the payload executes in the victim's browser.
poc_script_code:
```
GET /search?q=<script>alert(document.domain)</script>
```
evidence:
Response echoes the payload verbatim:
```html
<h2>Results for <script>alert(document.domain)</script></h2>
```
assumptions:
Assumes a victim can be induced to open a crafted link.
remediation_steps:
Context-encode all user input rendered into HTML; prefer the
template engine's auto-escaping over string interpolation.
counterevidence:
No output encoding, CSP, or WAF observed on this response;
payload executed in a current browser. The parameter is
reflected on an unauthenticated route, so no privileged
position is required.
confidence: "high"
severity_change_conditions:
A restrictive CSP that blocks inline script execution would
reduce impact and lower the severity.
fix_effort: "low"
Nice to have: for code findings, if the checkout has git history, a quick
``git blame`` (quote the paths) on the vulnerable line is worth weaving into
``technical_analysis`` — who last touched it, when, and in which commit, as
part of the prose, not a separate section. Skip it if the line is
uncommitted or the command fails.
Args:
title: Specific finding title (e.g.
``"SQL Injection in /api/users login parameter"``). Don't
@@ -1359,57 +1410,6 @@ async def create_vulnerability_report(
fix (summary + rationale). Prose/markdown only — the code
change itself belongs in ``code_locations``. Omit for
black-box findings.
Example (abbreviated — mirror this structure)::
title: "Reflected XSS in /search q parameter"
description:
The **`q`** parameter of `/search` reflects user input into
the HTML response without encoding, allowing script
injection.
technical_analysis:
The handler interpolates `q` directly into the page body:
```python title=views.py startLineNumber=42 endLineNumber=44
html = f"<h2>Results for {q}</h2>"
return HttpResponse(html)
```
No output encoding is applied, so `<script>` executes.
poc_description:
1. Navigate to `/search?q=<payload>`.
2. Observe the payload executes in the victim's browser.
poc_script_code:
```
GET /search?q=<script>alert(document.domain)</script>
```
evidence:
Response echoes the payload verbatim:
```html
<h2>Results for <script>alert(document.domain)</script></h2>
```
assumptions:
Assumes a victim can be induced to open a crafted link.
remediation_steps:
Context-encode all user input rendered into HTML; prefer the
template engine's auto-escaping over string interpolation.
counterevidence:
No output encoding, CSP, or WAF observed on this response;
payload executed in a current browser. The parameter is
reflected on an unauthenticated route, so no privileged
position is required.
confidence: "high"
severity_change_conditions:
A restrictive CSP that blocks inline script execution would
reduce impact and lower the severity.
fix_effort: "low"
Nice to have: for code findings, if the checkout has git history, a quick
``git blame`` (quote the paths) on the vulnerable line is worth weaving into
``technical_analysis`` — who last touched it, when, and in which commit, as
part of the prose, not a separate section. Skip it if the line is
uncommitted or the command fails.
"""
(
http_exchange_ids,
+112
View File
@@ -0,0 +1,112 @@
"""Tests for the --fail-on headless severity gate."""
from __future__ import annotations
import importlib
import sys
from types import SimpleNamespace
from typing import Any
import pytest
cli_main: Any = importlib.import_module("strix.interface.main")
def _stub_settings(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
cli_main,
"load_settings",
lambda: SimpleNamespace(runtime=SimpleNamespace(max_local_copy_mb=1024)),
)
def _reports(*severities: str | None) -> list[dict[str, Any]]:
return [{"id": f"vuln-{i:04d}", "severity": sev} for i, sev in enumerate(severities, 1)]
def test_no_findings_never_fail() -> None:
assert not cli_main.findings_fail_build([], None)
assert not cli_main.findings_fail_build([], "info")
@pytest.mark.parametrize("severity", ["critical", "high", "medium", "low", "info", "none"])
def test_without_threshold_any_finding_fails(severity: str) -> None:
assert cli_main.findings_fail_build(_reports(severity), None)
@pytest.mark.parametrize(
("fail_on", "severity", "expected"),
[
("high", "critical", True),
("high", "high", True),
("high", "medium", False),
("high", "low", False),
("high", "info", False),
("critical", "high", False),
("critical", "critical", True),
("info", "info", True),
("info", "low", True),
("medium", "HIGH", True),
("medium", " Low ", False),
],
)
def test_threshold_compares_severity(fail_on: str, severity: str, expected: bool) -> None:
assert cli_main.findings_fail_build(_reports(severity), fail_on) is expected
def test_one_finding_at_threshold_fails_a_mixed_run() -> None:
assert cli_main.findings_fail_build(_reports("info", "low", "high"), "high")
@pytest.mark.parametrize("severity", ["severe", "", None])
def test_unrecognized_severity_fails_closed(severity: str | None) -> None:
assert cli_main.findings_fail_build(_reports(severity), "critical")
def test_none_severity_passes_any_threshold() -> None:
assert not cli_main.findings_fail_build(_reports("none"), "info")
def test_parse_fail_on_is_case_insensitive(monkeypatch: pytest.MonkeyPatch) -> None:
_stub_settings(monkeypatch)
monkeypatch.setattr(
sys, "argv", ["strix", "-t", "https://test.com/", "-n", "--fail-on", "HIGH"]
)
args = cli_main.parse_arguments()
assert args.fail_on == "high"
def test_parse_fail_on_defaults_to_none(monkeypatch: pytest.MonkeyPatch) -> None:
_stub_settings(monkeypatch)
monkeypatch.setattr(sys, "argv", ["strix", "-t", "https://test.com/", "-n"])
assert cli_main.parse_arguments().fail_on is None
def test_parse_fail_on_rejects_unknown_severity(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
_stub_settings(monkeypatch)
monkeypatch.setattr(
sys, "argv", ["strix", "-t", "https://test.com/", "-n", "--fail-on", "severe"]
)
with pytest.raises(SystemExit):
cli_main.parse_arguments()
assert "--fail-on" in capsys.readouterr().err
def test_parse_fail_on_requires_non_interactive(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
_stub_settings(monkeypatch)
monkeypatch.setattr(sys, "argv", ["strix", "-t", "https://test.com/", "--fail-on", "high"])
with pytest.raises(SystemExit):
cli_main.parse_arguments()
assert "--fail-on only applies to headless runs" in capsys.readouterr().err
+123
View File
@@ -0,0 +1,123 @@
"""Tests for `strix --resume` without a run name."""
from __future__ import annotations
import importlib
import json
import sys
from typing import TYPE_CHECKING, Any
import pytest
if TYPE_CHECKING:
from pathlib import Path
cli_main: Any = importlib.import_module("strix.interface.main")
def _write_run(base: Path, name: str, *, state: bool = True) -> None:
run_dir = base / name
(run_dir / ".state").mkdir(parents=True)
(run_dir / "run.json").write_text(
json.dumps(
{
"run_name": name,
"status": "completed",
"start_time": "2026-10-04T10:00:00+00:00",
"targets_info": [
{
"type": "web_application",
"details": {"target_url": "https://example.com"},
"original": "https://example.com",
}
],
}
),
encoding="utf-8",
)
if state:
(run_dir / ".state" / "agents.json").write_text("{}", encoding="utf-8")
def test_bare_resume_defers_to_the_picker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.chdir(tmp_path)
_write_run(tmp_path / "strix_runs", "example-com_1111")
monkeypatch.setattr(sys, "argv", ["strix", "--resume"])
args = cli_main.parse_arguments()
assert args.resume is None
assert args.resume_picker is True
assert args.needs_setup is False
assert args.targets_info == []
def test_bare_resume_keeps_the_new_instruction_as_resume_guidance(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
_write_run(tmp_path / "strix_runs", "example-com_1111")
monkeypatch.setattr(sys, "argv", ["strix", "--resume", "--instruction", "focus on auth"])
args = cli_main.parse_arguments()
assert args.resume_picker is True
assert args.user_explicit_instruction == "focus on auth"
def test_bare_resume_with_no_runs_is_an_error(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(sys, "argv", ["strix", "--resume"])
with pytest.raises(SystemExit):
cli_main.parse_arguments()
assert "no runs in ./strix_runs" in capsys.readouterr().err
def test_bare_resume_headless_lists_the_runs_instead_of_prompting(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.chdir(tmp_path)
_write_run(tmp_path / "strix_runs", "example-com_1111")
_write_run(tmp_path / "strix_runs", "example-com_2222", state=False)
monkeypatch.setattr(sys, "argv", ["strix", "-n", "--resume"])
with pytest.raises(SystemExit):
cli_main.parse_arguments()
err = capsys.readouterr().err
assert "--resume needs a run name in headless mode" in err
assert "example-com_1111 completed 2026-10-04T10:00:00 https://example.com" in err
assert "example-com_2222 completed 2026-10-04T10:00:00 https://example.com" in err
assert "example-com_2222" in err
def test_bare_resume_still_rejects_targets(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.chdir(tmp_path)
_write_run(tmp_path / "strix_runs", "example-com_1111")
monkeypatch.setattr(sys, "argv", ["strix", "--resume", "-t", "https://example.com"])
with pytest.raises(SystemExit):
cli_main.parse_arguments()
assert "Cannot combine --resume with --target" in capsys.readouterr().err
def test_named_resume_without_agent_state_is_an_error(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.chdir(tmp_path)
_write_run(tmp_path / "strix_runs", "example-com_1111", state=False)
monkeypatch.setattr(sys, "argv", ["strix", "--resume", "example-com_1111"])
with pytest.raises(SystemExit):
cli_main.parse_arguments()
assert "never reached its first agent snapshot" in capsys.readouterr().err
+105
View File
@@ -0,0 +1,105 @@
from __future__ import annotations
import io
import logging
import sys
from typing import TYPE_CHECKING
import pytest
from strix.telemetry import logging as strix_logging
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
@pytest.fixture
def clean_loggers() -> Iterator[None]:
tracked = [logging.getLogger(name) for name in strix_logging._TRACKED_ROOTS]
saved = [(logger.handlers[:], logger.level, logger.propagate) for logger in tracked]
for logger in tracked:
logger.handlers = []
try:
yield
finally:
for logger, (handlers, level, propagate) in zip(tracked, saved, strict=True):
logger.handlers = handlers
logger.setLevel(level)
logger.propagate = propagate
def _stderr_handlers(logger: logging.Logger) -> list[logging.Handler]:
return [h for h in logger.handlers if getattr(h, strix_logging._STREAM_TAG, False)]
@pytest.mark.usefixtures("clean_loggers")
def test_console_logging_hides_warnings_and_shows_errors(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("STRIX_DEBUG", raising=False)
err = io.StringIO()
monkeypatch.setattr(sys, "stderr", err)
strix_logging.setup_console_logging()
logger = logging.getLogger("strix.llm.request_log")
logger.warning("llm_request outcome=error")
logger.error("boom")
out = err.getvalue()
assert "llm_request" not in out
assert "ERROR" in out
assert "strix.llm.request_log: boom" in out
@pytest.mark.usefixtures("clean_loggers")
def test_console_logging_follows_the_current_stderr(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("STRIX_DEBUG", raising=False)
first = io.StringIO()
monkeypatch.setattr(sys, "stderr", first)
strix_logging.setup_console_logging()
second = io.StringIO()
monkeypatch.setattr(sys, "stderr", second)
logging.getLogger("strix.test").error("after swap")
assert first.getvalue() == ""
assert "after swap" in second.getvalue()
@pytest.mark.usefixtures("clean_loggers")
def test_console_logging_is_idempotent() -> None:
strix_logging.setup_console_logging()
strix_logging.setup_console_logging()
for name in strix_logging._TRACKED_ROOTS:
assert len(_stderr_handlers(logging.getLogger(name))) == 1
@pytest.mark.usefixtures("clean_loggers")
def test_scan_logging_reuses_the_console_handler(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.delenv("STRIX_DEBUG", raising=False)
err = io.StringIO()
monkeypatch.setattr(sys, "stderr", err)
strix_logging.setup_console_logging()
teardown = strix_logging.setup_scan_logging(tmp_path)
strix_logger = logging.getLogger("strix")
assert len(_stderr_handlers(strix_logger)) == 1
logging.getLogger("strix.test").error("once")
assert err.getvalue().count("once") == 1
assert "once" in (tmp_path / "strix.log").read_text(encoding="utf-8")
teardown()
assert len(_stderr_handlers(strix_logger)) == 1
assert not any(isinstance(h, logging.FileHandler) for h in strix_logger.handlers)
@pytest.mark.usefixtures("clean_loggers")
def test_scan_logging_alone_still_attaches_a_stderr_handler(tmp_path: Path) -> None:
teardown = strix_logging.setup_scan_logging(tmp_path)
strix_logger = logging.getLogger("strix")
assert len(_stderr_handlers(strix_logger)) == 1
teardown()
assert _stderr_handlers(strix_logger) == []
+11 -31
View File
@@ -197,7 +197,7 @@ async def _send_message(connection: socket.socket, message: dict[str, Any]) -> N
@pytest.mark.asyncio
async def test_runtime_does_not_initialize_or_scan_before_ready(
async def test_runtime_prepares_state_before_launch_and_scans_only_after_ready(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime_args = args()
@@ -231,9 +231,9 @@ async def test_runtime_does_not_initialize_or_scan_before_ready(
monkeypatch.setattr(go_tui, "launch_tui_process", launch)
monkeypatch.setattr(go_tui, "wait_process", wait_process)
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
monkeypatch.setattr(go_tui, "persist_current", lambda: None)
monkeypatch.setattr(go_tui, "prepare_run", lambda _args: None)
monkeypatch.setattr(go_tui, "telemetry_start", lambda _args: None)
monkeypatch.setattr(go_tui, "persist_current", lambda: calls.append("persist"))
monkeypatch.setattr(go_tui, "prepare_run", lambda _args: calls.append("prepare"))
monkeypatch.setattr(go_tui, "telemetry_start", lambda _args: calls.append("telemetry"))
monkeypatch.setattr(runtime, "init_run_state", init_state)
monkeypatch.setattr(runtime, "start_scan", start_scan)
@@ -241,7 +241,7 @@ async def test_runtime_does_not_initialize_or_scan_before_ready(
try:
hello = await _receive_message(child)
assert hello["type"] == "hello"
assert calls == []
assert calls == ["state"]
await _send_message(
child,
{
@@ -258,7 +258,7 @@ async def test_runtime_does_not_initialize_or_scan_before_ready(
},
)
await asyncio.wait_for(run_task, timeout=2)
assert calls == ["preflight", "state", "scan"]
assert calls == ["state", "scan"]
finally:
child.close()
if not run_task.done():
@@ -355,7 +355,7 @@ async def test_setup_preflights_model_before_starting(
await runtime.ensure_model_verified()
await runtime.start_from_setup()
# The same steps, in the same order, as a direct launch's prepare_and_start.
# The same steps, in the same order, as main() takes before a direct launch.
assert calls == ["preflight", "persist", "targets", "prepare", "telemetry", "state", "scan"]
assert runtime.args.scan_mode == "quick"
assert runtime.args.instruction == ""
@@ -1009,28 +1009,7 @@ def _direct_launch_args() -> argparse.Namespace:
@pytest.mark.asyncio
async def test_prepare_and_start_reports_ordinary_connection_failures(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime = GoTuiRuntime(_direct_launch_args())
started: list[str] = []
async def preflight(_model: str) -> None:
raise TimeoutError("connection timed out")
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
monkeypatch.setattr(runtime, "start_scan", lambda: started.append("scan"))
await runtime.prepare_and_start()
assert started == []
assert runtime.controller.setup_mode is False
assert runtime.controller.scan_state == "failed"
assert "connection timed out" in (runtime.controller.error or "")
@pytest.mark.asyncio
async def test_prepare_and_start_runs_the_scan_after_preparation(
async def test_direct_launch_starts_the_prepared_scan_without_a_model_round_trip(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime = GoTuiRuntime(_direct_launch_args())
@@ -1046,9 +1025,10 @@ async def test_prepare_and_start_runs_the_scan_after_preparation(
monkeypatch.setattr(runtime, "init_run_state", lambda: order.append("state"))
monkeypatch.setattr(runtime, "start_scan", lambda: order.append("scan"))
await runtime.prepare_and_start()
assert runtime._start_preparation() is None
assert order == ["preflight", "persist", "prepare", "telemetry", "state", "scan"]
assert order == ["scan"]
assert runtime.controller.setup_mode is False
assert runtime.controller.scan_state == "running"
+7
View File
@@ -430,3 +430,10 @@ def test_user_headers_override_openrouter_attribution() -> None:
assert headers["X-Title"] == "Custom"
assert headers["X-Tenant"] == "acme"
assert headers["HTTP-Referer"] == "https://strix.ai"
def test_reasoning_effort_sent_as_configured_and_none_omitted() -> None:
assert make_model_settings("none", model_name="gpt-5.6-sol").reasoning is None
settings = make_model_settings("high", model_name="gpt-daybreak-blue-latest")
assert settings.reasoning is not None
assert settings.reasoning.effort == "high"
+127
View File
@@ -0,0 +1,127 @@
from __future__ import annotations
import argparse
import importlib
import sys
from types import SimpleNamespace
from typing import Any
import pytest
cli_main: Any = importlib.import_module("strix.interface.main")
report_state_module: Any = importlib.import_module("strix.report.state")
def _launch(
monkeypatch: pytest.MonkeyPatch, *, needs_setup: bool, resume_picker: bool = False
) -> list[str]:
calls: list[str] = []
args = argparse.Namespace(
non_interactive=False,
needs_setup=needs_setup,
resume_picker=resume_picker,
run_name=None,
fail_on=None,
)
async def run_tui(_args: argparse.Namespace) -> None:
calls.append("tui")
monkeypatch.setattr(sys, "argv", ["strix", "--target", "https://example.com"])
monkeypatch.setattr(cli_main, "setup_console_logging", lambda: None)
monkeypatch.setattr(cli_main, "start_import_warmup", lambda: None)
monkeypatch.setattr(cli_main, "parse_arguments", lambda: args)
monkeypatch.setattr(cli_main, "start_background_check", lambda: None)
monkeypatch.setattr(cli_main, "prompt_update_if_available", lambda _console: False)
monkeypatch.setattr(cli_main, "check_docker_installed", lambda: None)
monkeypatch.setattr(cli_main, "pull_docker_image", lambda: None)
monkeypatch.setattr(cli_main, "validate_environment", lambda: None)
monkeypatch.setattr(cli_main, "wait_for_import_warmup", lambda: None)
monkeypatch.setattr(cli_main, "_bootstrap_scan", lambda _args: calls.append("bootstrap"))
monkeypatch.setattr(cli_main, "_pick_run_to_resume", lambda _args: calls.append("pick"))
monkeypatch.setattr(cli_main, "run_tui", run_tui)
monkeypatch.setattr(cli_main, "notify_update", lambda _console: None)
cli_main.main()
return calls
def test_direct_launch_verifies_the_model_before_the_tui(monkeypatch: pytest.MonkeyPatch) -> None:
assert _launch(monkeypatch, needs_setup=False) == ["bootstrap", "tui"]
def test_start_screen_launch_defers_the_model_check_to_the_tui(
monkeypatch: pytest.MonkeyPatch,
) -> None:
assert _launch(monkeypatch, needs_setup=True) == ["tui"]
def test_bare_resume_picks_a_run_before_the_preflight(monkeypatch: pytest.MonkeyPatch) -> None:
assert _launch(monkeypatch, needs_setup=False, resume_picker=True) == [
"pick",
"bootstrap",
"tui",
]
def test_direct_launch_with_a_bad_key_prints_the_panel_and_exits_before_the_tui(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[str] = []
failure = cli_main.ModelConnectionError("openai/gpt-4o", RuntimeError("Error code: 401"))
async def warm_up_llm() -> None:
raise failure
monkeypatch.setattr(cli_main, "warm_up_llm", warm_up_llm)
monkeypatch.setattr(cli_main, "report_error", lambda *_args, **_kwargs: None)
monkeypatch.setattr(
cli_main,
"_print_model_connection_error",
lambda exc, model: calls.append(f"panel:{model}:{exc}"),
)
monkeypatch.setattr(cli_main, "persist_current", lambda: calls.append("persist"))
monkeypatch.setattr(cli_main, "prepare_run", lambda _args: calls.append("prepare"))
monkeypatch.setattr(cli_main, "set_scan_phase", lambda _phase: None)
with pytest.raises(SystemExit) as exit_info:
cli_main._bootstrap_scan(argparse.Namespace(non_interactive=False, needs_setup=False))
assert exit_info.value.code == 1
assert calls == ["panel:openai/gpt-4o:Error code: 401"]
def test_tui_startup_failure_marks_the_prepared_run_failed(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[str] = []
report_state = SimpleNamespace(cleanup=lambda status: calls.append(f"cleanup:{status}"))
args = argparse.Namespace(
non_interactive=False, needs_setup=False, resume_picker=False, run_name="run", fail_on=None
)
async def run_tui(_args: argparse.Namespace) -> None:
raise cli_main.InteractiveSetupUnavailableError("no sidecar")
monkeypatch.setattr(sys, "argv", ["strix", "--target", "https://example.com"])
monkeypatch.setattr(cli_main, "setup_console_logging", lambda: None)
monkeypatch.setattr(cli_main, "start_import_warmup", lambda: None)
monkeypatch.setattr(cli_main, "parse_arguments", lambda: args)
monkeypatch.setattr(cli_main, "start_background_check", lambda: None)
monkeypatch.setattr(cli_main, "prompt_update_if_available", lambda _console: False)
monkeypatch.setattr(cli_main, "check_docker_installed", lambda: None)
monkeypatch.setattr(cli_main, "pull_docker_image", lambda: None)
monkeypatch.setattr(cli_main, "validate_environment", lambda: None)
monkeypatch.setattr(cli_main, "wait_for_import_warmup", lambda: None)
monkeypatch.setattr(cli_main, "_bootstrap_scan", lambda _args: calls.append("bootstrap"))
monkeypatch.setattr(cli_main, "run_tui", run_tui)
monkeypatch.setattr(cli_main, "report_error", lambda *_args, **_kwargs: None)
monkeypatch.setattr(cli_main, "_print_error_panel", lambda *_args: calls.append("panel"))
monkeypatch.setattr(cli_main.posthog, "end", lambda *_args, **_kwargs: None)
monkeypatch.setattr(cli_main.scarf, "end", lambda *_args, **_kwargs: None)
monkeypatch.setattr(report_state_module, "get_global_report_state", lambda: report_state)
with pytest.raises(SystemExit) as exit_info:
cli_main.main()
assert exit_info.value.code == 1
assert calls == ["bootstrap", "panel", "cleanup:failed"]
+66 -2
View File
@@ -10,12 +10,14 @@ from agents.models import _openai_shared
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
from agents.models.openai_responses import OpenAIResponsesModel
from strix.config import models
from strix.config.models import (
StrixProvider,
_NonStreamingModel,
_TurnGuardModel,
configure_sdk_model_defaults,
request_timeout_extra_args,
resolve_api_type,
routes_through_litellm,
supports_strict_tool_schemas,
uses_chat_completions_tool_schema,
@@ -111,7 +113,7 @@ def test_api_type_override_settings(monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.mark.parametrize(
("api_type", "expected"),
[
(None, OpenAIChatCompletionsModel),
(None, OpenAIResponsesModel),
("chat_completions", OpenAIChatCompletionsModel),
("responses", OpenAIResponsesModel),
],
@@ -119,7 +121,7 @@ def test_api_type_override_settings(monkeypatch: pytest.MonkeyPatch) -> None:
def test_api_type_overrides_the_api_base_route(
monkeypatch: pytest.MonkeyPatch, api_type: str | None, expected: type
) -> None:
"""``LLM_API_BASE`` defaults to chat completions. ``STRIX_API_TYPE`` must win."""
"""gpt-5 is catalogued on /v1/responses, so a base URL alone changes nothing."""
monkeypatch.setattr(_openai_shared, "_use_responses_by_default", True)
monkeypatch.setattr(_openai_shared, "_default_openai_client", None)
monkeypatch.setattr(_openai_shared, "_default_openai_key", None)
@@ -138,3 +140,65 @@ def test_api_type_overrides_the_api_base_route(
while isinstance(model, _NonStreamingModel | _TurnGuardModel | RequestLoggingModel):
model = model._inner
assert isinstance(model, expected)
def _settings(monkeypatch: pytest.MonkeyPatch, model: str, api_base: str | None) -> Settings:
monkeypatch.setenv("STRIX_LLM", model)
monkeypatch.delenv("STRIX_API_TYPE", raising=False)
for name in ("LLM_API_BASE", "OPENAI_API_BASE", "OPENAI_BASE_URL"):
monkeypatch.delenv(name, raising=False)
if api_base is not None:
monkeypatch.setenv("LLM_API_BASE", api_base)
return Settings()
@pytest.mark.parametrize(
"api_base", [None, "https://api.openai.com/v1", "https://gateway.example/v1"]
)
def test_resolve_api_type_follows_the_catalog_not_the_base_url(
monkeypatch: pytest.MonkeyPatch, api_base: str | None
) -> None:
"""Responses when LiteLLM lists /v1/responses for the model, chat completions otherwise."""
settings = _settings(monkeypatch, "gpt-5", api_base)
for model in ("gpt-5", "gpt-5.6-sol", "openai/gpt-5.4", "gpt-daybreak-blue-latest"):
assert resolve_api_type(model, settings) == "responses", model
assert uses_chat_completions_tool_schema(model, settings) is False, model
for model in ("gpt-4o", "my-private-model"):
assert resolve_api_type(model, settings) == "chat_completions", model
assert uses_chat_completions_tool_schema(model, settings) is True, model
def test_resolve_api_type_explicit_override_wins(monkeypatch: pytest.MonkeyPatch) -> None:
_settings(monkeypatch, "gpt-daybreak-blue-latest", "https://gateway.example/v1")
monkeypatch.setenv("STRIX_API_TYPE", "chat_completions")
assert resolve_api_type("gpt-daybreak-blue-latest", Settings()) == "chat_completions"
monkeypatch.setenv("STRIX_API_TYPE", "Responses")
assert resolve_api_type("gpt-5", Settings()) == "responses"
@pytest.mark.parametrize(
("raw", "expected"),
[("None", "none"), ("HIGH", "high"), (" xhigh ", "xhigh"), ("Max", "max")],
)
def test_reasoning_effort_is_case_insensitive(
monkeypatch: pytest.MonkeyPatch, raw: str, expected: str
) -> None:
monkeypatch.setenv("STRIX_REASONING_EFFORT", raw)
monkeypatch.setenv("STRIX_DEDUPE_REASONING_EFFORT", raw)
settings = Settings()
assert settings.llm.reasoning_effort == expected
assert settings.dedupe.reasoning_effort == expected
def test_configure_sdk_api_route_follows_the_given_model(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A ``model=`` override picks its own route, not ``STRIX_LLM``'s."""
routes: list[str] = []
monkeypatch.setattr(models, "set_default_openai_api", routes.append)
settings = _settings(monkeypatch, "gpt-5", "https://gateway.example/v1")
models.configure_sdk_api_route("gpt-5", settings)
models.configure_sdk_api_route("my-private-model", settings)
assert routes == ["responses", "chat_completions"]
+152
View File
@@ -0,0 +1,152 @@
from __future__ import annotations
import asyncio
import pytest
from strix.config import Settings
from strix.interface.scan_setup import check_header_safe_credentials, preflight_model_connection
_ENV_KEYS = (
"LLM_API_KEY",
"OPENAI_API_KEY",
"LLM_EXTRA_HEADERS",
"LLM_API_BASE",
"STRIX_LLM",
"STRIX_DEDUPE_MODEL",
"DEDUPE_LLM_API_KEY",
"DEDUPE_LLM_EXTRA_HEADERS",
)
@pytest.fixture(autouse=True)
def _clean_env(monkeypatch: pytest.MonkeyPatch) -> None:
for key in _ENV_KEYS:
monkeypatch.delenv(key, raising=False)
def _settings(monkeypatch: pytest.MonkeyPatch, **env: str) -> Settings:
for key, value in env.items():
monkeypatch.setenv(key, value)
return Settings()
def test_ascii_credentials_pass(monkeypatch: pytest.MonkeyPatch) -> None:
settings = _settings(
monkeypatch,
LLM_API_KEY="sk-plain-ascii",
LLM_EXTRA_HEADERS='{"X-Team": "security"}',
)
check_header_safe_credentials(settings)
def test_smart_quote_in_api_key_is_named_without_leaking_the_key(
monkeypatch: pytest.MonkeyPatch,
) -> None:
settings = _settings(monkeypatch, LLM_API_KEY="sk-abc\u201d")
with pytest.raises(ValueError, match="LLM_API_KEY") as excinfo:
check_header_safe_credentials(settings)
message = str(excinfo.value)
assert "U+201D" in message
assert "RIGHT DOUBLE QUOTATION MARK" in message
assert "position 7 of 7" in message
assert "sk-abc" not in message
@pytest.mark.parametrize("char", ["\u00a0", "\ufeff", "\u200b"])
def test_invisible_characters_are_reported(monkeypatch: pytest.MonkeyPatch, char: str) -> None:
settings = _settings(monkeypatch, LLM_API_KEY=f"{char}sk-abc")
with pytest.raises(ValueError, match=f"U\\+{ord(char):04X}"):
check_header_safe_credentials(settings)
def test_extra_header_value_is_checked(monkeypatch: pytest.MonkeyPatch) -> None:
settings = _settings(
monkeypatch,
LLM_API_KEY="sk-plain",
LLM_EXTRA_HEADERS='{"X-Team": "s\u00e9curit\u00e9"}',
)
with pytest.raises(ValueError, match=r"LLM_EXTRA_HEADERS value for 'X-Team'.*U\+00E9"):
check_header_safe_credentials(settings)
def test_subscription_model_ignores_an_unused_main_key(monkeypatch: pytest.MonkeyPatch) -> None:
settings = _settings(monkeypatch, STRIX_LLM="chatgpt/gpt-5", LLM_API_KEY="sk-abc\u201d")
check_header_safe_credentials(settings)
def test_subscription_model_still_checks_extra_headers(monkeypatch: pytest.MonkeyPatch) -> None:
settings = _settings(
monkeypatch,
STRIX_LLM="chatgpt/gpt-5",
LLM_EXTRA_HEADERS='{"X-Team": "s\u00e9curit\u00e9"}',
)
with pytest.raises(ValueError, match=r"LLM_EXTRA_HEADERS value for 'X-Team'.*U\+00E9"):
check_header_safe_credentials(settings)
def test_dedupe_key_is_checked_when_a_dedupe_model_is_set(
monkeypatch: pytest.MonkeyPatch,
) -> None:
settings = _settings(
monkeypatch,
LLM_API_KEY="sk-plain",
STRIX_DEDUPE_MODEL="openai/gpt-4o-mini",
DEDUPE_LLM_API_KEY="sk-de\u00a0dupe",
)
with pytest.raises(ValueError, match=r"DEDUPE_LLM_API_KEY.*U\+00A0.*position 6 of 10"):
check_header_safe_credentials(settings)
def test_dedupe_key_is_checked_as_sent_after_resolve_dedupe_model_strips_it(
monkeypatch: pytest.MonkeyPatch,
) -> None:
settings = _settings(
monkeypatch,
LLM_API_KEY="sk-plain",
STRIX_DEDUPE_MODEL="openai/gpt-4o-mini",
DEDUPE_LLM_API_KEY="\u00a0sk-dedupe\u00a0",
)
check_header_safe_credentials(settings)
def test_dedupe_subscription_model_checks_headers_but_not_the_key(
monkeypatch: pytest.MonkeyPatch,
) -> None:
settings = _settings(
monkeypatch,
LLM_API_KEY="sk-plain",
STRIX_DEDUPE_MODEL="chatgpt/gpt-5",
DEDUPE_LLM_API_KEY="sk-dedupe\u201d",
DEDUPE_LLM_EXTRA_HEADERS='{"X-Team": "s\u00e9curit\u00e9"}',
)
with pytest.raises(ValueError, match=r"DEDUPE_LLM_EXTRA_HEADERS value for 'X-Team'"):
check_header_safe_credentials(settings)
def test_dedupe_key_is_ignored_without_a_dedupe_model(monkeypatch: pytest.MonkeyPatch) -> None:
settings = _settings(monkeypatch, LLM_API_KEY="sk-plain", DEDUPE_LLM_API_KEY="sk-dedupe\u00a0")
check_header_safe_credentials(settings)
def test_preflight_rejects_the_key_before_any_request(monkeypatch: pytest.MonkeyPatch) -> None:
settings = _settings(
monkeypatch,
LLM_API_KEY="sk-abc\u201d",
LLM_API_BASE="http://127.0.0.1:9/v1",
)
with pytest.raises(ValueError, match="LLM_API_KEY"):
asyncio.run(preflight_model_connection("openai/gpt-4o-mini", settings=settings))
+7 -1
View File
@@ -13,7 +13,13 @@ def test_resolves_common_bare_model_names() -> None:
resolve_litellm_model.cache_clear()
assert resolve_litellm_model("deepseek-v4-flash") == "deepseek/deepseek-v4-flash"
assert resolve_litellm_model("openai/deepseek-v4-flash") == "deepseek/deepseek-v4-flash"
assert resolve_litellm_model("grok-4.5") == "xai/grok-4.5"
# LiteLLM prices grok-4.5 identically under xai/, perplexity/xai/ and
# openrouter/x-ai/, so the resolver may return any of them; the contract
# is a provider-qualified name LiteLLM can price.
grok = resolve_litellm_model("grok-4.5")
assert grok is not None
assert grok.endswith("/grok-4.5")
assert grok in litellm.model_cost
# MiniMax-M3 is sold by several LiteLLM providers at different prices, so
# the resolver must not guess from its bare name. A provider-qualified
# model remains deterministic.
+6 -1
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import json
import re
from typing import TYPE_CHECKING, Any, cast
import pytest
@@ -1085,7 +1086,11 @@ def test_tool_descriptions_include_formatting_guidance() -> None:
finish_desc = finish_scan.description
assert "markdown" in finish_desc.lower()
assert "# Executive Summary" in finish_desc
assert "section body" in finish_desc.lower()
assert "do not start a field with a heading" in finish_desc.lower()
example = finish_desc.split("Example (abbreviated", 1)[1]
heading_lines = [line for line in example.splitlines() if re.match(r"^\s*#{1,6}\s", line)]
assert heading_lines == []
dep_desc = create_dependency_report.description
assert "cve" in dep_desc.lower()
+260
View File
@@ -0,0 +1,260 @@
from __future__ import annotations
import io
import os
import sys
from datetime import UTC, datetime, timedelta
import pytest
from rich.console import Console
from strix.interface.resume_picker import (
KEY_BACKSPACE,
KEY_DOWN,
KEY_END,
KEY_ENTER,
KEY_ESCAPE,
KEY_INTERRUPT,
KEY_UP,
PickerUnavailableError,
ResumePicker,
filter_runs,
pick_run,
read_raw_key,
relative_time,
translate_key,
)
from strix.report.runs import RunSummary
NOW = datetime(2026, 10, 4, 16, 0, tzinfo=UTC)
def _run(
name: str,
target: str,
*,
minutes_ago: int,
status: str = "completed",
findings: int = 0,
resumable: bool = True,
) -> RunSummary:
started = (NOW - timedelta(minutes=minutes_ago)).isoformat()
return RunSummary(
run_name=name,
target=target,
started_at=started,
ended_at=started,
status=status,
findings=findings,
resumable=resumable,
)
RUNS = [
_run("example-com_e5e6", "https://example.com", minutes_ago=12),
_run(
"example-com_223e", "https://example.com", minutes_ago=40, status="interrupted", findings=2
),
_run(
"juice-shop_8a1f",
"https://juice-shop.herokuapp.com",
minutes_ago=60 * 26,
status="stopped",
findings=7,
),
_run(
"strix_41c0",
"/home/me/code/strix (workspace)",
minutes_ago=60 * 50,
status="failed",
resumable=False,
),
]
def _console(width: int = 110, height: int = 30) -> tuple[Console, io.StringIO]:
buffer = io.StringIO()
console = Console(
file=buffer, width=width, height=height, force_terminal=True, color_system="truecolor"
)
return console, buffer
def _picker(runs: list[RunSummary] = RUNS) -> ResumePicker:
console, _ = _console()
return ResumePicker(runs, console=console, runs_dir="strix_runs", now=NOW)
def _plain(picker: ResumePicker) -> str:
return "\n".join(line.plain for line in picker.render())
def test_translate_key_maps_control_characters() -> None:
assert translate_key("\r") == KEY_ENTER
assert translate_key("\n") == KEY_ENTER
assert translate_key("\x1b") == KEY_ESCAPE
assert translate_key("\x7f") == KEY_BACKSPACE
assert translate_key("\x03") == KEY_INTERRUPT
assert translate_key("a") == "a"
def test_relative_time_buckets() -> None:
assert relative_time((NOW - timedelta(seconds=5)).isoformat(), NOW) == "just now"
assert relative_time((NOW - timedelta(minutes=12)).isoformat(), NOW) == "12 min ago"
assert relative_time((NOW - timedelta(hours=3, minutes=1)).isoformat(), NOW) == "3 h ago"
assert relative_time((NOW - timedelta(days=2)).isoformat(), NOW) == "2 d ago"
assert relative_time("2026-01-02T15:04:00+00:00", NOW).startswith("Jan 2")
assert relative_time("2025-01-02T15:04:00+00:00", NOW).endswith("2025")
assert relative_time("not a date", NOW) == "not a date"
assert relative_time("", NOW) == "unknown"
def test_filter_matches_name_target_and_status() -> None:
assert [run.run_name for run in filter_runs(RUNS, "juice")] == ["juice-shop_8a1f"]
assert [run.run_name for run in filter_runs(RUNS, "INTERRUPTED")] == ["example-com_223e"]
assert [run.run_name for run in filter_runs(RUNS, "example.com")] == [
"example-com_e5e6",
"example-com_223e",
]
assert filter_runs(RUNS, "") == RUNS
def test_render_lists_every_run_with_its_metadata() -> None:
text = _plain(_picker())
assert "Resume a run" in text
assert "4 runs in ./strix_runs" in text
for run in RUNS:
assert run.run_name in text
assert "12 min ago" in text
assert "https://juice-shop.herokuapp.com" in text
assert "interrupted" in text
assert "failed · no state" in text
assert text.splitlines()[3].startswith(" \u276f ")
def test_enter_returns_the_highlighted_run() -> None:
keys = iter([KEY_DOWN, KEY_ENTER])
assert _picker().run(lambda: next(keys)) == RUNS[1]
def test_cursor_is_clamped_and_end_jumps_to_the_last_run() -> None:
picker = _picker()
picker.handle(KEY_UP)
assert picker.cursor == 0
picker.handle(KEY_END)
assert picker.cursor == len(RUNS) - 1
for _ in range(10):
picker.handle(KEY_DOWN)
assert picker.cursor == len(RUNS) - 1
def test_escape_cancels_and_clears_the_search_first() -> None:
keys = iter(["j", "u", KEY_ESCAPE, KEY_ESCAPE])
picker = _picker()
assert picker.run(lambda: next(keys)) is None
assert picker.filter == ""
def test_typing_filters_and_backspace_restores() -> None:
picker = _picker()
for char in "juice":
picker.handle(char)
assert [run.run_name for run in picker.rows] == ["juice-shop_8a1f"]
assert "search: juice" in _plain(picker)
for _ in range(5):
picker.handle(KEY_BACKSPACE)
assert len(picker.rows) == len(RUNS)
def test_runs_without_agent_state_cannot_be_picked() -> None:
picker = _picker()
picker.handle(KEY_END)
assert picker.handle(KEY_ENTER) == (False, None)
assert "no saved agent state" in _plain(picker)
def test_no_match_renders_a_hint_and_enter_does_nothing() -> None:
picker = _picker()
for char in "zzz":
picker.handle(char)
assert "no runs match 'zzz'" in _plain(picker)
assert picker.handle(KEY_ENTER) == (False, None)
def test_ctrl_c_raises_keyboard_interrupt() -> None:
with pytest.raises(KeyboardInterrupt):
_picker().handle(KEY_INTERRUPT)
def test_long_lists_scroll_to_keep_the_cursor_visible() -> None:
runs = [
_run(f"run_{index:02d}", f"https://host{index}.example", minutes_ago=index)
for index in range(40)
]
console, _ = _console(height=12)
picker = ResumePicker(runs, console=console, runs_dir="strix_runs", now=NOW)
text = _plain(picker)
assert "run_00" in text
assert "more above" not in text
assert "more below" in text
picker.handle(KEY_END)
text = _plain(picker)
assert "run_39" in text
assert "run_00" not in text
assert "more above" in text
assert "more below" not in text
def test_run_redraws_in_place_and_restores_the_cursor() -> None:
console, buffer = _console()
picker = ResumePicker(RUNS, console=console, runs_dir="strix_runs", now=NOW)
keys = iter([KEY_DOWN, KEY_ENTER])
picker.run(lambda: next(keys))
output = buffer.getvalue()
assert output.startswith("\x1b[?25l")
assert output.endswith("\x1b[?25h")
assert "\x1b[J" in output
assert "example-com_223e" in output
def test_pick_run_needs_a_terminal() -> None:
console, _ = _console()
with pytest.raises(PickerUnavailableError):
pick_run(RUNS, runs_dir="strix_runs", console=console, stdin=io.StringIO())
def test_narrow_terminals_never_wrap_a_row() -> None:
console, _ = _console(width=48)
picker = ResumePicker(RUNS, console=console, runs_dir="strix_runs", now=NOW)
for line in picker.render():
assert len(line.plain) <= 47, line.plain
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX key reader")
def test_posix_key_reader_handles_multibyte_and_arrows() -> None:
master, slave = os.openpty()
try:
with os.fdopen(slave, "r+b", buffering=0) as stream:
os.write(master, "\u00e9".encode())
assert read_raw_key(stream) == "\u00e9" # type: ignore[arg-type]
os.write(master, b"\x1b[B")
assert read_raw_key(stream) == KEY_DOWN # type: ignore[arg-type]
os.write(master, b"\x1b[6~")
assert read_raw_key(stream) == "pagedown" # type: ignore[arg-type]
os.write(master, b"\r")
assert translate_key(read_raw_key(stream)) == KEY_ENTER # type: ignore[arg-type]
finally:
os.close(master)
def test_tall_terminals_still_show_a_short_window() -> None:
console, _ = _console(height=60)
runs = [
_run(f"run_{index:02d}", f"https://host{index}.example", minutes_ago=index)
for index in range(40)
]
picker = ResumePicker(runs, console=console, runs_dir="strix_runs", now=NOW)
plain = [line.plain for line in picker.render()]
assert sum(1 for line in plain if "run_" in line) == 8
assert any("more below" in line for line in plain)
+141
View File
@@ -0,0 +1,141 @@
"""Tests for the run listing behind the resume picker."""
from __future__ import annotations
import json
import os
from typing import TYPE_CHECKING
from strix.report.runs import list_run_summaries
if TYPE_CHECKING:
from pathlib import Path
import pytest
def _write_run(
base: Path,
name: str,
record: object,
*,
modified: float,
state: bool = True,
findings: int | None = None,
) -> None:
run_dir = base / name
run_dir.mkdir(parents=True)
path = run_dir / "run.json"
path.write_text(record if isinstance(record, str) else json.dumps(record), encoding="utf-8")
if state:
(run_dir / ".state").mkdir()
(run_dir / ".state" / "agents.json").write_text("{}", encoding="utf-8")
if findings is not None:
(run_dir / "vulnerabilities.json").write_text(
json.dumps([{"id": index} for index in range(findings)]), encoding="utf-8"
)
os.utime(path, (modified, modified))
def test_lists_runs_newest_first_with_target_status_and_findings(tmp_path: Path) -> None:
base = tmp_path / "strix_runs"
_write_run(
base,
"example-com_1111",
{
"status": "completed",
"start_time": "2026-10-04T10:00:00+00:00",
"end_time": "2026-10-04T10:05:00+00:00",
"targets_info": [{"type": "web_application", "original": "https://example.com"}],
},
modified=1_000,
findings=2,
)
_write_run(
base,
"example-com_2222",
{
"status": "interrupted",
"start_time": "2026-10-04T11:00:00+00:00",
"targets_info": [
{"type": "web_application", "original": "https://example.com"},
{"type": "web_application", "original": "https://api.example.com"},
],
},
modified=3_000,
)
(base / "not-a-run").mkdir()
runs = list_run_summaries(cwd=tmp_path)
assert [run.run_name for run in runs] == ["example-com_2222", "example-com_1111"]
newest, oldest = runs
assert newest.target == "https://example.com +1 more"
assert newest.status == "interrupted"
assert newest.started_at == "2026-10-04T11:00:00+00:00"
assert newest.ended_at == ""
assert newest.findings == 0
assert newest.resumable is True
assert oldest.target == "https://example.com"
assert oldest.findings == 2
assert oldest.ended_at == "2026-10-04T10:05:00+00:00"
def test_target_less_runs_describe_their_workspace_or_instruction(tmp_path: Path) -> None:
base = tmp_path / "strix_runs"
_write_run(
base,
"pentest_aaaa",
{"status": "stopped", "targets_info": [], "workspace_mount": "/home/me/app"},
modified=2_000,
)
_write_run(
base,
"pentest_bbbb",
{"status": "failed", "targets_info": [], "user_instruction": "audit the login\nflow"},
modified=1_000,
)
runs = {run.run_name: run for run in list_run_summaries(cwd=tmp_path)}
assert runs["pentest_aaaa"].target == "/home/me/app (workspace)"
assert runs["pentest_bbbb"].target == "audit the login"
def test_runs_without_agent_state_or_with_a_broken_record_are_listed_not_resumable(
tmp_path: Path,
) -> None:
base = tmp_path / "strix_runs"
_write_run(base, "no-state_cccc", {"status": "failed"}, modified=2_000, state=False)
_write_run(base, "broken_dddd", "{not json", modified=1_000)
_write_run(base, "list_eeee", "[]", modified=500)
runs = list_run_summaries(cwd=tmp_path)
assert [(run.run_name, run.resumable) for run in runs] == [
("no-state_cccc", False),
("broken_dddd", True),
("list_eeee", True),
]
assert runs[1].status == "unknown"
assert runs[1].target == ""
def test_missing_runs_dir_lists_nothing(tmp_path: Path) -> None:
assert list_run_summaries(cwd=tmp_path) == []
def test_blank_instruction_does_not_break_the_listing(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
run_dir = tmp_path / "strix_runs" / "blank_0001"
run_dir.mkdir(parents=True)
(run_dir / "run.json").write_text(
json.dumps({"status": "completed", "instruction": " \n "}), encoding="utf-8"
)
[summary] = list_run_summaries()
assert summary.target == ""
+1 -1
View File
@@ -106,7 +106,7 @@ class _Gateway(BaseHTTPRequestHandler):
def log_message(self, *args: Any) -> None:
pass
def do_POST(self) -> None: # noqa: N802
def do_POST(self) -> None:
length = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(length) or b"{}")
messages = body.get("messages", [])
+45
View File
@@ -0,0 +1,45 @@
"""Every parameter of every scan-agent tool carries a description in its JSON schema.
The model only sees a parameter's docstring text through the schema. An ``Args:``
section that the docstring parser fails to recognise (for example because an
earlier line opens a markdown code fence it never closes) silently drops every
parameter description, so optional-but-conditionally-required fields such as
``confidence_rationale`` become anonymous nullable strings the model never fills.
"""
from __future__ import annotations
import pytest
from agents.tool import FunctionTool
from strix.agents import factory
from strix.tools.agents_graph.tools import agent_finish
from strix.tools.finish.tool import finish_scan
from strix.tools.reporting.tool import create_vulnerability_report
from strix.tools.respond.tool import respond_to_user
_SCAN_AGENT_TOOLS = [
tool
for tool in (*factory._BASE_TOOLS, finish_scan, agent_finish, respond_to_user)
if isinstance(tool, FunctionTool)
]
@pytest.mark.parametrize("tool", _SCAN_AGENT_TOOLS, ids=lambda tool: tool.name)
def test_every_parameter_has_a_description(tool: FunctionTool) -> None:
properties = tool.params_json_schema.get("properties", {})
missing = sorted(
name
for name, schema in properties.items()
if not (isinstance(schema.get("description"), str) and schema["description"].strip())
)
assert not missing, f"{tool.name}: parameters without a description: {missing}"
def test_create_vulnerability_report_explains_confidence_rationale() -> None:
schema = create_vulnerability_report.params_json_schema["properties"]
assert "confidence" in create_vulnerability_report.params_json_schema["required"]
rationale = schema["confidence_rationale"]["description"]
assert "Required when" in rationale
assert "high" in rationale
-53
View File
@@ -1,53 +0,0 @@
import sys
import pytest
import urllib3.response
from strix.telemetry import logging as tlog
from strix.telemetry.logging import _is_urllib3_closed_file_noise
class _Args:
def __init__(self, exc_value: BaseException | None, obj: object) -> None:
self.exc_type = type(exc_value) if exc_value is not None else None
self.exc_value = exc_value
self.exc_traceback = None
self.err_msg = None
self.object = obj
def _urllib3_response() -> urllib3.response.HTTPResponse:
return urllib3.response.HTTPResponse(body=b"")
def test_filters_urllib3_closed_file_noise() -> None:
args = _Args(ValueError("I/O operation on closed file."), _urllib3_response())
assert _is_urllib3_closed_file_noise(args) # type: ignore[arg-type]
def test_passes_through_other_unraisables() -> None:
assert not _is_urllib3_closed_file_noise(
_Args(ValueError("I/O operation on closed file."), object()) # type: ignore[arg-type]
)
assert not _is_urllib3_closed_file_noise(
_Args(RuntimeError("boom"), _urllib3_response()) # type: ignore[arg-type]
)
assert not _is_urllib3_closed_file_noise(
_Args(ValueError("something else"), _urllib3_response()) # type: ignore[arg-type]
)
def test_installed_hook_filters_and_delegates(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[object] = []
monkeypatch.setattr(sys, "unraisablehook", calls.append)
monkeypatch.setattr(tlog, "_unraisable_hook_installed", False)
tlog._silence_urllib3_finalizer_noise()
hook = sys.unraisablehook
assert hook is not calls.append
hook(_Args(ValueError("I/O operation on closed file."), _urllib3_response())) # type: ignore[arg-type]
assert calls == []
other = _Args(RuntimeError("boom"), object())
hook(other) # type: ignore[arg-type]
assert calls == [other]
+140
View File
@@ -0,0 +1,140 @@
from __future__ import annotations
import logging
import os
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace
from typing import TYPE_CHECKING
import pytest
from strix.telemetry import logging as tlog
if TYPE_CHECKING:
from collections.abc import Iterator
def _unraisable(exc: BaseException, obj: object, err_msg: str | None = None) -> object:
return SimpleNamespace(
exc_type=type(exc),
exc_value=exc,
exc_traceback=None,
err_msg=err_msg,
object=obj,
)
@pytest.fixture
def strix_records() -> Iterator[list[logging.LogRecord]]:
records: list[logging.LogRecord] = []
class _Collect(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
records.append(record)
handler = _Collect()
root = logging.getLogger("strix")
root.addHandler(handler)
try:
yield records
finally:
root.removeHandler(handler)
@pytest.fixture
def default_hook_calls(monkeypatch: pytest.MonkeyPatch) -> list[object]:
calls: list[object] = []
monkeypatch.setattr(sys, "unraisablehook", calls.append)
tlog.configure_dependency_logging()
return calls
@pytest.mark.parametrize(
"args",
[
_unraisable(ValueError("I/O operation on closed file."), object()),
_unraisable(
ValueError("I/O operation on closed file."),
None,
err_msg=(
"Exception ignored while finalizing file "
"<urllib3.response.HTTPResponse object at 0x1>"
),
),
_unraisable(RuntimeError("boom"), object()),
_unraisable(KeyError("x"), None, err_msg="Exception ignored in: <function f>"),
],
)
def test_every_unraisable_is_logged_not_printed(
args: object,
default_hook_calls: list[object],
strix_records: list[logging.LogRecord],
capsys: pytest.CaptureFixture[str],
) -> None:
sys.unraisablehook(args) # type: ignore[arg-type]
assert default_hook_calls == []
assert capsys.readouterr().err == ""
[record] = strix_records
assert record.levelno == logging.WARNING
assert record.name == "strix.telemetry"
message = logging.Formatter().format(record)
if args.err_msg: # type: ignore[attr-defined]
assert message.startswith(args.err_msg) # type: ignore[attr-defined]
else:
assert message.startswith("Exception ignored in <object object")
assert str(args.exc_value) in message # type: ignore[attr-defined]
@pytest.mark.usefixtures("default_hook_calls")
def test_unraisable_is_dropped_when_strix_has_no_handlers(
capsys: pytest.CaptureFixture[str],
) -> None:
root = logging.getLogger("strix")
saved, root.handlers = root.handlers, []
try:
sys.unraisablehook(_unraisable(RuntimeError("late"), object())) # type: ignore[arg-type]
finally:
root.handlers = saved
assert capsys.readouterr().err == ""
_EXIT_SCRIPT = r"""
import io, sys
from strix.telemetry.logging import setup_console_logging
setup_console_logging()
class Leaky:
def __del__(self):
raise ValueError("I/O operation on closed file.")
Leaky()
keep = Leaky()
import urllib3.response
resp = urllib3.response.HTTPResponse(body=io.BytesIO(b""), preload_content=False)
resp._fp.close()
del resp
print("done")
"""
def test_process_exit_prints_nothing_to_stderr() -> None:
env = {
k: v for k, v in os.environ.items() if k in {"PATH", "HOME", "SYSTEMROOT", "TEMP", "TMP"}
}
env |= {"PYTHONPATH": str(Path(__file__).resolve().parents[1]), "STRIX_DEBUG": ""}
proc = subprocess.run( # noqa: S603
[sys.executable, "-c", _EXIT_SCRIPT],
capture_output=True,
text=True,
timeout=60,
env=env,
check=False,
)
assert proc.returncode == 0, proc.stderr
assert proc.stdout.strip() == "done"
assert proc.stderr == ""
+63
View File
@@ -0,0 +1,63 @@
from __future__ import annotations
import io
import sys
import pytest
from rich.console import Console
from strix.interface.main import _force_utf8_streams
from strix.interface.tui.runtime import _open_output_sink
UNICODE_SAMPLE = "Reflected XSS in / \u2713 \u2192 \U0001f50d \u00b7 done"
def _cp1252_stream() -> tuple[io.TextIOWrapper, io.BytesIO]:
raw = io.BytesIO()
return io.TextIOWrapper(raw, encoding="cp1252", line_buffering=True), raw
def test_rich_output_raises_on_a_cp1252_stream(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(sys, "stdout", _cp1252_stream()[0])
monkeypatch.setattr(sys, "stderr", _cp1252_stream()[0])
with pytest.raises(UnicodeEncodeError):
Console().print(UNICODE_SAMPLE)
def test_force_utf8_streams_lets_rich_print_unicode(monkeypatch: pytest.MonkeyPatch) -> None:
stdout, stdout_bytes = _cp1252_stream()
stderr, stderr_bytes = _cp1252_stream()
monkeypatch.setattr(sys, "stdout", stdout)
monkeypatch.setattr(sys, "stderr", stderr)
_force_utf8_streams()
assert stdout.encoding == "utf-8"
assert stderr.encoding == "utf-8"
Console().print(UNICODE_SAMPLE)
Console(stderr=True).print(UNICODE_SAMPLE)
stdout.flush()
stderr.flush()
assert UNICODE_SAMPLE in stdout_bytes.getvalue().decode("utf-8")
assert UNICODE_SAMPLE in stderr_bytes.getvalue().decode("utf-8")
def test_force_utf8_streams_skips_streams_without_reconfigure(
monkeypatch: pytest.MonkeyPatch,
) -> None:
stdout = io.StringIO()
monkeypatch.setattr(sys, "stdout", stdout)
monkeypatch.setattr(sys, "stderr", io.StringIO())
_force_utf8_streams()
sys.stdout.write(UNICODE_SAMPLE)
assert stdout.getvalue() == UNICODE_SAMPLE
def test_tui_output_sink_accepts_unicode() -> None:
with _open_output_sink() as sink:
assert sink.encoding == "utf-8"
sink.write(UNICODE_SAMPLE + "\n")