9 Commits

Author SHA1 Message Date
bearsyankees 639ce5b31a fix(report): correct csv_path indentation in write_vulnerabilities
Line 72 was over-indented, causing an IndentationError on import of
strix/report/writer.py and breaking main. Also bump the mirrors-mypy
pre-commit hook to v1.17.1 to avoid the mypy 1.16.0 internal crash
(python/mypy#19412) on openai/_client.py.

Co-Authored-By: Alex Schapiro <bearsyankees@gmail.com>
2026-07-02 19:20:47 +00:00
ASTITVA BHARDWAJ 5ee34481fe Fix non-atomic CSV and MD writes to prevent corruption on crash (#628) (#631) 2026-07-02 07:53:30 -07:00
Rome Thorstenson f342808d2b test: add unit tests for config loader (strix/config/loader.py) (#596)
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-30 04:31:29 -07:00
Dominic White f554523378 Remove collection of unhandled exception error messages from telemetry (#585) 2026-06-29 19:22:06 -07:00
Ahmed Allam 69e82f0258 chore(deps): refresh uv.lock to latest compatible versions (#606) 2026-06-29 19:10:18 -07:00
Ahmed Allam 52ca641679 Update readme (#607) 2026-06-29 19:09:58 -07:00
Ahmed Allam 60d68d85f3 Readme update 2026-06-29 19:00:46 -07:00
Rome Thorstenson 777005a42b fix: stop gracefully with resume hint on persistent RateLimitError (#261) (#593) 2026-06-29 07:31:54 -07:00
Rome Thorstenson 8cdf0683a3 fix(core): collapse child agent initial input into a single user message (#589) 2026-06-29 06:51:47 -07:00
12 changed files with 1545 additions and 1053 deletions
+1 -1
View File
@@ -11,7 +11,7 @@ repos:
# MyPy for static type checking
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.16.0
rev: v1.17.1
hooks:
- id: mypy
additional_dependencies: [
+44 -43
View File
@@ -8,7 +8,7 @@
# Strix
### Open-source AI hackers to find and fix your apps vulnerabilities.
### The open-source AI pentesting tool. Autonomous AI hackers that find and fix your apps vulnerabilities.
<br/>
@@ -40,15 +40,15 @@
## Strix Overview
Strix are autonomous AI agents that act just like real hackers - they run your code dynamically, find vulnerabilities, and validate them through actual proof-of-concepts. Built for developers and security teams who need fast, accurate security testing without the overhead of manual pentesting or the false positives of static analysis tools.
Strix are autonomous AI penetration testing agents that act just like real hackers - they run your code dynamically, find vulnerabilities, and validate them through actual proof-of-concepts. Built for developers and security teams who need fast, accurate security testing without the overhead of manual pentesting or the false positives of static analysis tools.
**Key Capabilities:**
- **Full hacker toolkit** out of the box
- **Teams of agents** that collaborate and scale
- **Real validation** with PoCs, not false positives
- **Developerfirst** CLI with actionable reports
- **Autofix & reporting** to accelerate remediation
- **Full pentesting toolkit** - reconnaissance, exploitation, and validation out of the box
- **Multi-agent orchestration** - teams of AI pentesters that collaborate and scale
- **Real exploit validation** - working PoCs, not false positives like legacy vulnerability scanners
- **Developerfirst CLI** - actionable findings with remediation guidance
- **Autofix & reporting** - generate patches and compliance-ready pentest reports
<br>
@@ -95,13 +95,13 @@ strix --target ./app-directory
## ☁️ Strix Platform
Try the Strix full-stack security platform at **[app.strix.ai](https://app.strix.ai)** sign up for free, connect your repos and domains, and launch a pentest in minutes.
Try the Strix full-stack penetration testing platform at **[app.strix.ai](https://app.strix.ai)** - sign up for free, connect your repos and domains, and launch a pentest in minutes.
- **Validated findings with PoCs** and reproduction steps
- **One-click autofix** as ready-to-merge pull requests
- **Continuous monitoring** across code, cloud, and infrastructure
- **Integrations** with GitHub, Slack, Jira, Linear, and CI/CD pipelines
- **Continuous learning** that builds on past findings and remediations
- **Validated findings with PoCs** - every vulnerability includes a working proof-of-concept exploit and reproduction steps
- **One-click autofix** - AI-generated security patches as ready-to-merge pull requests
- **Continuous pentesting** - always-on vulnerability scanning that keeps pace with your deployments
- **DevSecOps integrations** - GitHub, GitLab, Bitbucket, Slack, Jira, Linear, and CI/CD pipelines
- **Continuous learning** - AI that builds on past findings, adapts to your codebase, and reduces false positives over time
[**Start your first pentest →**](https://app.strix.ai)
@@ -109,37 +109,38 @@ Try the Strix full-stack security platform at **[app.strix.ai](https://app.strix
## ✨ Features
### Agentic Security Tools
### Agentic Pentesting Tools
Strix agents come equipped with a comprehensive security testing toolkit:
Strix agents come equipped with a comprehensive offensive security toolkit - the same tools used by professional penetration testers and ethical hackers:
- **Full HTTP Proxy** - Full request/response manipulation and analysis
- **Browser Automation** - Multi-tab browser for testing of XSS, CSRF, auth flows
- **Terminal Environments** - Interactive shells for command execution and testing
- **Python Runtime** - Custom exploit development and validation
- **Reconnaissance** - Automated OSINT and attack surface mapping
- **Code Analysis** - Static and dynamic analysis capabilities
- **Knowledge Management** - Structured findings and attack documentation
- **HTTP Interception Proxy** - Full request/response manipulation and analysis with Caido
- **Browser Exploitation** - Automated browser for testing XSS, CSRF, clickjacking, and auth bypass flows
- **Shell & Command Execution** - Interactive terminal for exploit development and post-exploitation
- **Custom Exploit Runtime** - Python sandbox for writing and validating proof-of-concept exploits
- **Reconnaissance & OSINT** - Automated attack surface mapping, subdomain enumeration, and fingerprinting
- **Static & Dynamic Code Analysis** - SAST + DAST capabilities for comprehensive application security testing
- **Vulnerability Knowledge Base** - Structured findings with CVSS scoring and OWASP classification
### Comprehensive Vulnerability Detection
### Comprehensive Vulnerability Scanner
Strix can identify and validate a wide range of security vulnerabilities:
Strix identifies, validates, and exploits a wide range of security vulnerabilities across the OWASP Top 10 and beyond:
- **Access Control** - IDOR, privilege escalation, auth bypass
- **Injection Attacks** - SQL, NoSQL, command injection
- **Server-Side** - SSRF, XXE, deserialization flaws
- **Client-Side** - XSS, prototype pollution, DOM vulnerabilities
- **Business Logic** - Race conditions, workflow manipulation
- **Authentication** - JWT vulnerabilities, session management
- **Infrastructure** - Misconfigurations, exposed services
- **Broken Access Control** - IDOR, privilege escalation, auth bypass
- **Injection Attacks** - SQL injection, NoSQL injection, OS command injection, SSTI
- **Server-Side Vulnerabilities** - SSRF, XXE, insecure deserialization, RCE
- **Client-Side Attacks** - XSS (stored/reflected/DOM), prototype pollution, CSRF
- **Business Logic Flaws** - Race conditions, payment manipulation, workflow bypass
- **Authentication & Session** - JWT attacks, session fixation, credential stuffing vectors
- **Infrastructure & Cloud** - Misconfigurations, exposed services, cloud security issues
- **API Security** - Broken authentication, mass assignment, rate limiting bypass
### Graph of Agents
### Graph of Agents (Multi-Agent Pentesting)
Advanced multi-agent orchestration for comprehensive security testing:
Advanced multi-agent orchestration for comprehensive automated penetration testing:
- **Distributed Workflows** - Specialized agents for different attacks and assets
- **Scalable Testing** - Parallel execution for fast comprehensive coverage
- **Dynamic Coordination** - Agents collaborate and share discoveries
- **Distributed Pentesting** - Specialized AI agents for recon, exploitation, and post-exploitation
- **Scalable Security Testing** - Parallel execution across multiple targets for fast, comprehensive coverage
- **Dynamic Coordination** - Agents share discoveries, chain vulnerabilities, and collaborate like a red team
---
@@ -182,7 +183,7 @@ strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main
### Headless Mode
Run Strix programmatically without interactive UI using the `-n/--non-interactive` flagperfect for servers and automated jobs. The CLI prints real-time vulnerability findings, and the final report before exiting. Exits with non-zero code when vulnerabilities are found.
Run Strix programmatically without interactive UI using the `-n/--non-interactive` flag - perfect for servers and automated jobs. The CLI prints real-time vulnerability findings, and the final report before exiting. Exits with non-zero code when vulnerabilities are found.
```bash
strix -n --target https://your-app.com
@@ -239,19 +240,19 @@ export STRIX_REASONING_EFFORT="high" # control thinking effort (default: high,
**Recommended models for best results:**
- [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`
- [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`
See the [LLM Providers documentation](https://docs.strix.ai/llm-providers/overview) for all supported providers including Vertex AI, Bedrock, Azure, and local models.
## Enterprise
## Enterprise Pentesting
Get the same Strix experience with [enterprise-grade](https://strix.ai/demo) controls: SSO (SAML/OIDC), custom compliance reports, dedicated support & SLA, custom deployment options (VPC/self-hosted), BYOK model support, and tailored agents optimized for your environment. [Learn more](https://strix.ai/demo).
Get the same Strix experience with [enterprise-grade](https://strix.ai/demo) controls: SSO (SAML/OIDC), custom compliance-ready penetration testing reports (SOC 2, ISO 27001, PCI DSS), dedicated support & SLA, custom deployment options (VPC/self-hosted), BYOK model support, and tailored AI pentesting agents optimized for your environment. [Learn more](https://strix.ai/demo).
## Documentation
Full documentation is available at **[docs.strix.ai](https://docs.strix.ai)** including detailed guides for usage, CI/CD integrations, skills, and advanced configuration.
Full documentation is available at **[docs.strix.ai](https://docs.strix.ai)** - including detailed guides for usage, CI/CD integrations, skills, and advanced configuration.
## Contributing
+20 -23
View File
@@ -136,30 +136,27 @@ def child_initial_input(
task: str,
parent_history: list[Any],
) -> list[dict[str, Any]]:
initial_input: list[dict[str, Any]] = []
"""Build the initial input for a child agent as a single user message.
Collapsing the inherited-context block, the identity line, and the task into
one ``{"role": "user"}`` message keeps providers that require strictly
alternating roles (e.g. Perplexity, llama.cpp) from rejecting consecutive
user messages.
"""
parts: list[str] = []
if parent_history:
rendered = json.dumps(parent_history, ensure_ascii=False, default=str)
initial_input.append(
{
"role": "user",
"content": (
"== Inherited context from parent (background only) ==\n"
f"{rendered}\n"
"== End of inherited context ==\n"
"Use the above as background only; do not continue the "
"parent's work. Your task follows."
),
},
parts.append(
"== Inherited context from parent (background only) ==\n"
f"{rendered}\n"
"== End of inherited context ==\n"
"Use the above as background only; do not continue the "
"parent's work. Your task follows.",
)
initial_input.append(
{
"role": "user",
"content": (
f"You are agent {name} ({child_id}); your parent is {parent_id}. "
"Maintain your own identity. Call agent_finish when your task "
"is complete."
),
}
parts.append(
f"You are agent {name} ({child_id}); your parent is {parent_id}. "
"Maintain your own identity. Call agent_finish when your task "
"is complete.",
)
initial_input.append({"role": "user", "content": task})
return initial_input
parts.append(task)
return [{"role": "user", "content": "\n\n".join(parts)}]
+14
View File
@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any
from agents import RunConfig
from agents.sandbox import SandboxRunConfig
from openai import RateLimitError
from strix.agents.factory import build_strix_agent, make_child_factory
from strix.config import load_settings
@@ -308,6 +309,19 @@ async def run_strix_scan(
with contextlib.suppress(Exception):
await coordinator.set_status(root_id, "stopped")
return None
except RateLimitError as exc:
logger.warning(
"Scan %s stopped: persistent rate limit from the LLM provider (%s). "
"Resume with 'strix --resume %s' once the limit clears.",
scan_id,
exc,
scan_id,
)
if root_id is not None:
await coordinator.cancel_descendants(root_id)
with contextlib.suppress(Exception):
await coordinator.set_status(root_id, "stopped")
return None
except BaseException:
logger.exception("Strix scan %s failed", scan_id)
if root_id is not None:
+3 -3
View File
@@ -820,10 +820,10 @@ def main() -> None:
asyncio.run(run_tui(args))
except KeyboardInterrupt:
exit_reason = "interrupted"
except Exception as e:
except Exception:
exit_reason = "error"
posthog.error("unhandled_exception", str(e))
scarf.error("unhandled_exception", str(e))
posthog.error("unhandled_exception")
scarf.error("unhandled_exception")
raise
finally:
report_state = get_global_report_state()
+18 -16
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import csv
import io
import json
import logging
import tempfile
@@ -58,9 +59,9 @@ def write_vulnerabilities(
new_reports = [r for r in vulnerability_reports if r["id"] not in saved_vuln_ids]
for report in new_reports:
(vuln_dir / f"{report['id']}.md").write_text(
_atomic_write_text(
vuln_dir / f"{report['id']}.md",
render_vulnerability_md(report),
encoding="utf-8",
)
saved_vuln_ids.add(report["id"])
@@ -69,20 +70,21 @@ def write_vulnerabilities(
key=lambda r: (_SEVERITY_ORDER.get(r["severity"], 5), r["timestamp"]),
)
csv_path = run_dir / "vulnerabilities.csv"
with csv_path.open("w", encoding="utf-8", newline="") as f:
fieldnames = ["id", "title", "severity", "timestamp", "file"]
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for report in sorted_reports:
writer.writerow(
{
"id": report["id"],
"title": report["title"],
"severity": report["severity"].upper(),
"timestamp": report["timestamp"],
"file": f"vulnerabilities/{report['id']}.md",
},
)
csv_buf = io.StringIO()
fieldnames = ["id", "title", "severity", "timestamp", "file"]
csv_writer = csv.DictWriter(csv_buf, fieldnames=fieldnames, lineterminator="\r\n")
csv_writer.writeheader()
for report in sorted_reports:
csv_writer.writerow(
{
"id": report["id"],
"title": report["title"],
"severity": report["severity"].upper(),
"timestamp": report["timestamp"],
"file": f"vulnerabilities/{report['id']}.md",
},
)
_atomic_write_text(csv_path, csv_buf.getvalue())
_atomic_write_text(
run_dir / "vulnerabilities.json",
+1 -3
View File
@@ -123,8 +123,6 @@ def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
)
def error(error_type: str, error_msg: str | None = None) -> None:
def error(error_type: str) -> None:
props = {**base_props(), "error_type": error_type}
if error_msg:
props["error_msg"] = error_msg
_send("error", props)
+1 -3
View File
@@ -129,12 +129,10 @@ def end(report_state: ReportState, exit_reason: str = "completed") -> None:
)
def error(error_type: str, error_msg: str | None = None) -> None:
def error(error_type: str) -> None:
props: dict[str, Any] = {
**base_props(),
"session": SESSION_ID,
"error_type": error_type,
}
if error_msg:
props["error_msg"] = error_msg
_send("error", props)
+178
View File
@@ -0,0 +1,178 @@
"""Tests for strix.config.loader: JSON overrides, alias resolution, persistence."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
import pytest
from pydantic import AliasChoices, Field
from pydantic.fields import FieldInfo
from strix.config import loader
if TYPE_CHECKING:
from pathlib import Path
_LLM_ENV_KEYS = [
"STRIX_LLM",
"LLM_API_KEY",
"OPENAI_API_KEY",
"LLM_API_BASE",
"OPENAI_API_BASE",
"OPENAI_BASE_URL",
"LITELLM_BASE_URL",
"OLLAMA_API_BASE",
"STRIX_REASONING_EFFORT",
"LLM_TIMEOUT",
"PERPLEXITY_API_KEY",
# RuntimeSettings
"STRIX_IMAGE",
"STRIX_RUNTIME_BACKEND",
"STRIX_MAX_LOCAL_COPY_MB",
# TelemetrySettings
"STRIX_TELEMETRY",
]
@pytest.fixture(autouse=True)
def _reset_loader_state(monkeypatch: pytest.MonkeyPatch) -> None:
"""Reset module globals and clear known env vars for deterministic runs."""
for key in _LLM_ENV_KEYS:
monkeypatch.delenv(key, raising=False)
monkeypatch.setattr(loader, "_cached", None)
monkeypatch.setattr(loader, "_override", None)
# --------------------------------------------------------------------------- #
# _read_json_overrides
# --------------------------------------------------------------------------- #
def test_read_json_overrides_missing_file(tmp_path: Path) -> None:
assert loader._read_json_overrides(tmp_path / "nope.json") == {}
def test_read_json_overrides_corrupt_json(tmp_path: Path) -> None:
path = tmp_path / "cli-config.json"
path.write_text("{not valid json", encoding="utf-8")
assert loader._read_json_overrides(path) == {}
def test_read_json_overrides_non_dict_env(tmp_path: Path) -> None:
path = tmp_path / "cli-config.json"
path.write_text(json.dumps({"env": ["not", "a", "dict"]}), encoding="utf-8")
assert loader._read_json_overrides(path) == {}
def test_read_json_overrides_maps_to_nested_settings(tmp_path: Path) -> None:
path = tmp_path / "cli-config.json"
path.write_text(
json.dumps({"env": {"STRIX_LLM": "my-model", "PERPLEXITY_API_KEY": "pk"}}),
encoding="utf-8",
)
assert loader._read_json_overrides(path) == {
"llm": {"model": "my-model"},
"integrations": {"perplexity_api_key": "pk"},
}
def test_read_json_overrides_skips_keys_already_in_environ(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("STRIX_LLM", "from-env")
path = tmp_path / "cli-config.json"
path.write_text(json.dumps({"env": {"STRIX_LLM": "from-file"}}), encoding="utf-8")
# env wins -> the JSON value is not surfaced as an init kwarg.
assert loader._read_json_overrides(path) == {}
# --------------------------------------------------------------------------- #
# _aliases_for
# --------------------------------------------------------------------------- #
def test_aliases_for_simple_alias() -> None:
finfo = FieldInfo(alias="SIMPLE_ALIAS")
assert loader._aliases_for(finfo) == ["SIMPLE_ALIAS"]
def test_aliases_for_alias_choices() -> None:
finfo: FieldInfo = Field( # type: ignore[assignment]
default=None,
validation_alias=AliasChoices("FIRST", "SECOND"),
)
assert loader._aliases_for(finfo) == ["FIRST", "SECOND"]
def test_aliases_for_string_validation_alias() -> None:
finfo: FieldInfo = Field(default=None, validation_alias="STR_ALIAS") # type: ignore[assignment]
assert loader._aliases_for(finfo) == ["STR_ALIAS"]
def test_aliases_for_no_alias() -> None:
assert loader._aliases_for(FieldInfo()) == []
# --------------------------------------------------------------------------- #
# apply_config_override + load_settings round-trip
# --------------------------------------------------------------------------- #
def test_apply_override_and_load_settings_round_trip(tmp_path: Path) -> None:
path = tmp_path / "cli-config.json"
path.write_text(
json.dumps({"env": {"STRIX_LLM": "round-trip-model", "PERPLEXITY_API_KEY": "pk"}}),
encoding="utf-8",
)
loader.apply_config_override(path)
settings = loader.load_settings()
assert settings.llm.model == "round-trip-model"
assert settings.integrations.perplexity_api_key == "pk"
# Second call is memoized -> same object.
assert loader.load_settings() is settings
def test_apply_config_override_invalidates_cache(tmp_path: Path) -> None:
first = tmp_path / "first.json"
first.write_text(json.dumps({"env": {"STRIX_LLM": "first-model"}}), encoding="utf-8")
second = tmp_path / "second.json"
second.write_text(json.dumps({"env": {"STRIX_LLM": "second-model"}}), encoding="utf-8")
loader.apply_config_override(first)
assert loader.load_settings().llm.model == "first-model"
loader.apply_config_override(second)
assert loader.load_settings().llm.model == "second-model"
# --------------------------------------------------------------------------- #
# persist_current
# --------------------------------------------------------------------------- #
def test_persist_current_writes_env_block(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("STRIX_LLM", "persisted-model")
target = tmp_path / "sub" / "cli-config.json"
loader.apply_config_override(target)
loader.persist_current()
assert target.exists()
assert json.loads(target.read_text(encoding="utf-8")) == {
"env": {"STRIX_LLM": "persisted-model"}
}
def test_persist_current_sets_0600_mode(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("STRIX_LLM", "persisted-model")
target = tmp_path / "cli-config.json"
loader.apply_config_override(target)
loader.persist_current()
assert target.stat().st_mode & 0o777 == 0o600
+114
View File
@@ -0,0 +1,114 @@
"""Tests for pure input builders in strix.core.inputs."""
from __future__ import annotations
from itertools import pairwise
from typing import Any
import pytest
from strix.core.inputs import build_root_task, child_initial_input
def _child_kwargs(parent_history: list[Any]) -> dict[str, Any]:
return {
"name": "scout",
"child_id": "agent-2",
"parent_id": "agent-1",
"task": "Audit the login flow.",
"parent_history": parent_history,
}
def test_child_initial_input_single_message_without_history() -> None:
result = child_initial_input(**_child_kwargs([]))
assert len(result) == 1
assert result[0]["role"] == "user"
content = result[0]["content"]
assert "agent scout (agent-2)" in content
assert "Audit the login flow." in content
assert "Inherited context" not in content
def test_child_initial_input_single_message_with_history() -> None:
history = [{"role": "assistant", "content": "previous work"}]
result = child_initial_input(**_child_kwargs(history))
assert len(result) == 1
assert result[0]["role"] == "user"
content = result[0]["content"]
assert "Inherited context from parent" in content
assert "previous work" in content
assert "agent scout (agent-2)" in content
assert "Audit the login flow." in content
@pytest.mark.parametrize(
"parent_history",
[[], [{"role": "assistant", "content": "previous work"}]],
)
def test_child_initial_input_no_consecutive_same_role(parent_history: list[Any]) -> None:
result = child_initial_input(**_child_kwargs(parent_history))
roles = [msg["role"] for msg in result]
assert all(prev != nxt for prev, nxt in pairwise(roles))
def test_build_root_task_empty_config() -> None:
assert build_root_task({}) == ""
def test_build_root_task_repository_target() -> None:
config = {
"targets": [
{
"type": "repository",
"details": {
"target_repo": "https://example.com/repo.git",
"cloned_repo_path": "/workspace/repo",
"workspace_subdir": "repo",
},
},
],
}
task = build_root_task(config)
assert "Repositories:" in task
assert "/workspace/repo" in task
assert "https://example.com/repo.git" in task
def test_build_root_task_web_application_with_instructions() -> None:
config = {
"targets": [
{"type": "web_application", "details": {"target_url": "https://app.example.com"}},
],
"user_instructions": "Focus on auth.",
}
task = build_root_task(config)
assert "URLs:" in task
assert "https://app.example.com" in task
assert "Special instructions: Focus on auth." in task
def test_build_root_task_diff_scope() -> None:
config = {
"targets": [],
"diff_scope": {
"active": True,
"repos": [
{
"workspace_subdir": "repo",
"analyzable_files_count": 3,
"deleted_files_count": 2,
},
],
},
}
task = build_root_task(config)
assert "Scope Constraints:" in task
assert "3 changed file(s)" in task
assert "2 deleted file(s)" in task
+84
View File
@@ -0,0 +1,84 @@
"""Tests for graceful handling of persistent RateLimitError in run_strix_scan."""
from __future__ import annotations
import logging
import types
from typing import Any
import httpx
import pytest
from openai import RateLimitError
import strix.tools.notes.tools as notes_tools
import strix.tools.todo.tools as todo_tools
from strix.core import runner
from strix.core.agents import AgentCoordinator
def _make_rate_limit_error() -> RateLimitError:
request = httpx.Request("POST", "https://api.openai.com/v1/responses")
response = httpx.Response(status_code=429, request=request)
return RateLimitError("rate limited", response=response, body=None)
@pytest.mark.asyncio
async def test_persistent_rate_limit_stops_gracefully(
monkeypatch: pytest.MonkeyPatch, tmp_path: Any, caplog: pytest.LogCaptureFixture
) -> None:
"""A persistent RateLimitError stops the scan (root -> 'stopped') without raising."""
monkeypatch.setattr(runner, "run_dir_for", lambda _scan_id: tmp_path)
monkeypatch.setattr(runner, "runtime_state_dir", lambda _run_dir: tmp_path)
monkeypatch.setattr(runner, "setup_scan_logging", lambda _run_dir: lambda: None)
monkeypatch.setattr(runner, "set_scan_id", lambda _scan_id: None)
settings = types.SimpleNamespace(
llm=types.SimpleNamespace(model="openai/gpt-4o", reasoning_effort="high")
)
monkeypatch.setattr(runner, "load_settings", lambda: settings)
monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
monkeypatch.setattr(
runner, "uses_chat_completions_tool_schema", lambda _model, _settings: False
)
monkeypatch.setattr(todo_tools, "hydrate_todos_from_disk", lambda _state_dir: None)
monkeypatch.setattr(notes_tools, "hydrate_notes_from_disk", lambda _state_dir: None)
async def _create_or_reuse(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
return {"client": object(), "session": object(), "caido_client": None}
async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
return None
monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)
monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)
monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: "")
monkeypatch.setattr(runner, "make_model_settings", lambda *_args, **_kwargs: object())
monkeypatch.setattr(runner, "build_strix_agent", lambda **_kwargs: object())
monkeypatch.setattr(runner, "make_child_factory", lambda **_kwargs: lambda **_k: object())
monkeypatch.setattr(runner, "open_agent_session", lambda _root_id, _db: object())
async def _raise_rate_limit(*_args: Any, **_kwargs: Any) -> None:
raise _make_rate_limit_error()
monkeypatch.setattr(runner, "run_agent_loop", _raise_rate_limit)
coordinator = AgentCoordinator()
with caplog.at_level(logging.WARNING):
result = await runner.run_strix_scan(
scan_config={"targets": [], "scan_mode": "deep"},
scan_id="scan-test",
image="img",
coordinator=coordinator,
)
assert result is None
root_ids = [aid for aid, parent in coordinator.parent_of.items() if parent is None]
assert len(root_ids) == 1
assert coordinator.statuses[root_ids[0]] == "stopped"
# the resume hint must carry the real scan id, not a literal placeholder
assert "strix --resume scan-test" in caplog.text
assert "<run_name>" not in caplog.text
Generated
+1067 -961
View File
File diff suppressed because it is too large Load Diff