Compare commits

..

1 Commits

38 changed files with 284 additions and 570 deletions

View File

@@ -1,6 +1,3 @@
# Built viewer bundles are generated output, not hand-edited source.
exclude: ^strix/interface/viewer/static/assets/
repos:
# Ruff for fast linting and formatting
- repo: https://github.com/astral-sh/ruff-pre-commit
@@ -12,18 +9,21 @@ repos:
- id: ruff-format
name: ruff-format
# MyPy for static type checking. Runs the project's own mypy from the uv
# environment (`make dev-install`) so it sees the same dependencies and
# stubs as `make check-all`.
- repo: local
# MyPy for static type checking
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.17.1
hooks:
- id: mypy
name: mypy
entry: uv run mypy
language: system
types_or: [python, pyi]
files: ^(strix|tests)/
require_serial: true
additional_dependencies: [
types-requests,
types-python-dateutil,
pydantic,
fastapi,
pytest,
hatchling,
"openai-agents[litellm]>=0.19.0,<0.20",
]
args: [--install-types, --non-interactive]
# Built-in hooks for basic file checks
- repo: https://github.com/pre-commit/pre-commit-hooks
@@ -62,6 +62,5 @@ ci:
autoupdate_branch: ""
autoupdate_commit_msg: "[pre-commit.ci] pre-commit autoupdate"
autoupdate_schedule: weekly
# pre-commit.ci cannot run `language: system` hooks; mypy runs via `make check-all`.
skip: [mypy]
skip: []
submodules: false

View File

@@ -28,7 +28,7 @@ Target-specific workflows built on the same engine:
- **Open-source CLI (self-hosted):** free, fully local, BYO LLM key, needs Docker. Best for local dev loops, air-gapped/offline, and full control.
```bash
curl -sSL https://strix.ai/install | bash # install
export STRIX_LLM="openrouter/z-ai/glm-5.3" # any LiteLLM model id
export STRIX_LLM="openai/gpt-5.4" # any LiteLLM model id
export LLM_API_KEY="<key>"
strix -n -t ./ --scan-mode quick --max-budget 10 # headless scan; always use -n
```

View File

@@ -31,7 +31,7 @@ Thank you for your interest in contributing to Strix! This guide will help you g
3. **Configure your LLM provider**
```bash
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="openai/gpt-5.4"
export LLM_API_KEY="your-api-key"
```

View File

@@ -82,7 +82,7 @@ Strix are autonomous AI penetration testing agents that act just like real hacke
curl -sSL https://strix.ai/install | bash
# Configure your AI provider
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="openai/gpt-5.4"
export LLM_API_KEY="your-api-key"
# Run your first security assessment
@@ -294,7 +294,7 @@ jobs:
### Configuration
```bash
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="openai/gpt-5.4"
export LLM_API_KEY="your-api-key"
# Optional
@@ -448,12 +448,9 @@ Each server's tools are namespaced by `name` (for example `local_fs_read_file`).
**Recommended models for best results:**
- [Z.ai GLM-5.3 on OpenRouter](https://openrouter.ai/z-ai/glm-5.3) - `openrouter/z-ai/glm-5.3` (the default pick)
- [OpenAI GPT-5.4](https://openai.com/api/) - `openai/gpt-5.4`
- [Anthropic Claude Sonnet 4.6](https://claude.com/platform/api) - `anthropic/claude-sonnet-4-6`
- [Google Gemini 3 Pro Preview](https://cloud.google.com/vertex-ai) - `vertex_ai/gemini-3-pro-preview`
- [DeepSeek V4 Pro](https://platform.deepseek.com) - `deepseek/deepseek-v4-pro`
- [Moonshot Kimi K3](https://platform.kimi.ai) - `moonshot/kimi-k3`
See the [LLM Providers documentation](https://docs.strix.ai/llm-providers/overview) for all supported providers including Vertex AI, Bedrock, Azure, and local models.

View File

@@ -8,7 +8,7 @@ Configure Strix using environment variables or a config file.
## LLM Configuration
<ParamField path="STRIX_LLM" type="string" required>
Model name in LiteLLM format (e.g., `openrouter/z-ai/glm-5.3`, `openai/gpt-5.4`).
Model name in LiteLLM format (e.g., `openai/gpt-5.4`, `anthropic/claude-sonnet-4-6`).
</ParamField>
<ParamField path="LLM_API_KEY" type="string">
@@ -145,7 +145,7 @@ strix --target ./app --config /path/to/config.json
```json
{
"env": {
"STRIX_LLM": "openrouter/z-ai/glm-5.3",
"STRIX_LLM": "openai/gpt-5.4",
"LLM_API_KEY": "sk-...",
"STRIX_REASONING_EFFORT": "high"
}
@@ -156,7 +156,7 @@ strix --target ./app --config /path/to/config.json
```bash
# Required
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="openai/gpt-5.4"
export LLM_API_KEY="sk-..."
# Optional: Enable web search

View File

@@ -33,7 +33,7 @@ description: "Contribute to Strix development"
</Step>
<Step title="Configure LLM">
```bash
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="openai/gpt-5.4"
export LLM_API_KEY="your-api-key"
```
</Step>

View File

@@ -78,7 +78,7 @@ Strix uses a graph of specialized agents for comprehensive security testing:
curl -sSL https://strix.ai/install | bash
# Configure
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="openai/gpt-5.4"
export LLM_API_KEY="your-api-key"
# Scan

View File

@@ -37,7 +37,7 @@ Add these secrets to your repository:
| Secret | Description |
|--------|-------------|
| `STRIX_LLM` | Model name (e.g., `openrouter/z-ai/glm-5.3`) |
| `STRIX_LLM` | Model name (e.g., `openai/gpt-5.4`) |
| `LLM_API_KEY` | API key for your LLM provider |
## Exit Codes

View File

@@ -17,9 +17,6 @@ export LLM_API_BASE="https://api.novita.ai/openai"
| Model | Configuration |
|-------|---------------|
| GLM-5.3 | `openai/zai-org/glm-5.3` |
| Kimi K3 | `openai/moonshotai/kimi-k3` |
| DeepSeek V4 Pro | `openai/deepseek/deepseek-v4-pro` |
| Kimi K2.5 | `openai/moonshotai/kimi-k2.5` |
| GLM-5 | `openai/zai-org/glm-5` |
| MiniMax M2.5 | `openai/minimax/minimax-m2.5` |

View File

@@ -8,7 +8,7 @@ description: "Configure Strix with models via OpenRouter"
## Setup
```bash
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="openrouter/openai/gpt-5.4"
export LLM_API_KEY="sk-or-..."
```
@@ -18,12 +18,9 @@ Access any model on OpenRouter using the format `openrouter/<provider>/<model>`:
| Model | Configuration |
|-------|---------------|
| GLM-5.3 (default) | `openrouter/z-ai/glm-5.3` |
| GPT-5.4 | `openrouter/openai/gpt-5.4` |
| Claude Sonnet 4.6 | `openrouter/anthropic/claude-sonnet-4.6` |
| Gemini 3 Pro | `openrouter/google/gemini-3-pro-preview` |
| DeepSeek V4 Pro | `openrouter/deepseek/deepseek-v4-pro` |
| Kimi K3 | `openrouter/moonshotai/kimi-k3` |
| GLM-4.7 | `openrouter/z-ai/glm-4.7` |
## Get API Key

View File

@@ -9,17 +9,14 @@ Strix uses [LiteLLM](https://docs.litellm.ai/docs/providers) for model compatibi
Set your model and API key:
| Model | Provider | Configuration |
| -------------------- | ----------------- | -------------------------------- |
| GLM-5.3 (default) | Z.ai (OpenRouter) | `openrouter/z-ai/glm-5.3` |
| GPT-5.4 | OpenAI | `openai/gpt-5.4` |
| Claude Sonnet 4.6 | Anthropic | `anthropic/claude-sonnet-4-6` |
| Gemini 3 Pro | Google Vertex | `vertex_ai/gemini-3-pro-preview` |
| DeepSeek V4 Pro | DeepSeek | `deepseek/deepseek-v4-pro` |
| Kimi K3 | Moonshot | `moonshot/kimi-k3` |
| Model | Provider | Configuration |
| ----------------- | ------------- | -------------------------------- |
| GPT-5.4 | OpenAI | `openai/gpt-5.4` |
| Claude Sonnet 4.6 | Anthropic | `anthropic/claude-sonnet-4-6` |
| Gemini 3 Pro | Google Vertex | `vertex_ai/gemini-3-pro-preview` |
```bash
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="openai/gpt-5.4"
export LLM_API_KEY="your-api-key"
```
@@ -65,7 +62,6 @@ See the [Local Models guide](/llm-providers/local) for setup instructions and re
Use LiteLLM's `provider/model-name` format:
```
openrouter/z-ai/glm-5.3
openai/gpt-5.4
anthropic/claude-sonnet-4-6
vertex_ai/gemini-3-pro-preview

View File

@@ -28,12 +28,12 @@ description: "Install Strix and run your first security scan"
Set your LLM provider:
```bash
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export STRIX_LLM="openai/gpt-5.4"
export LLM_API_KEY="your-api-key"
```
<Tip>
For best results, use `openrouter/z-ai/glm-5.3` (the default pick), `openai/gpt-5.4`, `anthropic/claude-opus-4-6`, or `openai/gpt-5.2`.
For best results, use `openai/gpt-5.4`, `anthropic/claude-opus-4-6`, or `openai/gpt-5.2`.
</Tip>
## Run Your First Scan

View File

@@ -250,7 +250,6 @@ ignore = [
# Stdlib HTTP handler overrides (do_GET/do_POST) and lazy imports that avoid a
# circular dependency with strix.telemetry / strix.interface.viewer.report_pdf.
"strix/interface/viewer/server.py" = ["N802", "PLC0415"]
"strix/interface/cloud/payment_proxy.py" = ["N802"]
# Lazy telemetry import to avoid importing PostHog before the viewer starts.
"strix/interface/viewer/cli.py" = ["PLC0415"]
# Lazy imports inside functions to avoid circular dependency with
@@ -414,8 +413,6 @@ known_third_party = ["pydantic", "litellm"]
# ============================================================================
[tool.bandit]
# Tests are covered by ruff's flake8-bandit rules (see per-file-ignores above),
# which is where fixture tokens and loopback URL opens are already waived.
exclude_dirs = ["docs", "build", "dist", "tests"]
exclude_dirs = ["docs", "build", "dist"]
skips = ["B101", "B601", "B404", "B603", "B607"] # Skip assert, shell injection, subprocess import and partial path checks
severity = "medium"

View File

@@ -183,7 +183,7 @@ def build_authorize_url(challenge: str, state: str) -> str:
"code_challenge": challenge,
"code_challenge_method": "S256",
"state": state,
"id_token_add_organizations": "true", # nosec B105 - boolean flag, not a secret
"id_token_add_organizations": "true",
"codex_cli_simplified_flow": "true",
"originator": ORIGINATOR,
}

View File

@@ -562,8 +562,6 @@ DEFAULT_MODEL_RETRY = ModelRetrySettings(
)
RECOMMENDED_MODEL_NAMES = (
"zai/glm-5.3",
"zai/glm-5.3-flash",
"openai/gpt-5.6-sol",
"openai/gpt-5.6-terra",
"openai/gpt-5.6-luna",
@@ -572,7 +570,6 @@ RECOMMENDED_MODEL_NAMES = (
"openai/gpt-5.5",
"openai/gpt-5.4",
"openai/gpt-5.3-codex",
"anthropic/claude-fable-5-1",
"anthropic/claude-fable-5",
"anthropic/claude-opus-5",
"anthropic/claude-opus-4-8",
@@ -580,8 +577,6 @@ RECOMMENDED_MODEL_NAMES = (
"anthropic/claude-sonnet-4-6",
"vertex_ai/gemini-3.1-pro-preview",
"gemini/gemini-3.1-pro-preview",
"vertex_ai/gemini-3.7-flash",
"gemini/gemini-3.7-flash",
"gemini/gemini-3.6-flash",
"deepseek/deepseek-v4-pro",
"deepseek/deepseek-v4-flash",
@@ -603,7 +598,6 @@ FRONTIER_MODEL_FAMILIES = (
(("deepseek",), ("deepseek-v4", "deepseek-r1", "deepseek-reasoner")),
(("alibaba", "dashscope", "qwen"), ("qwen3.8", "qwen3.7", "qwen3-max")),
(("moonshot", "moonshotai", "kimi"), ("kimi-k3", "kimi-k2.7", "kimi-k2.6")),
(("zai", "z-ai", "zai-org", "zhipuai"), ("glm-5.3", "glm-5.2")),
)

View File

@@ -428,7 +428,6 @@ async def run_strix_scan(
}
for summary in mcp_registry.summaries()
]
# Feed a non-secret connection roster (name / provider /
# tool_count / dead) to two consumers: once now (all
# currently healthy) and again whenever a connection later

View File

@@ -14,7 +14,7 @@ import sys
from rich.console import Console
from rich.markup import escape
from strix.interface.cloud import http
import strix.interface.cloud.http as http # noqa: PLR0402
from strix.interface.cloud.render import json_mode
from strix.interface.cloud.runner import resolve, run
from strix.interface.cloud.session import run_session

View File

@@ -70,7 +70,7 @@ def validate_environment() -> None:
error_text.append("", style="white")
error_text.append("STRIX_LLM", style="bold cyan")
error_text.append(
" - Model name to use (e.g., 'openrouter/z-ai/glm-5.3' or "
" - Model name to use (e.g., 'openai/gpt-5.4' or "
"'anthropic/claude-opus-4-7')\n",
style="white",
)
@@ -102,7 +102,7 @@ def validate_environment() -> None:
)
error_text.append("\nExample setup:\n", style="white")
error_text.append("export STRIX_LLM='openrouter/z-ai/glm-5.3'\n", style="dim white")
error_text.append("export STRIX_LLM='openai/gpt-5.4'\n", style="dim white")
if missing_optional_vars:
for var in missing_optional_vars:

View File

@@ -391,10 +391,13 @@ def _print_model_connection_error(exc: BaseException, model_name: str) -> None:
def _bootstrap_scan(args: argparse.Namespace) -> None:
"""Warm up the model and prepare the run for a non-interactive scan.
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.
Interactive launches only validate the environment here; the model
preflight and run preparation happen inside the TUI so the interface
paints immediately instead of waiting on a model round trip.
"""
validate_environment()
if not args.non_interactive:
return
try:
asyncio.run(warm_up_llm(show_model_warning=True))
except ModelConnectionError as exc:
@@ -464,9 +467,10 @@ def main() -> None:
check_docker_installed()
pull_docker_image()
validate_environment()
if args.non_interactive:
# In setup mode the TUI collects the target, then runs prepare_run(),
# warm-up, and telemetry itself once the user starts the scan.
if not args.needs_setup:
_bootstrap_scan(args)
from strix.report.state import get_global_report_state
@@ -507,7 +511,6 @@ def main() -> None:
if not args.run_name:
# Setup mode where the user quit before starting a scan: nothing ran.
notify_update(Console())
return
results_path = run_dir_for(args.run_name)

View File

@@ -36,8 +36,7 @@ if TYPE_CHECKING:
_STOPPABLE_AGENT_STATUSES = frozenset({"running", "waiting", "budget_paused"})
ChangeCallback = Callable[[], None]
StartCallback = Callable[[], Awaitable[None]]
VerifyCallback = Callable[[], Awaitable[None]]
StartCallback = Callable[[bool], Awaitable[None]]
QuitCallback = Callable[[], Awaitable[None]]
@@ -52,7 +51,6 @@ class TuiController:
coordinator: Any = None,
report_state: ReportState | None = None,
on_start: StartCallback | None = None,
on_verify: VerifyCallback | None = None,
on_quit: QuitCallback | None = None,
on_change: ChangeCallback | None = None,
) -> None:
@@ -101,6 +99,7 @@ class TuiController:
# A target-less launch enters the live view and asks there before
# anything is prepared; this holds the directory awaiting that answer.
self.pending_workspace_mount: str | None = None
self._pending_verify = True
self.messages: list[dict[str, str]] = []
self._next_message_id = 1
self.error: str | None = None
@@ -113,7 +112,6 @@ class TuiController:
self.viewer_url: str | None = None
self._viewer_httpd: Any = None
self._on_start = on_start
self._on_verify = on_verify
self._on_quit = on_quit
self._on_change = on_change
@@ -330,6 +328,12 @@ class TuiController:
async def _start(self, payload: dict[str, Any]) -> dict[str, Any]:
if self.scan_started or self._start_in_progress:
raise RuntimeError("Scan is already starting or running")
# A bare prompt launches optimistically, like a coding agent: it skips
# the network model preflight and surfaces any model error live. A named
# target keeps the preflight so a real scan does not commit blind.
verify = payload.get("verify", True)
if not isinstance(verify, bool):
raise TypeError("verify must be a boolean")
# Launching with no target mounts the working directory, so it requires
# the user's explicit confirmation rather than happening silently.
mount_working_dir = payload.get("mount_working_dir", False)
@@ -340,44 +344,27 @@ class TuiController:
raise ValueError("No model configured. Set STRIX_LLM first.")
if self._on_start is None:
raise RuntimeError("Scan start is unavailable")
if not self.targets and not mount_working_dir:
raise ValueError("No target set. Add a target first.")
# The model check runs while still on the start screen, for a bare
# prompt as much as for a named target, so a failure lands in the setup
# log where the user can fix it and retry rather than in a dead run.
await self._verify_model()
if not self.targets:
if not mount_working_dir:
raise ValueError("No target set. Add a target first.")
# Mounting the working directory needs the user's confirmation, and
# that is asked in the live view. Enter it now and prepare nothing
# until the answer arrives, so declining leaves no run behind.
self.pending_workspace_mount = str(Path.cwd())
self._pending_verify = verify
self.setup_mode = False
self.scan_started = True
self.scan_state = "preparing"
return {"started": True}
await self._begin_scan()
await self._begin_scan(verify)
return {"started": True}
async def _verify_model(self) -> None:
if self._on_verify is None:
return
self._start_in_progress = True
try:
await self._on_verify()
finally:
self._start_in_progress = False
async def _begin_scan(self) -> None:
async def _begin_scan(self, verify: bool) -> None:
if self._on_start is None:
raise RuntimeError("Scan start is unavailable")
self._start_in_progress = True
try:
await self._on_start()
except Exception as exc:
if not self.setup_mode:
# The live view is already up, so the failure has to show there.
self.fail_preparation(str(exc))
raise
await self._on_start(verify)
finally:
self._start_in_progress = False
self.setup_mode = False
@@ -397,7 +384,7 @@ class TuiController:
# the whole of the input either way; the working directory is only an
# extra the agent may look at, so the run goes ahead without one.
self.workspace_mount = mount if approved else None
await self._begin_scan()
await self._begin_scan(self._pending_verify)
return {"approved": approved}
async def _send_message(self, payload: dict[str, Any]) -> dict[str, Any]:

View File

@@ -45,20 +45,23 @@ func (m *Model) submitSetupPrompt(value string) (tea.Model, tea.Cmd) {
if len(fields) > targets {
commands = append(commands, send(m.client, "setup.set_instruction", map[string]any{"instruction": value}))
}
// The backend verifies the model connection before either kind of launch
// and reports on it through the setup log. A bare prompt mounts the working
// directory - the backend asks about that from the live view, so the prompt
// is held here in case it is declined.
payload := map[string]any{}
if targets == 0 && len(m.snapshot.Targets) == 0 {
// With a target, verify the model connection before the scan commits to it.
// A bare prompt launches optimistically, like a coding agent, and mounts the
// working directory - the backend asks about that from the live view, so the
// prompt is held here in case it is declined.
verify := targets > 0 || len(m.snapshot.Targets) > 0
payload := map[string]any{"verify": verify}
if verify {
m.setupMsg("Verifying model connection...", render.Col(amber))
} else {
m.pendingPrompt = value
payload["mount_working_dir"] = true
}
commands = append(commands, send(m.client, "setup.start", payload))
// Ordered, not batched: setup.start leaves setup mode, so it must be the
// last command to reach the backend. Batched sends race, and if setup.start
// wins the target and instruction commands land after the guard closes and
// fail with a red error.
// last command to reach the backend. Batched sends race, and once the
// preflight is skipped setup.start wins, making the target and instruction
// commands land after the guard closes and fail with a red error.
return *m, tea.Sequence(commands...)
}

View File

@@ -94,6 +94,25 @@ func commandTypes(envelopes []protocol.Envelope) []string {
return types
}
// startVerify returns the verify flag on the setup.start command, and whether
// a setup.start command was present at all.
func startVerify(t *testing.T, envelopes []protocol.Envelope) (verify, found bool) {
t.Helper()
for _, envelope := range envelopes {
if envelope.Type != "setup.start" {
continue
}
var payload struct {
Verify bool `json:"verify"`
}
if err := json.Unmarshal(envelope.Payload, &payload); err != nil {
t.Fatal(err)
}
return payload.Verify, true
}
return false, false
}
func contains(values []string, want string) bool {
for _, value := range values {
if value == want {
@@ -141,6 +160,10 @@ func TestSetupPromptWithoutTargetLaunchesAndRequestsMount(t *testing.T) {
if mount, found := startPayloadFlag(t, envelopes, "mount_working_dir"); !found || !mount {
t.Fatalf("mount was not requested: mount_working_dir=%v found=%v", mount, found)
}
// A bare prompt launches optimistically: no model preflight.
if verify, found := startVerify(t, envelopes); !found || verify {
t.Fatalf("bare prompt should launch with verify=false, got verify=%v found=%v", verify, found)
}
// setup.start leaves setup mode, so it must be the last command sent.
if start, instr := firstIndex(types, "setup.start"), lastIndex(types, "setup.set_instruction"); start < instr {
t.Fatalf("setup.start (%d) must come after setup.set_instruction (%d): %v", start, instr, types)
@@ -250,8 +273,9 @@ func TestSetupPromptWithTargetLaunches(t *testing.T) {
t.Fatalf("missing %s in %v", want, types)
}
}
if _, found := startPayloadFlag(t, envelopes, "mount_working_dir"); found {
t.Fatalf("a targeted prompt must not ask to mount the working directory: %v", types)
// A named target keeps the upfront model check.
if verify, found := startVerify(t, envelopes); !found || !verify {
t.Fatalf("targeted prompt should launch with verify=true, got verify=%v found=%v", verify, found)
}
// The target and instruction must reach the backend before setup.start
// closes the setup guard.

View File

@@ -63,14 +63,11 @@ class GoTuiRuntime:
self.scan_error: BaseException | None = None
self._last_sync_fingerprint = ""
self._error_noted_agents: set[str] = set()
self.model_verified = False
self._setup_preflight: asyncio.Task[None] | None = None
self.controller = TuiController(
args,
live_view=self.live_view,
coordinator=self.coordinator,
on_start=self.start_from_setup,
on_verify=self.ensure_model_verified,
on_quit=self.quit,
)
self.server = TuiBackendServer(self.controller)
@@ -110,51 +107,7 @@ class GoTuiRuntime:
)
self.controller.notify_changed()
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
setup log before the user has finished typing.
"""
if not (load_settings().llm.model or "").strip():
return
try:
await self._preflight_model()
except Exception as exc:
logger.exception("Go TUI setup model preflight failed")
self.controller.add_message(f"Model connection failed: {exc}", "error")
return
self.controller.add_message("Model connection verified")
async def ensure_model_verified(self) -> None:
"""Hold a setup launch until the model has answered once."""
preflight = self._setup_preflight
if preflight is not None and not preflight.done():
await asyncio.shield(preflight)
if self.model_verified:
return
try:
await self._preflight_model()
except Exception as exc:
logger.exception("Go TUI setup model preflight failed")
raise RuntimeError(f"Model connection failed: {exc}") from exc
async def _preflight_model(self) -> None:
model = (load_settings().llm.model or "").strip()
self.controller.add_message("Verifying model connection...")
await preflight_model_connection(model)
self.model_verified = True
def _start_preparation(self) -> asyncio.Task[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())
async def start_from_setup(self) -> None:
async def start_from_setup(self, verify: bool = True) -> None:
candidate = deepcopy(self.args)
candidate.scan_mode = self.controller.scan_mode
candidate.instruction = self.controller.instruction
@@ -171,7 +124,16 @@ class GoTuiRuntime:
if isinstance(target, dict) and target.get("original")
]
targets_changed = self.controller.targets != existing_targets
persist_current()
model = (load_settings().llm.model or "").strip()
# A bare prompt launches optimistically: it skips the network preflight
# and lets any model error surface once the agent starts, like a coding
# agent. A named target keeps the upfront check.
if verify:
try:
await preflight_model_connection(model)
except Exception as exc:
logger.exception("Go TUI setup model preflight failed")
raise RuntimeError(f"Model connection failed: {exc}") from exc
# A confirmed target-less launch mounts the working directory for the
# agent to work in, without making it a scan target.
candidate.workspace_mount = self.controller.workspace_mount
@@ -414,7 +376,9 @@ class GoTuiRuntime:
)
process, backend_socket = await launch_tui_process(command, env, cwd)
await self.server.start(backend_socket)
prepare_task = self._start_preparation()
if not self.controller.setup_mode:
self.controller.begin_preparation()
prepare_task = asyncio.create_task(self.prepare_and_start())
sync_task = asyncio.create_task(self.sync_state())
return_code = await wait_process(process)
check_return_code(return_code)

View File

@@ -167,7 +167,7 @@ def get_global_report_state() -> Optional["ReportState"]:
return _global_report_state
def set_global_report_state(report_state: Optional["ReportState"]) -> None:
def set_global_report_state(report_state: "ReportState") -> None:
global _global_report_state # noqa: PLW0603
_global_report_state = report_state
# New run: drop any streamed-cost entries a prior run left unconsumed.

View File

@@ -24,30 +24,3 @@ def _isolate_mcp_config(
monkeypatch.setenv("STRIX_MCP_CONFIG", str(missing))
monkeypatch.delenv("STRIX_MCP_ONLY", raising=False)
monkeypatch.delenv("STRIX_MCP_EXCLUDE", raising=False)
@pytest.fixture(autouse=True)
def _plain_terminal(monkeypatch: pytest.MonkeyPatch) -> None:
"""Make Rich output identical on every developer's machine.
Many CLI tests force ``isatty()`` to ``True`` to exercise the human-readable
code path and then assert on the plain text. Rich picks its color system
from ``TERM``, ``COLORTERM``, and ``FORCE_COLOR``, so on a real terminal
those assertions would meet ANSI escape codes instead of the words they
look for. A dumb terminal renders the same text without any styling.
"""
monkeypatch.setenv("TERM", "dumb")
for name in ("COLORTERM", "FORCE_COLOR", "NO_COLOR", "TTY_COMPATIBLE"):
monkeypatch.delenv(name, raising=False)
@pytest.fixture(autouse=True)
def _isolate_wallet_config(monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep a developer's real mppx wallet out of the top-up tests.
``strix cloud billing topup`` chooses the Stripe Link flow or the
preconfigured mppx wallet from these variables, so leaving them set would
silently switch which branch a test runs.
"""
for name in ("MPPX_ACCOUNT", "MPPX_STRIPE_SECRET_KEY", "MPPX_STRIPE_PAYMENT_METHOD"):
monkeypatch.delenv(name, raising=False)

View File

@@ -6,7 +6,6 @@ import io
import json
import shutil
import subprocess
import sys
import urllib.request
import webbrowser
from pathlib import Path
@@ -17,7 +16,7 @@ import requests
from rich.console import Console
from strix.interface import cloud, platform_cli
from strix.interface.cloud import http, render, runner, workspaces
from strix.interface.cloud import billing, http, payment_proxy, render, runner, workspaces
from strix.interface.cloud.spec import GROUP_HELP, SPEC
@@ -148,7 +147,7 @@ def test_read_groups_have_safe_defaults(group: str, verb: str) -> None:
resolved = runner.resolve(group, [])
assert resolved is not None
command, remaining = resolved
assert command is SPEC[group][verb]
assert command is runner.SPEC[group][verb]
assert remaining == []
@@ -562,7 +561,7 @@ def test_stored_token_is_never_sent_to_a_different_platform_origin(
lambda: {"api_token": "stored-secret", "app_url": "https://app.strix.ai"},
)
monkeypatch.setattr(
requests,
http.requests,
"request",
lambda *_args, **_kwargs: pytest.fail("a mismatched origin must not receive the token"),
)
@@ -577,7 +576,7 @@ def test_stored_token_requires_an_issuer_binding(monkeypatch: pytest.MonkeyPatch
monkeypatch.setattr(http, "_app_url_override", "https://app.strix.ai")
monkeypatch.setattr(http, "read_record", lambda: {"api_token": "legacy-secret"})
monkeypatch.setattr(
requests,
http.requests,
"request",
lambda *_args, **_kwargs: pytest.fail("an unbound token must not be sent"),
)
@@ -603,7 +602,7 @@ def test_stored_token_is_sent_only_to_its_bound_platform(
seen.update(url=url, headers=kwargs["headers"])
return FakeResponse(payload={"balance": 1})
monkeypatch.setattr(requests, "request", request)
monkeypatch.setattr(http.requests, "request", request)
response = http.request("GET", "/billing/credits")
assert response.status_code == 200
@@ -627,7 +626,7 @@ def test_explicit_token_can_target_an_explicit_platform(
seen.update(url=url, headers=kwargs["headers"])
return FakeResponse(payload={"balance": 1})
monkeypatch.setattr(requests, "request", request)
monkeypatch.setattr(http.requests, "request", request)
override_value = "explicit-preview-" + str(1)
response = http.request("GET", "/billing/credits", token=override_value)
@@ -707,9 +706,9 @@ def test_topup_noninteractive_requires_explicit_payment_approval(
monkeypatch.setattr(
http, "request", lambda *_a, **_k: FakeResponse(status_code=402, payload=challenge)
)
monkeypatch.setattr(sys.stdin, "isatty", lambda: False)
monkeypatch.setattr(runner.sys.stdin, "isatty", lambda: False)
monkeypatch.setattr(
subprocess,
billing.subprocess,
"run",
lambda *_a, **_k: pytest.fail("wallet must not run without --yes"),
)
@@ -733,10 +732,10 @@ def test_topup_machine_output_never_prompts_even_with_terminal_stdin(
monkeypatch.setattr(
http, "request", lambda *_a, **_k: FakeResponse(status_code=402, payload=challenge)
)
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
monkeypatch.setattr(sys.stdout, "isatty", lambda: stdout_tty)
monkeypatch.setattr(runner.sys.stdin, "isatty", lambda: True)
monkeypatch.setattr(runner.sys.stdout, "isatty", lambda: stdout_tty)
monkeypatch.setattr(
Console,
runner.Console,
"input",
lambda *_a, **_k: pytest.fail("machine-readable top-up must not prompt"),
)
@@ -840,7 +839,7 @@ def test_topup_keeps_token_out_of_wallet_process_and_forwards_payment(
upstream.update(method=method, url=url, **kwargs)
return FakeResponse(payload=receipt, content=json.dumps(receipt).encode())
monkeypatch.setattr(requests, "request", fake_upstream_request)
monkeypatch.setattr(payment_proxy.requests, "request", fake_upstream_request)
def fake_run(command: list[str], **kwargs: Any) -> Any:
commands.append(command)
@@ -906,7 +905,6 @@ def test_topup_wallet_failure_is_one_redacted_json_object(
http, "request", lambda *_a, **_k: FakeResponse(status_code=402, payload=challenge)
)
monkeypatch.setattr(http, "api_token", lambda *_a, **_k: "tok")
monkeypatch.setenv("MPPX_ACCOUNT", "agent")
monkeypatch.setattr(shutil, "which", lambda _name: "/usr/bin/npx")
monkeypatch.setattr(
subprocess,
@@ -948,7 +946,6 @@ def test_topup_wallet_interruption_reports_unknown_payment_outcome(
),
)
monkeypatch.setattr(http, "api_token", lambda *_a, **_k: "tok")
monkeypatch.setenv("MPPX_ACCOUNT", "agent")
monkeypatch.setattr(shutil, "which", lambda _name: "/usr/bin/npx")
monkeypatch.setattr(
subprocess, "run", lambda *_a, **_k: (_ for _ in ()).throw(KeyboardInterrupt)
@@ -974,7 +971,6 @@ def test_topup_non_json_wallet_success_requires_balance_verification(
),
)
monkeypatch.setattr(http, "api_token", lambda *_a, **_k: "tok")
monkeypatch.setenv("MPPX_ACCOUNT", "agent")
monkeypatch.setattr(shutil, "which", lambda _name: "/usr/bin/npx")
monkeypatch.setattr(
subprocess,
@@ -1001,7 +997,6 @@ def test_topup_rejects_parseable_wallet_error_as_a_success(
),
)
monkeypatch.setattr(http, "api_token", lambda *_a, **_k: "tok")
monkeypatch.setenv("MPPX_ACCOUNT", "agent")
monkeypatch.setattr(shutil, "which", lambda _name: "/usr/bin/npx")
monkeypatch.setattr(
subprocess,
@@ -1041,7 +1036,6 @@ def test_topup_does_not_trust_an_unobserved_wallet_receipt(
),
)
monkeypatch.setattr(http, "api_token", lambda *_a, **_k: "tok")
monkeypatch.setenv("MPPX_ACCOUNT", "agent")
monkeypatch.setattr(shutil, "which", lambda _name: "/usr/bin/npx")
monkeypatch.setattr(
subprocess,
@@ -1062,7 +1056,7 @@ def test_topup_does_not_trust_an_unobserved_wallet_receipt(
def test_topup_human_mode_requires_a_bridge_confirmed_receipt(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -1072,10 +1066,9 @@ def test_topup_human_mode_requires_a_bridge_confirmed_receipt(
),
)
monkeypatch.setattr(http, "api_token", lambda *_a, **_k: "tok")
monkeypatch.setenv("MPPX_ACCOUNT", "agent")
monkeypatch.setattr(shutil, "which", lambda _name: "/usr/bin/npx")
monkeypatch.setattr(
requests,
payment_proxy.requests,
"request",
lambda *_a, **_k: FakeResponse(status_code=200, content=b"<html>not a receipt</html>"),
)
@@ -1604,7 +1597,7 @@ def test_handoff_links_reject_non_http_schemes(
monkeypatch: pytest.MonkeyPatch,
capsys: Any,
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(runner.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -1889,7 +1882,7 @@ def test_workspace_use_preserves_definitive_conflict(
def test_group_help_lists_all_verbs_instead_of_default_verb_help(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
assert cloud.run_cloud(["workspaces", "-h"]) == 0
output = capsys.readouterr().out
assert "workspaces verbs" in output
@@ -1913,7 +1906,7 @@ def test_workspace_alias_routes_to_workspaces(monkeypatch: pytest.MonkeyPatch) -
def test_workspace_human_list_is_numbered_and_hides_ids(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -1961,7 +1954,7 @@ def test_integrations_human_list_exposes_installation_id_and_json_stays_full(
],
"bitbucket_oauth_enabled": True,
}
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload=payload))
assert cloud.run_cloud(["integrations", "list"]) == 0
@@ -1979,7 +1972,7 @@ def test_integrations_human_list_exposes_installation_id_and_json_stays_full(
def test_pr_review_human_list_prioritizes_actionable_fields(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -2054,7 +2047,7 @@ def test_pr_review_human_list_shows_pull_request_state(
capsys: Any,
pr_state: str,
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -2117,7 +2110,7 @@ def test_scan_human_list_identifies_internal_and_uploaded_targets(
record: dict[str, Any],
expected_targets: tuple[str, ...],
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -2281,7 +2274,7 @@ def test_human_lists_prioritize_actionable_fields(
visible: tuple[str, ...],
hidden: tuple[str, ...],
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload=payload))
assert cloud.run_cloud(command) == 0
@@ -2295,7 +2288,7 @@ def test_human_lists_prioritize_actionable_fields(
def test_token_human_list_shows_lifecycle_status(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -2652,7 +2645,7 @@ def test_nonstandard_human_list_envelopes_are_actionable(
visible: tuple[str, ...],
hidden: tuple[str, ...],
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload=payload))
assert cloud.run_cloud(command) == 0
@@ -2702,7 +2695,7 @@ def test_chat_credentials_human_view_separates_attached_and_available_sources(
}
],
}
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload=payload))
command = ["chat", "credentials", "chat-id", "--scan-ids", "source-scan-id"]
@@ -2928,7 +2921,7 @@ def test_named_human_list_views_match_api_fields(
visible: tuple[str, ...],
hidden: tuple[str, ...],
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload=payload))
assert cloud.run_cloud(command) == 0
@@ -2942,7 +2935,7 @@ def test_named_human_list_views_match_api_fields(
def test_supply_chain_org_summary_human_view_shows_totals_and_repository_risk(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -3103,7 +3096,7 @@ def test_wrapped_detail_human_views_are_unwrapped_and_actionable(
visible: tuple[str, ...],
hidden: tuple[str, ...],
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload=payload))
assert cloud.run_cloud(command) == 0
@@ -3117,7 +3110,7 @@ def test_wrapped_detail_human_views_are_unwrapped_and_actionable(
def test_trace_human_view_summarizes_events_and_preserves_selector(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
seen_query: dict[str, Any] = {}
def fake_trace_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse:
@@ -3213,7 +3206,7 @@ def test_trace_human_view_summarizes_events_and_preserves_selector(
def test_paginated_human_list_shows_total_and_continuation_command(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -3315,7 +3308,7 @@ def test_page_pagination_explains_an_out_of_range_page() -> None:
def test_human_detail_preserves_long_prose_beyond_table_cell_limit(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
description = (
" ".join(["authorization context"] * 12) + " final-description-marker\nsecond-line-marker"
)
@@ -3393,7 +3386,7 @@ def test_large_vulnerability_detail_prioritizes_evidence_and_remediation() -> No
def test_test_user_human_view_joins_latest_verification(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -3490,7 +3483,7 @@ def test_wide_knowledge_table_keeps_title_readable_with_long_identifiers() -> No
def test_human_get_prioritizes_details_and_hides_internal_identity_fields(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",

View File

@@ -6,16 +6,13 @@ import argparse
import io
import json
import sys
import time
from typing import TYPE_CHECKING, Any
import pytest
import requests
from rich.console import Console
from strix.interface import cloud, platform_cli
from strix.interface.cloud import http, render, runner, source_scan
from strix.interface.cloud.source_upload import prepare_source
from strix.interface.main import main as interface_main
@@ -172,7 +169,7 @@ def test_connector_enrollment_command_is_complete_multiline_and_terminal_safe(
" -e LABEL=before\x1b]52;c;copied\x07after \\\n"
" ghcr.io/usestrix/connector:latest"
)
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -296,8 +293,8 @@ def test_source_prompt_shows_paths_and_literal_confirmation(
return "n"
monkeypatch.setattr(console, "input", answer)
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(source_scan.sys.stdin, "isatty", lambda: True)
monkeypatch.setattr(source_scan.sys.stdout, "isatty", lambda: True)
args = argparse.Namespace(
source=str(tmp_path),
dry_run=False,
@@ -327,7 +324,7 @@ def test_source_prompt_interruption_removes_temporary_archive(
(tmp_path / "app.py").write_text("print('ok')\n", encoding="utf-8")
console = Console(file=io.StringIO(), width=100)
archive_paths: list[Path] = []
original_prepare = prepare_source
original_prepare = source_scan.prepare_source
def capture_bundle(*args: Any, **kwargs: Any) -> Any:
bundle = original_prepare(*args, **kwargs)
@@ -339,8 +336,8 @@ def test_source_prompt_interruption_removes_temporary_archive(
monkeypatch.setattr(source_scan, "prepare_source", capture_bundle)
monkeypatch.setattr(console, "input", interrupt)
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(source_scan.sys.stdin, "isatty", lambda: True)
monkeypatch.setattr(source_scan.sys.stdout, "isatty", lambda: True)
args = argparse.Namespace(
source=str(tmp_path),
dry_run=False,
@@ -397,7 +394,7 @@ def test_device_login_rejects_non_http_verification_url(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
requests,
platform_cli.requests,
"post",
lambda *_a, **_k: FakeResponse(
{
@@ -430,7 +427,7 @@ def test_boolean_query_values_are_lowercase_for_url_search_params(
seen["params"] = kwargs.get("params")
return FakeResponse({"items": []})
monkeypatch.setattr(requests, "request", fake_request)
monkeypatch.setattr(http.requests, "request", fake_request)
http.request("GET", "/test", query={"enabled": True, "disabled": False})
assert seen["params"] == {"enabled": "true", "disabled": "false"}
@@ -484,7 +481,7 @@ def test_binary_response_refuses_to_write_to_a_terminal(
monkeypatch: pytest.MonkeyPatch,
capsys: Any,
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(runner.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",
@@ -516,7 +513,7 @@ def test_binary_response_can_be_intentionally_redirected(
return None
redirected = RedirectedStdout()
monkeypatch.setattr(sys, "stdout", redirected)
monkeypatch.setattr(runner.sys, "stdout", redirected)
monkeypatch.setattr(
http,
"request",
@@ -537,7 +534,7 @@ def test_redirected_binary_errors_never_append_diagnostics_to_stdout(
def iter_content(self, *, chunk_size: int) -> Any:
assert chunk_size == 1024 * 1024
yield b"%PDF-partial"
raise requests.ConnectionError("connection lost")
raise http.requests.ConnectionError("connection lost")
response = (
FakeResponse({"detail": "report rejected"}, status_code=500)
@@ -648,7 +645,7 @@ def test_binary_download_streams_and_preserves_existing_file_on_failure(
def iter_content(self, *, chunk_size: int) -> Any:
assert chunk_size == 1024 * 1024
yield b"partial"
raise requests.ConnectionError("connection lost")
raise http.requests.ConnectionError("connection lost")
def close(self) -> None:
self.closed = True
@@ -773,7 +770,7 @@ def test_session_help_is_specific_and_human_whoami_shows_scopes(
"scopes": ["scans:read", "organizations:read"],
}
)
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(platform_cli.sys.stdout, "isatty", lambda: True)
assert cloud.run_cloud(["whoami", "--help"]) == 0
who_help = capsys.readouterr().out
assert "strix cloud whoami" in who_help
@@ -787,7 +784,7 @@ def test_non_tty_whoami_and_logout_emit_json(
) -> None:
monkeypatch.delenv("STRIX_API_TOKEN", raising=False)
monkeypatch.setattr(platform_cli, "AUTH_PATH", tmp_path / "auth.json")
monkeypatch.setattr(sys.stdout, "isatty", lambda: False)
monkeypatch.setattr(platform_cli.sys.stdout, "isatty", lambda: False)
platform_cli.save_record(
{
"api_token": "secret",
@@ -802,7 +799,7 @@ def test_non_tty_whoami_and_logout_emit_json(
assert json.loads(capsys.readouterr().out)["email"] == "agent@example.test"
monkeypatch.setattr(
requests,
platform_cli.requests,
"delete",
lambda *_args, **_kwargs: type("Response", (), {"status_code": 200})(),
)
@@ -834,7 +831,7 @@ def test_scope_picker_labels_match_the_server_presets() -> None:
def test_noninteractive_login_never_prompts_for_workspace(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(sys.stdin, "isatty", lambda: False)
monkeypatch.setattr(platform_cli.sys.stdin, "isatty", lambda: False)
console = Console(file=io.StringIO())
console.input = lambda *_args, **_kwargs: pytest.fail("must not prompt") # type: ignore[method-assign]
@@ -899,10 +896,10 @@ def test_device_flow_slow_down_never_exceeds_the_poll_interval_cap(
sleeps.append(seconds)
now += seconds
monkeypatch.setattr(requests, "post", post)
monkeypatch.setattr(platform_cli.requests, "post", post)
monkeypatch.setattr(platform_cli, "_app_url", lambda: "https://example.test")
monkeypatch.setattr(time, "monotonic", monotonic)
monkeypatch.setattr(time, "sleep", sleep)
monkeypatch.setattr(platform_cli.time, "monotonic", monotonic)
monkeypatch.setattr(platform_cli.time, "sleep", sleep)
with pytest.raises(platform_cli.PlatformAuthError, match="expired"):
platform_cli._run_device_flow(
@@ -936,9 +933,9 @@ def test_device_flow_accepts_external_authkit_url_and_binds_token_origin(
),
]
)
monkeypatch.setattr(requests, "post", lambda *_a, **_k: next(responses))
monkeypatch.setattr(platform_cli.requests, "post", lambda *_a, **_k: next(responses))
monkeypatch.setattr(platform_cli, "_app_url", lambda: "https://preview.strix.ai")
monkeypatch.setattr(time, "sleep", lambda _seconds: None)
monkeypatch.setattr(platform_cli.time, "sleep", lambda _seconds: None)
record = platform_cli._run_device_flow(
Console(file=io.StringIO()),
@@ -987,7 +984,7 @@ def test_root_help_accepts_json_before_help_and_leaf_help_stays_specific(
def test_non_tty_dispatcher_always_emits_structured_json(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: False)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: False)
assert cloud.run_cloud([]) == 0
assert json.loads(capsys.readouterr().out)["command"] == "strix cloud"
@@ -1023,7 +1020,7 @@ def test_source_upload_rejects_untrusted_destinations_before_reading_file(
source.write_bytes(b"approved source")
monkeypatch.setattr(http, "_app_url_override", "https://app.strix.ai")
monkeypatch.setattr(
requests,
http.requests,
"put",
lambda *_args, **_kwargs: pytest.fail("an untrusted URL must not receive source bytes"),
)
@@ -1062,7 +1059,7 @@ def test_source_upload_allows_only_managed_or_same_origin_storage(
return response
monkeypatch.setattr(http, "_app_url_override", app_url)
monkeypatch.setattr(requests, "put", put)
monkeypatch.setattr(http.requests, "put", put)
http.upload_file(signed_url, "upload-token", source)
assert request_options["allow_redirects"] is False
@@ -1083,7 +1080,7 @@ def test_source_upload_refuses_redirects_without_following_them(
return response
monkeypatch.setattr(http, "_app_url_override", "https://app.strix.ai")
monkeypatch.setattr(requests, "put", put)
monkeypatch.setattr(http.requests, "put", put)
with pytest.raises(http.CloudError, match="unexpected redirect"):
http.upload_file(
@@ -1098,7 +1095,7 @@ def test_source_upload_refuses_redirects_without_following_them(
def test_one_time_api_token_has_save_now_warning(
monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True)
monkeypatch.setattr(
http,
"request",

View File

@@ -3,11 +3,9 @@
from __future__ import annotations
import json
import time
from typing import Any
import pytest
import requests
from strix.interface import cloud
from strix.interface.cloud import http, runner
@@ -32,7 +30,7 @@ class FakeResponse:
@pytest.fixture(autouse=True)
def _token(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("STRIX_API_TOKEN", "idempotency-test-token")
monkeypatch.setattr(time, "sleep", lambda _seconds: None)
monkeypatch.setattr(runner.time, "sleep", lambda _seconds: None)
def test_scan_start_generates_and_sends_one_stable_key(
@@ -199,7 +197,7 @@ def test_http_client_places_key_in_the_header(monkeypatch: pytest.MonkeyPatch) -
seen.update(kwargs)
return FakeResponse({"ok": True})
monkeypatch.setattr(requests, "request", request)
monkeypatch.setattr(http.requests, "request", request)
http.request("POST", "/scans", body={}, idempotency_key="header-key")
assert seen["headers"]["Idempotency-Key"] == "header-key"
assert seen["headers"]["Authorization"] == "Bearer idempotency-test-token"

View File

@@ -7,7 +7,6 @@ import urllib.request
from typing import TYPE_CHECKING, Any
import pytest
import requests
from strix.interface.cloud import payment_proxy
@@ -40,8 +39,7 @@ def _post(url: str, body: bytes, headers: dict[str, str] | None = None) -> bytes
method="POST",
)
with urllib.request.urlopen(request, timeout=2) as response: # noqa: S310
body_bytes: bytes = response.read()
return body_bytes
return response.read()
def test_bridge_bounds_decompressed_upstream_response(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -52,7 +50,7 @@ def test_bridge_bounds_decompressed_upstream_response(monkeypatch: pytest.Monkey
return response
monkeypatch.setattr(payment_proxy, "_MAX_UPSTREAM_RESPONSE_BYTES", 4)
monkeypatch.setattr(requests, "request", fake_request)
monkeypatch.setattr(payment_proxy.requests, "request", fake_request)
with payment_proxy.wallet_payment_bridge(
upstream_url="https://app.example.test/api/v1/billing/topup",
@@ -82,7 +80,7 @@ def test_bridge_forwards_only_the_approved_request_and_protected_headers(
captured.append(kwargs)
return _StreamingResponse([b'{"ok":true}'])
monkeypatch.setattr(requests, "request", fake_request)
monkeypatch.setattr(payment_proxy.requests, "request", fake_request)
with payment_proxy.wallet_payment_bridge(
upstream_url="https://app.example.test/api/v1/billing/topup",
api_token="strix-secret", # noqa: S106
@@ -126,7 +124,7 @@ def test_bridge_limits_valid_wallet_attempts(monkeypatch: pytest.MonkeyPatch) ->
calls += 1
return _StreamingResponse([b"{}"])
monkeypatch.setattr(requests, "request", fake_request)
monkeypatch.setattr(payment_proxy.requests, "request", fake_request)
with payment_proxy.wallet_payment_bridge(
upstream_url="https://app.example.test/api/v1/billing/topup",
api_token="strix-secret", # noqa: S106

View File

@@ -6,7 +6,6 @@ import json
from typing import TYPE_CHECKING, Any
import pytest
import requests
from rich.console import Console
from strix.interface import cloud, platform_cli, platform_identity
@@ -56,7 +55,7 @@ def test_http_workspace_pin_is_captured_once(
sent.append(dict(kwargs["headers"]))
return Response({})
monkeypatch.setattr(requests, "request", fake_request)
monkeypatch.setattr(http.requests, "request", fake_request)
http.configure()
platform_cli.save_record(
{
@@ -117,7 +116,7 @@ def test_logout_keeps_local_token_when_remote_outcome_is_not_definitive(
}
)
monkeypatch.setattr(
requests,
platform_cli.requests,
"delete",
lambda *_args, **_kwargs: Response({"detail": "unavailable"}, 503),
)

View File

@@ -12,7 +12,6 @@ would let it escape and surface a traceback on every teardown.
from __future__ import annotations
from types import SimpleNamespace
from typing import TYPE_CHECKING, cast
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -23,10 +22,6 @@ from requests.exceptions import ConnectionError as RequestsConnectionError
from strix.runtime.docker_client import StrixDockerSandboxClient
if TYPE_CHECKING:
from agents.sandbox.session.sandbox_session import SandboxSession
def _client_with_kill_error(exc: Exception) -> StrixDockerSandboxClient:
"""A StrixDockerSandboxClient whose containers.get(...).kill() raises ``exc``."""
client = StrixDockerSandboxClient.__new__(StrixDockerSandboxClient)
@@ -36,10 +31,9 @@ def _client_with_kill_error(exc: Exception) -> StrixDockerSandboxClient:
return client
def _session(container_id: str | None = "abc123") -> SandboxSession:
def _session() -> object:
# delete() reads session._inner.state.container_id
fake = SimpleNamespace(_inner=SimpleNamespace(state=SimpleNamespace(container_id=container_id)))
return cast("SandboxSession", fake)
return SimpleNamespace(_inner=SimpleNamespace(state=SimpleNamespace(container_id="abc123")))
@pytest.mark.parametrize(
@@ -51,7 +45,7 @@ def _session(container_id: str | None = "abc123") -> SandboxSession:
],
)
@pytest.mark.asyncio
async def test_delete_swallows_best_effort_kill_errors(exc: Exception) -> None:
async def test_delete_swallows_best_effort_kill_errors(exc):
"""A torn-down socket (ConnectionError) or a gone/unhappy container
(NotFound/APIError) during the kill must not propagate; delete() still
delegates to the SDK's delete()."""
@@ -68,7 +62,7 @@ async def test_delete_swallows_best_effort_kill_errors(exc: Exception) -> None:
@pytest.mark.asyncio
async def test_delete_does_not_swallow_unrelated_errors() -> None:
async def test_delete_does_not_swallow_unrelated_errors():
"""A programming error (e.g. ValueError) is not part of best-effort kill and
must still propagate."""
client = _client_with_kill_error(ValueError("boom"))
@@ -77,11 +71,11 @@ async def test_delete_does_not_swallow_unrelated_errors() -> None:
@pytest.mark.asyncio
async def test_delete_noop_without_container_id() -> None:
async def test_delete_noop_without_container_id():
"""No container_id -> no kill attempt, just delegate."""
client = StrixDockerSandboxClient.__new__(StrixDockerSandboxClient)
client.docker_client = MagicMock()
session = _session(container_id=None)
session = SimpleNamespace(_inner=SimpleNamespace(state=SimpleNamespace(container_id=None)))
with patch.object(
DockerSandboxClient, "delete", new=AsyncMock(return_value=session)

View File

@@ -343,19 +343,15 @@ async def test_setup_preflights_model_before_starting(
assert candidate.scope_mode == "diff"
assert candidate.diff_base == "origin/main"
monkeypatch.setattr(go_tui, "persist_current", lambda: calls.append("persist"))
monkeypatch.setattr(go_tui, "build_targets_info", build)
monkeypatch.setattr(go_tui, "prepare_run", prepare)
monkeypatch.setattr(go_tui, "telemetry_start", lambda _args: calls.append("telemetry"))
monkeypatch.setattr(runtime, "init_run_state", lambda: calls.append("state"))
monkeypatch.setattr(runtime, "start_scan", lambda: calls.append("scan"))
# The controller runs these two in turn for every setup launch.
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.
assert calls == ["preflight", "persist", "targets", "prepare", "telemetry", "state", "scan"]
assert calls == ["preflight", "targets", "prepare", "telemetry", "state", "scan"]
assert runtime.args.scan_mode == "quick"
assert runtime.args.instruction == ""
assert runtime.args.max_budget_usd == 8.5
@@ -364,138 +360,35 @@ async def test_setup_preflights_model_before_starting(
assert runtime.args.diff_base == "origin/main"
def _setup_model(
monkeypatch: pytest.MonkeyPatch, model: str | None = "openrouter/test-model"
@pytest.mark.asyncio
async def test_optimistic_setup_skips_model_preflight(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime = GoTuiRuntime(args())
runtime.controller.targets = [str(Path.cwd())]
calls: list[str] = []
async def preflight(_model: str) -> None:
calls.append("preflight")
monkeypatch.setattr(
go_tui,
"load_settings",
lambda: SimpleNamespace(llm=SimpleNamespace(model=model)),
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
)
def _setup_messages(runtime: GoTuiRuntime) -> list[tuple[str, str]]:
return [(message["level"], message["text"]) for message in runtime.controller.messages]
@pytest.mark.asyncio
async def test_setup_model_check_reports_success_in_the_setup_log(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime = GoTuiRuntime(args())
calls: list[str] = []
async def preflight(model: str) -> None:
calls.append(model)
_setup_model(monkeypatch)
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
monkeypatch.setattr(go_tui, "build_targets_info", lambda _args, **_kw: calls.append("targets"))
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", lambda: calls.append("state"))
monkeypatch.setattr(runtime, "start_scan", lambda: calls.append("scan"))
await runtime.check_setup_model()
await runtime.start_from_setup(verify=False)
assert calls == ["openrouter/test-model"]
assert runtime.model_verified is True
assert _setup_messages(runtime) == [
("info", "Verifying model connection..."),
("info", "Model connection verified"),
]
@pytest.mark.asyncio
async def test_setup_model_check_reports_failure_without_leaving_setup(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime = GoTuiRuntime(args())
async def preflight(_model: str) -> None:
raise TimeoutError("connection timed out")
_setup_model(monkeypatch)
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
await runtime.check_setup_model()
assert runtime.model_verified is False
assert runtime.controller.setup_mode is True
assert runtime.controller.scan_state == "setup"
assert _setup_messages(runtime)[-1] == (
"error",
"Model connection failed: connection timed out",
)
@pytest.mark.asyncio
async def test_setup_model_check_waits_for_a_configured_model(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime = GoTuiRuntime(args())
_setup_model(monkeypatch, model=None)
monkeypatch.setattr(
go_tui,
"preflight_model_connection",
lambda _model: pytest.fail("nothing to check without a model"),
)
await runtime.check_setup_model()
assert runtime.model_verified is False
assert runtime.controller.messages == []
@pytest.mark.asyncio
async def test_ensure_model_verified_reuses_the_startup_check(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime = GoTuiRuntime(args())
release = asyncio.Event()
calls: list[str] = []
async def preflight(_model: str) -> None:
calls.append("preflight")
await release.wait()
_setup_model(monkeypatch)
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
runtime._setup_preflight = asyncio.create_task(runtime.check_setup_model())
await asyncio.sleep(0)
# A launch that arrives mid-check waits for it rather than racing a second
# round trip.
ensure = asyncio.create_task(runtime.ensure_model_verified())
await asyncio.sleep(0)
assert not ensure.done()
release.set()
await ensure
assert calls == ["preflight"]
assert runtime.model_verified is True
@pytest.mark.asyncio
async def test_ensure_model_verified_retries_after_a_failed_startup_check(
monkeypatch: pytest.MonkeyPatch,
) -> None:
runtime = GoTuiRuntime(args())
outcomes = iter([TimeoutError("connection timed out"), None])
calls: list[str] = []
async def preflight(_model: str) -> None:
calls.append("preflight")
outcome = next(outcomes)
if outcome is not None:
raise outcome
_setup_model(monkeypatch)
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
await runtime.check_setup_model()
assert runtime.model_verified is False
await runtime.ensure_model_verified()
assert calls == ["preflight", "preflight"]
assert runtime.model_verified is True
# No preflight: the scan launches straight through and any model error
# surfaces once the agent runs.
assert "preflight" not in calls
assert calls == ["targets", "prepare", "telemetry", "state", "scan"]
@pytest.mark.asyncio
@@ -507,8 +400,15 @@ async def test_confirmed_target_less_launch_mounts_workspace_without_targets(
runtime.controller.workspace_mount = str(Path.home())
prepared: list[argparse.Namespace] = []
_setup_model(monkeypatch)
monkeypatch.setattr(go_tui, "persist_current", lambda: None)
async def preflight(_model: str) -> None:
return None
monkeypatch.setattr(
go_tui,
"load_settings",
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
)
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
monkeypatch.setattr(
go_tui,
"build_targets_info",
@@ -519,7 +419,7 @@ async def test_confirmed_target_less_launch_mounts_workspace_without_targets(
monkeypatch.setattr(runtime, "init_run_state", lambda: None)
monkeypatch.setattr(runtime, "start_scan", lambda: None)
await runtime.start_from_setup()
await runtime.start_from_setup(verify=False)
assert prepared[0].workspace_mount == str(Path.home())
assert prepared[0].targets_info == []
@@ -542,8 +442,15 @@ async def test_setup_preserves_prepared_cli_targets(
runtime = GoTuiRuntime(runtime_args)
calls: list[str] = []
_setup_model(monkeypatch)
monkeypatch.setattr(go_tui, "persist_current", lambda: calls.append("persist"))
async def preflight(_model: str) -> None:
calls.append("preflight")
monkeypatch.setattr(
go_tui,
"load_settings",
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
)
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
monkeypatch.setattr(
go_tui,
"build_targets_info",
@@ -558,7 +465,7 @@ async def test_setup_preserves_prepared_cli_targets(
assert runtime.controller.targets == ["https://example.com"]
assert runtime.args.targets_info[0]["type"] == "web"
assert calls == ["persist", "prepare", "telemetry", "state", "scan"]
assert calls == ["preflight", "prepare", "telemetry", "state", "scan"]
@pytest.mark.asyncio
@@ -891,17 +798,19 @@ async def test_setup_preflight_failure_does_not_start_scan(
nonlocal started
started = True
_setup_model(monkeypatch)
monkeypatch.setattr(
go_tui,
"load_settings",
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
)
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
monkeypatch.setattr(go_tui, "persist_current", mark_started)
monkeypatch.setattr(go_tui, "build_targets_info", mark_started)
monkeypatch.setattr(runtime, "init_run_state", mark_started)
monkeypatch.setattr(runtime, "start_scan", mark_started)
with pytest.raises(RuntimeError, match="Model connection failed: 401 Unauthorized"):
await runtime.ensure_model_verified()
await runtime.start_from_setup()
assert runtime.model_verified is False
assert started is False
assert runtime.scan_task is None

View File

@@ -72,13 +72,6 @@ def test_recommended_models_are_matched_case_insensitively() -> None:
"moonshot/kimi-k2.6",
"kimi-k2.7-code",
"moonshot/kimi-k3",
"anthropic/claude-fable-5-1",
"vertex_ai/claude-fable-5-1@default",
"gemini/gemini-3.7-flash",
"glm-5.3",
"zai/glm-5.3-flash",
"openrouter/z-ai/glm-5.3",
"novita/zai-org/glm-5.2",
],
)
def test_frontier_model_families_are_accepted(model_name: str) -> None:
@@ -99,9 +92,6 @@ def test_frontier_model_families_are_accepted(model_name: str) -> None:
"openrouter/x-ai/grok-4",
"mistral/mistral-medium-3-5",
"mistral/magistral-medium-latest",
"zai/glm-4.7",
"openrouter/z-ai/glm-5",
"custom-provider/glm-5.3-local",
],
)
def test_non_frontier_models_are_rejected(model_name: str) -> None:

View File

@@ -11,8 +11,7 @@ PYPROJECT = Path(__file__).resolve().parent.parent / "pyproject.toml"
def _optional_dependencies() -> dict[str, list[str]]:
data = tomllib.loads(PYPROJECT.read_text(encoding="utf-8"))
extras: dict[str, list[str]] = data["project"]["optional-dependencies"]
return extras
return data["project"]["optional-dependencies"]
def test_vertex_extra_pins_google_auth() -> None:

View File

@@ -185,4 +185,6 @@ async def test_roster_is_persisted_even_without_a_status_sink(
)
assert persisted, "roster must persist even when no status sink is attached"
assert persisted[-1] == [{"name": "local_fs", "provider": None, "tool_count": 3, "dead": False}]
assert persisted[-1] == [
{"name": "local_fs", "provider": None, "tool_count": 3, "dead": False}
]

View File

@@ -34,8 +34,7 @@ def _finding(**overrides: Any) -> dict[str, Any]:
def _rule_tags(doc: dict[str, Any]) -> list[str]:
tags: list[str] = doc["runs"][0]["tool"]["driver"]["rules"][0]["properties"]["tags"]
return tags
return doc["runs"][0]["tool"]["driver"]["rules"][0]["properties"]["tags"]
def test_stride_tags_on_rule_for_known_cwe() -> None:

View File

@@ -1,7 +1,6 @@
from __future__ import annotations
import asyncio
import os
from pathlib import Path
from typing import Any, cast
@@ -10,40 +9,13 @@ import pytest
from strix.core.sessions import open_agent_session
def _fd_dir() -> Path | None:
def _count_open_fds() -> int | None:
for path in (Path("/proc/self/fd"), Path("/dev/fd")):
if path.is_dir():
return path
return len(list(path.iterdir()))
return None
def _count_open_fds() -> int | None:
fd_dir = _fd_dir()
return None if fd_dir is None else len(list(fd_dir.iterdir()))
def _count_open_fds_to(files: list[Path]) -> int | None:
"""Count the descriptors this process holds on exactly ``files``.
Matching on inode rather than on the process-wide total keeps the check
immune to sockets and pipes that unrelated background threads open while
the test runs.
"""
fd_dir = _fd_dir()
if fd_dir is None:
return None
wanted = {(stat.st_dev, stat.st_ino) for stat in (path.stat() for path in files)}
held = 0
for entry in fd_dir.iterdir():
try:
stat = os.fstat(int(entry.name))
except (OSError, ValueError):
continue
if (stat.st_dev, stat.st_ino) in wanted:
held += 1
return held
@pytest.mark.asyncio
async def test_sessions_hold_no_descriptors_while_parked(tmp_path: Path) -> None:
"""Descriptor use must track live operations, not the number of sessions.
@@ -53,20 +25,21 @@ async def test_sessions_hold_no_descriptors_while_parked(tmp_path: Path) -> None
scan, and fan-out multiplies those handles until the process runs out of file
descriptors (#1018). A session that is not mid-operation should hold none.
"""
if _fd_dir() is None:
baseline = _count_open_fds()
if baseline is None:
pytest.skip("no /proc/self/fd or /dev/fd on this platform")
db_paths = [tmp_path / f"s{i}.db" for i in range(60)]
sessions = [open_agent_session(f"a{i}", path) for i, path in enumerate(db_paths)]
sessions = [open_agent_session(f"a{i}", tmp_path / f"s{i}.db") for i in range(60)]
try:
for _ in range(4):
await asyncio.gather(
*(s.add_items([{"role": "user", "content": "x"}]) for s in sessions)
)
await asyncio.gather(*(s.get_items() for s in sessions))
parked = _count_open_fds_to(db_paths)
# 60 parked sessions, yet none of them holds its database open.
assert parked == 0, f"parked sessions hold {parked} database descriptors"
parked = _count_open_fds()
assert parked is not None
# 60 parked sessions, yet descriptors are back at the baseline.
assert parked - baseline <= 5, f"parked fds grew by {parked - baseline}"
finally:
for s in sessions:
s.close()

View File

@@ -155,7 +155,7 @@ def test_setup_restores_prepared_cli_targets() -> None:
async def test_start_validates_model_before_callback() -> None:
started = False
async def start() -> None:
async def start(_verify: bool = True) -> None:
nonlocal started
started = True
@@ -170,7 +170,7 @@ async def test_start_validates_model_before_callback() -> None:
async def test_start_launches_with_a_configured_model() -> None:
started = False
async def start() -> None:
async def start(_verify: bool = True) -> None:
nonlocal started
started = True
@@ -189,7 +189,7 @@ async def test_start_launches_with_a_configured_model() -> None:
async def test_start_without_target_requires_mount_consent() -> None:
started = False
async def start() -> None:
async def start(_verify: bool = True) -> None:
nonlocal started
started = True
@@ -200,7 +200,7 @@ async def test_start_without_target_requires_mount_consent() -> None:
# Mounting the working directory is never silent.
with pytest.raises(ValueError, match="No target set"):
await controller.handle("setup.start", {})
await controller.handle("setup.start", {"verify": False})
assert started is False
assert controller.targets == []
assert controller.workspace_mount is None
@@ -211,7 +211,7 @@ async def test_target_less_start_enters_live_view_and_waits_for_the_mount() -> N
"""Nothing is prepared until the live-view confirmation is answered."""
started = False
async def start() -> None:
async def start(_verify: bool = True) -> None:
nonlocal started
started = True
@@ -220,7 +220,7 @@ async def test_target_less_start_enters_live_view_and_waits_for_the_mount() -> N
loader._cached = None
controller = TuiController(args(), on_start=start)
result = await controller.handle("setup.start", {"mount_working_dir": True})
result = await controller.handle("setup.start", {"verify": False, "mount_working_dir": True})
assert result == {"started": True}
# The live view is up so the prompt can be shown there, but the scan has not
@@ -236,23 +236,26 @@ async def test_target_less_start_enters_live_view_and_waits_for_the_mount() -> N
@pytest.mark.asyncio
async def test_confirming_the_mount_starts_the_scan_without_a_target() -> None:
started = False
seen_verify: bool | None = None
async def start() -> None:
nonlocal started
async def start(verify: bool = True) -> None:
nonlocal started, seen_verify
started = True
seen_verify = verify
os.environ["STRIX_LLM"] = "anthropic/claude-sonnet-4"
os.environ["ANTHROPIC_API_KEY"] = "test-key"
loader._cached = None
controller = TuiController(args(), on_start=start)
await controller.handle("setup.start", {"mount_working_dir": True})
await controller.handle("setup.start", {"verify": False, "mount_working_dir": True})
result = await controller.handle("setup.confirm_mount", {"approved": True})
assert result == {"approved": True}
assert started is True
# Mounted as a workspace: the scan genuinely has no target, so the
# instruction is the only source of truth.
# Launched optimistically, and mounted as a workspace: the scan genuinely
# has no target, so the instruction is the only source of truth.
assert seen_verify is False
assert controller.workspace_mount == str(Path.cwd())
assert controller.targets == []
assert controller.scan_state == "running"
@@ -261,23 +264,22 @@ async def test_confirming_the_mount_starts_the_scan_without_a_target() -> None:
@pytest.mark.asyncio
async def test_declining_the_mount_runs_without_one() -> None:
started = 0
started: list[bool] = []
async def start() -> None:
nonlocal started
started += 1
async def start(verify: bool = True) -> None:
started.append(verify)
os.environ["STRIX_LLM"] = "anthropic/claude-sonnet-4"
os.environ["ANTHROPIC_API_KEY"] = "test-key"
loader._cached = None
controller = TuiController(args(), on_start=start)
await controller.handle("setup.start", {"mount_working_dir": True})
await controller.handle("setup.start", {"verify": False, "mount_working_dir": True})
result = await controller.handle("setup.confirm_mount", {"approved": False})
assert result == {"approved": False}
# Declining skips the directory; it does not abandon the scan.
assert started == 1
assert started == [False]
assert controller.workspace_mount is None
assert controller.pending_workspace_mount is None
assert controller.setup_mode is False
@@ -287,22 +289,21 @@ async def test_declining_the_mount_runs_without_one() -> None:
@pytest.mark.asyncio
async def test_approving_the_mount_runs_with_it() -> None:
started = 0
started: list[bool] = []
async def start() -> None:
nonlocal started
started += 1
async def start(verify: bool = True) -> None:
started.append(verify)
os.environ["STRIX_LLM"] = "anthropic/claude-sonnet-4"
os.environ["ANTHROPIC_API_KEY"] = "test-key"
loader._cached = None
controller = TuiController(args(), on_start=start)
await controller.handle("setup.start", {"mount_working_dir": True})
await controller.handle("setup.start", {"verify": False, "mount_working_dir": True})
result = await controller.handle("setup.confirm_mount", {"approved": True})
assert result == {"approved": True}
assert started == 1
assert started == [False]
assert controller.workspace_mount == str(Path.cwd())
assert controller.scan_state == "running"
@@ -351,91 +352,23 @@ async def test_user_message_updates_live_agent_projection_immediately() -> None:
@pytest.mark.asyncio
async def test_start_verifies_the_model_before_a_targeted_launch() -> None:
order: list[str] = []
async def test_start_forwards_verify_flag_by_default() -> None:
seen_verify: bool | None = None
async def verify() -> None:
order.append("verify")
async def start() -> None:
order.append("start")
os.environ["STRIX_LLM"] = "anthropic/claude-sonnet-4"
os.environ["ANTHROPIC_API_KEY"] = "test-key"
loader._cached = None
controller = TuiController(args(), on_start=start, on_verify=verify)
await controller.handle("setup.add_target", {"target": "https://example.com"})
await controller.handle("setup.start", {})
assert order == ["verify", "start"]
@pytest.mark.asyncio
async def test_start_verifies_the_model_before_a_bare_prompt_leaves_setup() -> None:
"""A bare prompt gets the same model check as a named target, while the
setup log is still on screen to show the outcome."""
verified = 0
async def verify() -> None:
nonlocal verified
verified += 1
async def start() -> None:
return None
os.environ["STRIX_LLM"] = "anthropic/claude-sonnet-4"
os.environ["ANTHROPIC_API_KEY"] = "test-key"
loader._cached = None
controller = TuiController(args(), on_start=start, on_verify=verify)
await controller.handle("setup.start", {"mount_working_dir": True})
assert verified == 1
assert controller.setup_mode is False
assert controller.pending_workspace_mount == str(Path.cwd())
@pytest.mark.asyncio
async def test_failed_model_check_keeps_the_start_screen() -> None:
async def verify() -> None:
raise RuntimeError("Model connection failed: timed out")
async def start() -> None:
pytest.fail("the scan must not start when the model check fails")
os.environ["STRIX_LLM"] = "anthropic/claude-sonnet-4"
os.environ["ANTHROPIC_API_KEY"] = "test-key"
loader._cached = None
controller = TuiController(args(), on_start=start, on_verify=verify)
with pytest.raises(RuntimeError, match="Model connection failed"):
await controller.handle("setup.start", {"mount_working_dir": True})
# Still on the start screen, so the error lands in the setup log and the
# user can retry; no run was prepared behind a stuck live view.
assert controller.setup_mode is True
assert controller.scan_started is False
assert controller.scan_state == "setup"
assert controller.pending_workspace_mount is None
@pytest.mark.asyncio
async def test_confirmed_mount_launch_failure_is_reported_in_the_live_view() -> None:
async def start() -> None:
raise ValueError("Scan preparation failed")
async def start(verify: bool = True) -> None:
nonlocal seen_verify
seen_verify = verify
os.environ["STRIX_LLM"] = "anthropic/claude-sonnet-4"
os.environ["ANTHROPIC_API_KEY"] = "test-key"
loader._cached = None
controller = TuiController(args(), on_start=start)
await controller.handle("setup.start", {"mount_working_dir": True})
await controller.handle("setup.add_target", {"target": "https://example.com"})
with pytest.raises(ValueError, match="Scan preparation failed"):
await controller.handle("setup.confirm_mount", {"approved": True})
# A named target keeps the upfront model check.
await controller.handle("setup.start", {})
assert controller.scan_state == "failed"
assert controller.error == "Scan preparation failed"
assert seen_verify is True
@pytest.mark.asyncio
@@ -443,7 +376,7 @@ async def test_start_rejects_concurrent_and_repeated_submissions() -> None:
entered = asyncio.Event()
release = asyncio.Event()
async def start() -> None:
async def start(_verify: bool = True) -> None:
entered.set()
await release.wait()