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>

.pre-commit-config.yaml

@@ -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: [
@@ -19,6 +19,7 @@ repos:
           types-python-dateutil,
           pydantic,
           fastapi,
+          pytest,
           "openai-agents[litellm]==0.14.6",
         ]
         args: [--install-types, --non-interactive]
@@ -46,7 +47,7 @@ repos:
 
   # Additional Python code quality checks
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.20.0
+    rev: v3.21.2
     hooks:
       - id: pyupgrade
         args: [--py312-plus]

README.md

@@ -8,7 +8,7 @@
 
 # Strix
 
-### Open-source AI hackers to find and fix your app’s vulnerabilities.
+### The open-source AI pentesting tool. Autonomous AI hackers that find and fix your app’s 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
-- **Developer‑first** CLI with actionable reports
-- **Auto‑fix & 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
+- **Developer‑first CLI** - actionable findings with remediation guidance
+- **Auto‑fix & 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` 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.
+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
 

docs/advanced/configuration.mdx

@@ -79,6 +79,10 @@ When remote vars are set, Strix dual-writes telemetry to both local JSONL and th
   Runtime backend for the sandbox environment.
 </ParamField>
 
+<ParamField path="STRIX_MAX_LOCAL_COPY_MB" default="1024" type="integer">
+  Maximum size (in MB) of a local directory target that Strix will copy into the sandbox file-by-file. Larger targets exit early with a suggestion to use `--mount` instead. Set to `0` to disable the check.
+</ParamField>
+
 ## Sandbox Configuration
 
 <ParamField path="STRIX_SANDBOX_EXECUTION_TIMEOUT" default="120" type="integer">

docs/usage/cli.mdx

@@ -15,6 +15,20 @@ strix --target <target> [options]
   Target to test. Accepts URLs, repositories, local directories, domains, or IP addresses. Can be specified multiple times.
 </ParamField>
 
+<ParamField path="--mount" type="string">
+  Bind-mount a local directory into the sandbox (read-only) instead of copying it in file-by-file. Use this for large repositories that are too big to stream into the container. Can be specified multiple times.
+
+  Strix copies local `--target` directories into the sandbox one file at a time, which stalls on very large trees. When a local target exceeds the copy limit (see `STRIX_MAX_LOCAL_COPY_MB`, default 1024 MB) Strix exits early and asks you to re-run with `--mount`.
+
+  <Note>
+    The mount is read-only to protect your source from accidental modification. This is not a hard security boundary: a root process inside the container can remount it writable, so treat `--mount` as "scan my own code", not as isolation from untrusted code.
+  </Note>
+
+  <Note>
+    The size pre-flight only covers local directory targets. Remote repositories (cloned at scan time) are not size-checked.
+  </Note>
+</ParamField>
+
 <ParamField path="--instruction" type="string">
   Custom instructions for the scan. Use for credentials, focus areas, or specific testing approaches.
 </ParamField>
@@ -43,6 +57,24 @@ strix --target <target> [options]
   Path to a custom config file (JSON) to use instead of `~/.strix/cli-config.json`.
 </ParamField>
 
+<ParamField path="--max-budget-usd" type="number">
+  Maximum LLM spend in USD for the whole scan, counted cumulatively across the
+  root agent and every child agent. The budget is checked after each model
+  response; once the running cost reaches the threshold, the scan stops cleanly
+  with a `stopped` status (not a failure) and the sandbox is torn down.
+
+  Must be greater than `0`. Omit the flag for no limit.
+
+  **Limitations**
+
+  - The check fires *after* a response is returned, so the final spend can
+    slightly overshoot the limit by any calls already in flight when the
+    threshold is crossed (most relevant with several child agents running
+    concurrently).
+  - Cost is a best-effort estimate derived from token usage and model pricing;
+    providers that do not expose priced usage may under-count.
+</ParamField>
+
 ## Examples
 
 ```bash
@@ -63,6 +95,9 @@ strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main
 
 # Multi-target white-box testing
 strix -t https://github.com/org/app -t https://staging.example.com
+
+# Large local repository — bind-mount instead of copying it in
+strix --mount ./huge-monorepo
 ```
 
 ## Exit Codes

pyproject.toml

@@ -55,8 +55,13 @@ dev = [
   "bandit>=1.8.3",
   "pre-commit>=4.2.0",
   "pyinstaller>=6.17.0; python_version >= '3.12' and python_version < '3.15'",
+  "pytest>=8.3",
+  "pytest-asyncio>=0.24",
 ]
 
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"
@@ -104,6 +109,10 @@ module = [
 ignore_missing_imports = true
 disable_error_code = ["import-untyped"]
 
+[[tool.mypy.overrides]]
+module = ["tests.*"]
+disallow_untyped_decorators = false
+
 # ============================================================================
 # Ruff Configuration (Fast Python Linter & Formatter)
 # ============================================================================

strix/config/models.py

@@ -39,6 +39,8 @@ class StrixProvider(MultiProvider):
                 prefix=prefix,
                 stripped_model_name=stripped_model_name,
             )
+        if prefix == "ollama" and stripped_model_name:
+            return self._get_fallback_provider("litellm"), f"ollama_chat/{stripped_model_name}"
         return self._get_fallback_provider("litellm"), original_model_name
 
 

strix/config/settings.py

@@ -47,6 +47,11 @@ class RuntimeSettings(BaseSettings):
         alias="STRIX_IMAGE",
     )
     backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
+    # Hard cap on a local target's size before we refuse to stream it into the
+    # sandbox file-by-file (the SDK copies every file individually, which stalls
+    # on large repos). Above this, the user must bind-mount via ``--mount``.
+    # Set to 0 (or less) to disable the pre-flight check entirely.
+    max_local_copy_mb: int = Field(default=1024, alias="STRIX_MAX_LOCAL_COPY_MB")
 
 
 class TelemetrySettings(BaseSettings):

strix/core/agents.py

@@ -43,6 +43,7 @@ class AgentCoordinator:
         self._lock = asyncio.Lock()
         self._snapshot_path: Path | None = None
         self.is_shutting_down = False
+        self._budget_stopped = False
 
     def set_snapshot_path(self, path: Path) -> None:
         self._snapshot_path = path
@@ -50,6 +51,17 @@ class AgentCoordinator:
     def mark_shutting_down(self) -> None:
         self.is_shutting_down = True
 
+    @property
+    def budget_stopped(self) -> bool:
+        return self._budget_stopped
+
+    async def trigger_budget_stop(self) -> None:
+        """Signal a scan-wide budget stop and wake every parked agent so it exits."""
+        async with self._lock:
+            self._budget_stopped = True
+            for runtime in self.runtimes.values():
+                runtime.wake.set()
+
     async def register(
         self,
         agent_id: str,
@@ -143,7 +155,7 @@ class AgentCoordinator:
     async def wait_for_message(self, agent_id: str) -> None:
         while True:
             async with self._lock:
-                if self.pending_counts.get(agent_id, 0) > 0:
+                if self._budget_stopped or self.pending_counts.get(agent_id, 0) > 0:
                     return
                 wake = self.runtimes.setdefault(agent_id, AgentRuntime()).wake
                 wake.clear()

strix/core/execution.py

@@ -15,6 +15,7 @@ from agents.sandbox.errors import ExecTransportError
 from docker import errors as docker_errors  # type: ignore[import-untyped, unused-ignore]
 from openai import APIError
 
+from strix.core.hooks import BudgetExceededError
 from strix.core.inputs import child_initial_input
 from strix.core.sessions import open_agent_session, strip_all_images_from_session
 
@@ -97,6 +98,10 @@ async def run_agent_loop(
         except asyncio.CancelledError:
             return result
 
+        if coordinator.budget_stopped:
+            await coordinator.set_status(agent_id, "stopped")
+            raise BudgetExceededError("scan budget reached")
+
         await coordinator.consume_pending(agent_id)
         result = await _run_cycle(
             agent,
@@ -278,6 +283,10 @@ async def _run_noninteractive_until_lifecycle(
     invalid_final_output_limit = max(1, max_turns)
 
     while True:
+        if coordinator.budget_stopped:
+            await coordinator.set_status(agent_id, "stopped")
+            raise BudgetExceededError("scan budget reached")
+
         result = await _run_cycle(
             agent,
             coordinator,
@@ -360,6 +369,10 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                                 logger.exception("stream event sink failed for %s", agent_id)
                     if stream.run_loop_exception is not None:
                         raise stream.run_loop_exception
+                except BudgetExceededError:
+                    # A RuntimeError subclass: re-raise explicitly so it is never
+                    # mistaken for the LiteLLM "after shutdown" race below.
+                    raise
                 except RuntimeError as stream_exc:
                     if "after shutdown" not in str(stream_exc):
                         raise
@@ -377,6 +390,13 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                     )
             finally:
                 await coordinator.detach_stream(agent_id, stream)
+        except BudgetExceededError as exc:
+            logger.info(
+                "agent %s reached the scan budget limit; stopping the scan: %s", agent_id, exc
+            )
+            await coordinator.set_status(agent_id, "stopped")
+            await coordinator.trigger_budget_stop()
+            raise
         except Exception as exc:
             if (
                 image_strips < 3
@@ -527,21 +547,29 @@ async def _start_child_runner(
     child_ctx["parent_id"] = parent_id
     child_ctx["task"] = task
 
-    task_handle = asyncio.create_task(
-        run_agent_loop(
-            agent=child_agent,
-            initial_input=initial_input,
-            run_config=run_config,
-            context=child_ctx,
-            max_turns=max_turns,
-            coordinator=coordinator,
-            agent_id=child_id,
-            interactive=interactive,
-            session=session,
-            start_parked=start_parked,
-            event_sink=event_sink,
-            hooks=hooks,
-        ),
-        name=f"agent-{name}-{child_id}",
-    )
+    async def _child_loop() -> None:
+        # A budget stop is a clean scan-wide shutdown, not a child failure: the
+        # child's status and parent notification are already settled in
+        # ``_run_cycle``. Swallow it here so the detached task does not surface a
+        # spurious "Task exception was never retrieved" warning. The root agent
+        # hits the same limit on its next call and tears the scan down.
+        try:
+            await run_agent_loop(
+                agent=child_agent,
+                initial_input=initial_input,
+                run_config=run_config,
+                context=child_ctx,
+                max_turns=max_turns,
+                coordinator=coordinator,
+                agent_id=child_id,
+                interactive=interactive,
+                session=session,
+                start_parked=start_parked,
+                event_sink=event_sink,
+                hooks=hooks,
+            )
+        except BudgetExceededError:
+            logger.info("child %s stopped after reaching the scan budget limit", child_id)
+
+    task_handle = asyncio.create_task(_child_loop(), name=f"agent-{name}-{child_id}")
     await coordinator.attach_runtime(child_id, task=task_handle)

strix/core/hooks.py

@@ -19,11 +19,19 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+class BudgetExceededError(RuntimeError):
+    """Raised when the accumulated LLM cost reaches the configured budget."""
+
+
 class ReportUsageHooks(RunHooks[dict[str, Any]]):
     """Persist SDK-native usage after every model response."""
 
-    def __init__(self, *, model: str) -> None:
+    def __init__(self, *, model: str, max_budget_usd: float | None = None) -> None:
+        import math
+        if max_budget_usd is not None and (not math.isfinite(max_budget_usd) or max_budget_usd <= 0):
+            raise ValueError("max_budget_usd must be a finite number greater than 0")
         self._model = model
+        self._max_budget_usd = max_budget_usd
 
     async def on_llm_end(
         self,
@@ -52,3 +60,10 @@ class ReportUsageHooks(RunHooks[dict[str, Any]]):
             )
         except Exception:
             logger.exception("failed to record SDK usage for agent %s", agent_id)
+
+        if self._max_budget_usd is not None:
+            cost = report_state.get_total_llm_cost()
+            if cost >= self._max_budget_usd:
+                raise BudgetExceededError(
+                    f"Token budget of ${self._max_budget_usd:.2f} exceeded (spent ${cost:.4f})"
+                )

strix/core/inputs.py

@@ -44,7 +44,8 @@ def build_root_task(scan_config: dict[str, Any]) -> str:
             )
         elif ttype == "local_code":
             path = details.get("target_path", "unknown")
-            sections["Local Codebases"].append(f"- {path} (available at: {workspace_path})")
+            suffix = ", read-only mount" if details.get("mount") else ""
+            sections["Local Codebases"].append(f"- {path} (available at: {workspace_path}{suffix})")
         elif ttype == "web_application":
             sections["URLs"].append(f"- {details.get('target_url', '')}")
         elif ttype == "ip_address":
@@ -135,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)}]

strix/core/runner.py

@@ -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
@@ -27,7 +28,7 @@ from strix.core.execution import (
 from strix.core.execution import (
     spawn_child_agent as start_child_agent,
 )
-from strix.core.hooks import ReportUsageHooks
+from strix.core.hooks import BudgetExceededError, ReportUsageHooks
 from strix.core.inputs import (
     DEFAULT_MAX_TURNS,
     build_root_task,
@@ -55,10 +56,11 @@ async def run_strix_scan(
     scan_config: dict[str, Any],
     scan_id: str | None = None,
     image: str,
-    local_sources: list[dict[str, str]] | None = None,
+    local_sources: list[dict[str, Any]] | None = None,
     coordinator: AgentCoordinator | None = None,
     interactive: bool = False,
     max_turns: int = DEFAULT_MAX_TURNS,
+    max_budget_usd: float | None = None,
     model: str | None = None,
     cleanup_on_exit: bool = True,
     event_sink: StreamEventSink | None = None,
@@ -164,7 +166,7 @@ async def run_strix_scan(
             sandbox=SandboxRunConfig(client=bundle["client"], session=bundle["session"]),
             trace_include_sensitive_data=False,
         )
-        hooks = ReportUsageHooks(model=resolved_model)
+        hooks = ReportUsageHooks(model=resolved_model, max_budget_usd=max_budget_usd)
 
         scope_context = build_scope_context(scan_config)
 
@@ -300,6 +302,26 @@ async def run_strix_scan(
                     str(final)[:300],
                 )
         return result  # noqa: TRY300
+    except BudgetExceededError as exc:
+        logger.info("Scan %s stopped: %s", scan_id, exc)
+        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 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:

strix/interface/cli.py

@@ -183,6 +183,7 @@ async def run_cli(args: Any) -> None:  # noqa: PLR0915
                     image=_resolve_sandbox_image(),
                     local_sources=getattr(args, "local_sources", None) or [],
                     interactive=bool(getattr(args, "interactive", False)),
+                    max_budget_usd=getattr(args, "max_budget_usd", None),
                 )
             finally:
                 stop_updates.set()

strix/interface/main.py

@@ -33,9 +33,12 @@ from strix.interface.tui import run_tui
 from strix.interface.utils import (
     assign_workspace_subdirs,
     build_final_stats_text,
+    build_mount_targets_info,
     check_docker_connection,
     clone_repository,
     collect_local_sources,
+    dedupe_local_targets,
+    find_oversized_local_targets,
     generate_run_name,
     image_exists,
     infer_target_type,
@@ -301,6 +304,17 @@ def get_version() -> str:
         return "unknown"
 
 
+def _positive_budget(value: str) -> float:
+    try:
+        budget = float(value)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError(f"invalid float value: {value!r}") from exc
+    import math
+    if not math.isfinite(budget) or budget <= 0:
+        raise argparse.ArgumentTypeError("must be a finite number greater than 0")
+    return budget
+
+
 def parse_arguments() -> argparse.Namespace:
     parser = argparse.ArgumentParser(
         description="Strix Multi-Agent Cybersecurity Penetration Testing Tool",
@@ -317,6 +331,9 @@ Examples:
   # Local code analysis
   strix --target ./my-project
 
+  # Large local repository (bind-mounted read-only instead of copied)
+  strix --mount ./huge-monorepo
+
   # Domain penetration test
   strix --target example.com
 
@@ -352,6 +369,15 @@ Examples:
         "Can be specified multiple times for multi-target scans. "
         "Required for fresh runs; loaded from disk when ``--resume`` is set.",
     )
+    parser.add_argument(
+        "--mount",
+        type=str,
+        action="append",
+        metavar="PATH",
+        help="Bind-mount a local directory into the sandbox (read-only) instead of "
+        "copying it file-by-file. Use this for large repositories that are too big to "
+        "stream into the container. Can be specified multiple times.",
+    )
     parser.add_argument(
         "--instruction",
         type=str,
@@ -424,6 +450,13 @@ Examples:
         help="Path to a custom config file (JSON) to use instead of ~/.strix/cli-config.json",
     )
 
+    parser.add_argument(
+        "--max-budget-usd",
+        type=_positive_budget,
+        default=None,
+        help="Maximum LLM cost in USD (> 0). The scan stops cleanly when this limit is reached.",
+    )
+
     parser.add_argument(
         "--resume",
         type=str,
@@ -455,9 +488,9 @@ Examples:
     args.user_explicit_instruction = args.instruction if args.resume else None
 
     if args.resume:
-        if args.target:
+        if args.target or args.mount:
             parser.error(
-                "Cannot combine --resume with --target. --resume picks up where "
+                "Cannot combine --resume with --target/--mount. --resume picks up where "
                 "the prior run left off, including the original target list."
             )
         _load_resume_state(args, parser)
@@ -470,13 +503,13 @@ Examples:
                 f"or remove --resume to start over with the same targets."
             )
     else:
-        if not args.target:
+        if not args.target and not args.mount:
             parser.error(
-                "the following arguments are required: -t/--target "
+                "the following arguments are required: -t/--target or --mount "
                 "(or use --resume <run_name> to continue a prior scan)"
             )
         args.targets_info = []
-        for target in args.target:
+        for target in args.target or []:
             try:
                 target_type, target_dict = infer_target_type(target)
 
@@ -491,9 +524,30 @@ Examples:
             except ValueError:
                 parser.error(f"Invalid target '{target}'")
 
+        try:
+            args.targets_info.extend(build_mount_targets_info(args.mount or []))
+        except ValueError as e:
+            parser.error(str(e))
+
+        args.targets_info = dedupe_local_targets(args.targets_info)
+
         assign_workspace_subdirs(args.targets_info)
         rewrite_localhost_targets(args.targets_info, HOST_GATEWAY_HOSTNAME)
 
+        max_local_copy_mb = load_settings().runtime.max_local_copy_mb
+        max_copy_bytes = max_local_copy_mb * 1024 * 1024
+        oversized = find_oversized_local_targets(args.targets_info, max_copy_bytes)
+        if oversized:
+            details = "; ".join(
+                f"{path} ({size / (1024 * 1024):.0f} MB)" for path, size in oversized
+            )
+            parser.error(
+                f"Local target too large to stream into the sandbox: {details}. "
+                f"The limit is {max_local_copy_mb} MB "
+                "(set STRIX_MAX_LOCAL_COPY_MB to change it). Re-run with "
+                "--mount <path> to bind-mount the directory instead of copying it."
+            )
+
     return args
 
 
@@ -766,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()

strix/interface/tui/app.py

@@ -31,6 +31,7 @@ from textual.widgets import Button, Label, Static, TextArea, Tree
 from textual.widgets.tree import TreeNode
 
 from strix.config import load_settings
+from strix.core.hooks import BudgetExceededError
 from strix.core.runner import run_strix_scan
 from strix.interface.tui.live_view import TuiLiveView
 from strix.interface.tui.messages import send_user_message_to_agent
@@ -1369,12 +1370,18 @@ class StrixTUIApp(App):  # type: ignore[misc]
                                 local_sources=getattr(self.args, "local_sources", None) or [],
                                 coordinator=self.coordinator,
                                 interactive=True,
+                                max_budget_usd=getattr(self.args, "max_budget_usd", None),
                                 event_sink=self._capture_sdk_event,
                             ),
                         )
 
                 except (KeyboardInterrupt, asyncio.CancelledError):
                     logger.info("Scan interrupted by user")
+                except BudgetExceededError:
+                    # Defensive: the runner stops the scan cleanly on budget and
+                    # returns, so this normally never propagates. Treat it as a
+                    # graceful stop, not a scan error, if it ever does.
+                    logger.info("Scan stopped: --max-budget-usd limit reached")
                 except (ConnectionError, TimeoutError) as e:
                     logging.exception("Network error during scan")
                     self._scan_error = e

strix/interface/utils.py

@@ -1,5 +1,6 @@
 import ipaddress
 import json
+import logging
 import os
 import re
 import secrets
@@ -23,6 +24,9 @@ from rich.text import Text
 from strix.config import load_settings
 
 
+logger = logging.getLogger(__name__)
+
+
 def get_severity_color(severity: str) -> str:
     severity_colors = {
         "critical": "#dc2626",
@@ -1185,8 +1189,8 @@ def is_whitebox_scan(targets_info: list[dict[str, Any]]) -> bool:
     return any(t.get("type") == "local_code" for t in targets_info or [])
 
 
-def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str, str]]:
-    local_sources: list[dict[str, str]] = []
+def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    local_sources: list[dict[str, Any]] = []
 
     for target_info in targets_info:
         details = target_info["details"]
@@ -1197,6 +1201,7 @@ def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str,
                 {
                     "source_path": details["target_path"],
                     "workspace_subdir": workspace_subdir,
+                    "mount": bool(details.get("mount", False)),
                 }
             )
 
@@ -1205,12 +1210,126 @@ def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str,
                 {
                     "source_path": details["cloned_repo_path"],
                     "workspace_subdir": workspace_subdir,
+                    "mount": False,
                 }
             )
 
     return local_sources
 
 
+def directory_size_bytes(path: Path) -> int:
+    """Total size in bytes of regular files under ``path`` (symlinks not followed).
+
+    Best-effort: files that disappear or can't be stat'd mid-walk are skipped.
+    Used as a cheap (stat-only) pre-flight to estimate the cost of streaming a
+    local target into the sandbox before we actually try to copy it.
+
+    Directories that can't be listed (e.g. permission denied) are logged and
+    skipped rather than silently dropped — so an under-count is at least
+    visible — but the returned total then excludes their contents.
+    """
+
+    def _on_walk_error(error: OSError) -> None:
+        logger.warning("Could not read %s while measuring size: %s", error.filename, error)
+
+    total = 0
+    for root, _dirs, files in os.walk(path, followlinks=False, onerror=_on_walk_error):
+        for name in files:
+            file_path = os.path.join(root, name)  # noqa: PTH118
+            try:
+                if os.path.islink(file_path):  # noqa: PTH114
+                    continue
+                total += os.path.getsize(file_path)  # noqa: PTH202
+            except OSError:
+                continue
+    return total
+
+
+def find_oversized_local_targets(
+    targets_info: list[dict[str, Any]], max_bytes: int
+) -> list[tuple[str, int]]:
+    """Return ``(path, size_bytes)`` for non-mounted local targets over ``max_bytes``.
+
+    Mounted targets are bind-mounted rather than copied, so their size is
+    irrelevant and they are excluded. A ``max_bytes`` of zero or less disables
+    the check entirely (returns no targets).
+    """
+    if max_bytes <= 0:
+        return []
+    oversized: list[tuple[str, int]] = []
+    for target in targets_info:
+        if target.get("type") != "local_code":
+            continue
+        details = target.get("details") or {}
+        if details.get("mount"):
+            continue
+        target_path = details.get("target_path")
+        if not target_path:
+            continue
+        size = directory_size_bytes(Path(target_path))
+        if size > max_bytes:
+            oversized.append((target_path, size))
+    return oversized
+
+
+def build_mount_targets_info(mount_paths: list[str]) -> list[dict[str, Any]]:
+    """Build ``targets_info`` entries for ``--mount`` directories.
+
+    Each path must be an existing local directory; it is bind-mounted into the
+    sandbox (read-only) instead of being copied file-by-file. Raises
+    ``ValueError`` for an empty path, or one that does not exist or is not a
+    directory.
+    """
+    targets_info: list[dict[str, Any]] = []
+    for raw in mount_paths:
+        if not raw or not raw.strip():
+            raise ValueError("--mount path must not be empty.")
+        path = Path(raw).expanduser()
+        try:
+            resolved = path.resolve()
+            is_dir = resolved.is_dir()
+        except (OSError, RuntimeError) as e:
+            raise ValueError(f"Invalid mount path '{raw}': {e!s}") from e
+        if not is_dir:
+            raise ValueError(
+                f"Mount path '{raw}' is not an existing directory. "
+                "--mount requires a path to a local directory."
+            )
+        targets_info.append(
+            {
+                "type": "local_code",
+                "details": {"target_path": str(resolved), "mount": True},
+                "original": str(resolved),
+            }
+        )
+    return targets_info
+
+
+def dedupe_local_targets(targets_info: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Collapse local_code targets that resolve to the same path.
+
+    When a directory is supplied both as a copied ``--target`` and via
+    ``--mount`` (or as duplicate values of either), keep one entry and prefer
+    the bind-mounted one — so the same tree is never both streamed in and
+    mounted. Order is preserved; non-local targets pass through untouched.
+    """
+    result: list[dict[str, Any]] = []
+    index_by_path: dict[str, int] = {}
+    for target in targets_info:
+        details = target.get("details") or {}
+        path = details.get("target_path")
+        if target.get("type") != "local_code" or not path:
+            result.append(target)
+            continue
+        existing = index_by_path.get(path)
+        if existing is None:
+            index_by_path[path] = len(result)
+            result.append(target)
+        elif details.get("mount") and not (result[existing].get("details") or {}).get("mount"):
+            result[existing] = target  # bind mount supersedes the copied entry
+    return result
+
+
 def _is_localhost_host(host: str) -> bool:
     host_lower = host.lower().strip("[]")
 

strix/report/state.py

@@ -236,6 +236,10 @@ class ReportState:
     def get_total_llm_usage(self) -> dict[str, Any]:
         return dict(self.run_record.get("llm_usage") or self._build_llm_usage_record())
 
+    def get_total_llm_cost(self) -> float:
+        """Live accumulated LLM cost, independent of the persisted run-record snapshot."""
+        return self._llm_usage.total_cost
+
     def update_scan_final_fields(
         self,
         executive_summary: str,

strix/report/usage.py

@@ -52,6 +52,10 @@ class LLMUsageLedger:
         if isinstance(cost, int | float) and cost > 0:
             self._total_cost += float(cost)
 
+    @property
+    def total_cost(self) -> float:
+        return _round_cost(self._total_cost)
+
     def to_record(self) -> dict[str, Any]:
         record = serialize_usage(self._total_usage)
         record["cost"] = _round_cost(self._total_cost)

strix/report/writer.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import csv
+import io
 import json
 import logging
 import tempfile
@@ -27,7 +28,7 @@ def read_run_record(run_dir: Path) -> dict[str, Any]:
     except (OSError, json.JSONDecodeError) as exc:
         raise RuntimeError(f"run.json at {path} is unreadable: {exc}") from exc
     if not isinstance(data, dict):
-        raise RuntimeError(f"run.json at {path} is not an object")
+        raise TypeError(f"run.json at {path} is not an object")
     return data
 
 
@@ -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",

strix/runtime/backends.py

@@ -22,6 +22,7 @@ async def _docker_backend(
     image: str,
     manifest: Manifest,
     exposed_ports: tuple[int, ...],
+    bind_mounts: list[dict[str, Any]] | None = None,
 ) -> tuple[Any, Any]:
     """Bring up a session backed by the local Docker daemon.
 
@@ -31,11 +32,15 @@ async def _docker_backend(
     backend don't need the docker-py library installed.
 
     ``session.start()`` is what materializes the manifest entries
-    (LocalDir copies, mount setup, etc.) into the running container —
-    the SDK's ``client.create()`` only builds the inner session object
-    without applying the manifest. ``async with session:`` would call it
-    too, but Strix manages session lifetime explicitly via
+    (LocalDir copies and manifest-declared volume/FUSE mounts) into the
+    running container — the SDK's ``client.create()`` only builds the inner
+    session object without applying the manifest. ``async with session:``
+    would call it too, but Strix manages session lifetime explicitly via
     ``client.delete()`` so we trigger ``start()`` ourselves.
+
+    ``bind_mounts`` are host directories (e.g. large repos passed via
+    ``--mount``) bind-mounted read-only; unlike manifest entries they are
+    applied by Docker at container-create time, not by ``start()``.
     """
     import docker
     from agents.sandbox.sandboxes.docker import DockerSandboxClientOptions
@@ -43,6 +48,7 @@ async def _docker_backend(
     from strix.runtime.docker_client import StrixDockerSandboxClient
 
     client = StrixDockerSandboxClient(docker.from_env())
+    client.strix_bind_mounts = bind_mounts or []
     options = DockerSandboxClientOptions(image=image, exposed_ports=exposed_ports)
     session = await client.create(options=options, manifest=manifest)
     await session.start()

strix/runtime/docker_client.py

@@ -38,6 +38,7 @@ from agents.sandbox.sandboxes.docker import (
 from agents.sandbox.session.sandbox_session import SandboxSession
 from docker import errors as docker_errors  # type: ignore[import-untyped, unused-ignore]
 from docker.models.containers import Container  # type: ignore[import-untyped, unused-ignore]
+from docker.types import Mount as DockerSDKMount  # type: ignore[import-untyped, unused-ignore]
 from docker.utils import parse_repository_tag  # type: ignore[import-untyped, unused-ignore]
 
 
@@ -45,6 +46,10 @@ logger = logging.getLogger(__name__)
 
 
 class StrixDockerSandboxClient(DockerSandboxClient):
+    # Host directories to bind-mount into the container, set by the docker
+    # backend before ``create()``. Each item is ``{source, target, read_only}``.
+    strix_bind_mounts: list[dict[str, Any]] = []  # overridden per-instance in backends.py
+
     async def _create_container(
         self,
         image: str,
@@ -111,6 +116,21 @@ class StrixDockerSandboxClient(DockerSandboxClient):
         extra_hosts = create_kwargs.setdefault("extra_hosts", {})
         extra_hosts["host.docker.internal"] = "host-gateway"
 
+        # Strix injection: host bind mounts (e.g. large repos passed via --mount)
+        # that bypass the SDK's file-by-file LocalDir copy.
+        bind_mounts = getattr(self, "strix_bind_mounts", ())
+        if bind_mounts:
+            mounts = create_kwargs.setdefault("mounts", [])
+            for spec in bind_mounts:
+                mounts.append(
+                    DockerSDKMount(
+                        target=spec["target"],
+                        source=spec["source"],
+                        type="bind",
+                        read_only=spec.get("read_only", True),
+                    )
+                )
+
         logger.debug(
             "Creating sandbox container: image=%s caps=%s exposed_ports=%s",
             image,

strix/runtime/session_manager.py

@@ -23,30 +23,59 @@ _CONTAINER_CAIDO_PORT = 48080
 
 _SESSION_CACHE: dict[str, dict[str, Any]] = {}
 
+# Manifest root inside the container; entry keys hang off this path.
+_WORKSPACE_ROOT = "/workspace"
+
+
+def build_session_entries(
+    local_sources: list[dict[str, Any]],
+) -> tuple[dict[str | Path, BaseEntry], list[dict[str, Any]]]:
+    """Split local sources into copied manifest entries and host bind mounts.
+
+    Sources flagged ``mount`` are bind-mounted read-only at
+    ``/workspace/<workspace_subdir>`` (not added to the manifest, so the SDK
+    does not stream them in file-by-file). Every other source becomes a
+    ``LocalDir`` entry copied into the container as before.
+    """
+    entries: dict[str | Path, BaseEntry] = {}
+    bind_mounts: list[dict[str, Any]] = []
+    for src in local_sources:
+        ws_subdir = src.get("workspace_subdir") or ""
+        host_path = src.get("source_path") or ""
+        if not ws_subdir or not host_path:
+            continue
+        resolved = Path(host_path).expanduser().resolve()
+        if src.get("mount"):
+            bind_mounts.append(
+                {
+                    "source": str(resolved),
+                    "target": f"{_WORKSPACE_ROOT}/{ws_subdir}",
+                    "read_only": True,
+                }
+            )
+        else:
+            entries[ws_subdir] = LocalDir(src=resolved)
+    return entries, bind_mounts
+
 
 async def create_or_reuse(
     scan_id: str,
     *,
     image: str,
-    local_sources: list[dict[str, str]],
+    local_sources: list[dict[str, Any]],
 ) -> dict[str, Any]:
     """Return the existing session bundle for ``scan_id`` or create a new one.
 
-    Each ``local_sources`` entry mounts its host ``source_path`` at
-    ``/workspace/<workspace_subdir>`` inside the container.
+    Each ``local_sources`` entry exposes its host ``source_path`` at
+    ``/workspace/<workspace_subdir>`` inside the container — copied in, or
+    bind-mounted read-only when the entry is flagged ``mount``.
     """
     cached = _SESSION_CACHE.get(scan_id)
     if cached is not None:
         logger.info("Reusing existing sandbox session for scan %s", scan_id)
         return cached
 
-    entries: dict[str | Path, BaseEntry] = {}
-    for src in local_sources:
-        ws_subdir = src.get("workspace_subdir") or ""
-        host_path = src.get("source_path") or ""
-        if not ws_subdir or not host_path:
-            continue
-        entries[ws_subdir] = LocalDir(src=Path(host_path).expanduser().resolve())
+    entries, bind_mounts = build_session_entries(local_sources)
 
     # Caido runs as an in-container sidecar; HTTP(S) traffic from any
     # process started via ``session.exec`` (the SDK's Shell tool, etc.)
@@ -81,6 +110,7 @@ async def create_or_reuse(
         image=image,
         manifest=manifest,
         exposed_ports=(_CONTAINER_CAIDO_PORT,),
+        bind_mounts=bind_mounts,
     )
 
     caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)

strix/telemetry/posthog.py

@@ -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)

strix/telemetry/scarf.py

@@ -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)

tests/test_config_loader.py

@@ -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

tests/test_execution.py

@@ -0,0 +1,44 @@
+"""Tests for the scan-wide budget-stop signal on the agent coordinator."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from strix.core.agents import AgentCoordinator
+
+
+@pytest.mark.asyncio
+async def test_budget_stop_sets_flag() -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+
+    assert coordinator.budget_stopped is False
+    await coordinator.trigger_budget_stop()
+    assert coordinator.budget_stopped is True
+
+
+@pytest.mark.asyncio
+async def test_budget_stop_unblocks_parked_agent() -> None:
+    # A parent parked in wait_for_message (awaiting a child) must be released so
+    # it can exit, no matter where in the tree the budget limit was hit.
+    coordinator = AgentCoordinator()
+    await coordinator.register("parent", "strix", parent_id=None)
+
+    waiter = asyncio.create_task(coordinator.wait_for_message("parent"))
+    await asyncio.sleep(0)  # let the waiter park
+    assert not waiter.done()
+
+    await coordinator.trigger_budget_stop()
+    await asyncio.wait_for(waiter, timeout=1.0)
+
+
+@pytest.mark.asyncio
+async def test_wait_for_message_returns_immediately_after_budget_stop() -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("agent", "recon", parent_id="parent")
+    await coordinator.trigger_budget_stop()
+
+    # No pending messages, but the stop flag short-circuits the wait.
+    await asyncio.wait_for(coordinator.wait_for_message("agent"), timeout=1.0)

tests/test_hooks.py

@@ -0,0 +1,108 @@
+"""Tests for budget enforcement in ReportUsageHooks."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from strix.core.hooks import BudgetExceededError, ReportUsageHooks
+
+
+def _make_hooks(max_budget: float | None) -> ReportUsageHooks:
+    return ReportUsageHooks(model="test-model", max_budget_usd=max_budget)
+
+
+def _make_report_state(cost: float) -> MagicMock:
+    state = MagicMock()
+    state.get_total_llm_cost.return_value = cost
+    state.record_sdk_usage = MagicMock()
+    return state
+
+
+def _make_context(agent_id: str = "test-agent") -> MagicMock:
+    ctx: MagicMock = MagicMock()
+    ctx.context = {"agent_id": agent_id}
+    return ctx
+
+
+@pytest.mark.asyncio
+async def test_no_budget_never_raises() -> None:
+    hooks = _make_hooks(None)
+    state = _make_report_state(9999.0)
+    with patch("strix.core.hooks.get_global_report_state", return_value=state):
+        await hooks.on_llm_end(_make_context(), MagicMock(), MagicMock())
+
+
+@pytest.mark.asyncio
+async def test_under_budget_does_not_raise() -> None:
+    hooks = _make_hooks(10.0)
+    state = _make_report_state(9.99)
+    with patch("strix.core.hooks.get_global_report_state", return_value=state):
+        await hooks.on_llm_end(_make_context(), MagicMock(), MagicMock())
+
+
+@pytest.mark.asyncio
+async def test_at_budget_raises() -> None:
+    hooks = _make_hooks(10.0)
+    state = _make_report_state(10.0)
+    with (
+        patch("strix.core.hooks.get_global_report_state", return_value=state),
+        pytest.raises(BudgetExceededError),
+    ):
+        await hooks.on_llm_end(_make_context(), MagicMock(), MagicMock())
+
+
+@pytest.mark.asyncio
+async def test_over_budget_raises() -> None:
+    hooks = _make_hooks(10.0)
+    state = _make_report_state(10.01)
+    with (
+        patch("strix.core.hooks.get_global_report_state", return_value=state),
+        pytest.raises(BudgetExceededError),
+    ):
+        await hooks.on_llm_end(_make_context(), MagicMock(), MagicMock())
+
+
+@pytest.mark.asyncio
+async def test_budget_check_uses_live_cost_accessor() -> None:
+    # The check must read the live ledger, not the persisted run-record snapshot,
+    # so it stays accurate even when a save fails after a usage record.
+    hooks = _make_hooks(5.0)
+    state = _make_report_state(6.0)
+    with (
+        patch("strix.core.hooks.get_global_report_state", return_value=state),
+        pytest.raises(BudgetExceededError),
+    ):
+        await hooks.on_llm_end(_make_context(), MagicMock(), MagicMock())
+    state.get_total_llm_cost.assert_called_once()
+    state.get_total_llm_usage.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_error_message_includes_amounts() -> None:
+    hooks = _make_hooks(5.0)
+    state = _make_report_state(7.1234)
+    with patch("strix.core.hooks.get_global_report_state", return_value=state):
+        with pytest.raises(BudgetExceededError, match=r"\$5\.00") as exc_info:
+            await hooks.on_llm_end(_make_context(), MagicMock(), MagicMock())
+        assert "7.1234" in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_no_raise_when_report_state_none() -> None:
+    hooks = _make_hooks(1.0)
+    with patch("strix.core.hooks.get_global_report_state", return_value=None):
+        # Should return early without raising, even with budget set
+        await hooks.on_llm_end(_make_context(), MagicMock(), MagicMock())
+
+
+@pytest.mark.parametrize("bad_budget", [0.0, -0.01, -5.0])
+def test_non_positive_budget_rejected(bad_budget: float) -> None:
+    with pytest.raises(ValueError, match="greater than 0"):
+        ReportUsageHooks(model="test-model", max_budget_usd=bad_budget)
+
+
+def test_budget_exceeded_error_is_runtime_error() -> None:
+    err = BudgetExceededError("test")
+    assert isinstance(err, RuntimeError)

tests/test_inputs.py

@@ -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

tests/test_local_sources.py

@@ -0,0 +1,188 @@
+"""Tests for local-source sizing and ``--mount`` target helpers in interface.utils."""
+
+from __future__ import annotations
+
+import logging
+import os
+import sys
+from typing import TYPE_CHECKING, Any
+
+import pytest
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+from strix.interface.utils import (
+    build_mount_targets_info,
+    collect_local_sources,
+    dedupe_local_targets,
+    directory_size_bytes,
+    find_oversized_local_targets,
+)
+
+
+def _write_file(path: Path, size: int) -> None:
+    path.write_bytes(b"x" * size)
+
+
+def _local_target(target_path: str, *, mount: bool = False) -> dict[str, Any]:
+    details: dict[str, Any] = {"target_path": target_path, "workspace_subdir": "repo"}
+    if mount:
+        details["mount"] = True
+    return {"type": "local_code", "details": details, "original": target_path}
+
+
+def test_directory_size_empty_dir_is_zero(tmp_path: Path) -> None:
+    assert directory_size_bytes(tmp_path) == 0
+
+
+def test_directory_size_sums_flat_and_nested_files(tmp_path: Path) -> None:
+    _write_file(tmp_path / "a.txt", 100)
+    nested = tmp_path / "sub" / "deep"
+    nested.mkdir(parents=True)
+    _write_file(nested / "b.txt", 250)
+    assert directory_size_bytes(tmp_path) == 350
+
+
+def test_directory_size_skips_symlinks(tmp_path: Path) -> None:
+    _write_file(tmp_path / "real.txt", 100)
+    (tmp_path / "link.txt").symlink_to(tmp_path / "real.txt")
+    # The symlink target is counted once via the real file, not doubled.
+    assert directory_size_bytes(tmp_path) == 100
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="relies on POSIX permissions")
+def test_directory_size_logs_and_skips_unreadable_subdir(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    if hasattr(os, "geteuid") and os.geteuid() == 0:
+        pytest.skip("root bypasses directory permissions")
+    _write_file(tmp_path / "top.txt", 100)
+    locked = tmp_path / "locked"
+    locked.mkdir()
+    _write_file(locked / "secret.bin", 9999)
+    locked.chmod(0o000)
+    try:
+        with caplog.at_level(logging.WARNING):
+            size = directory_size_bytes(tmp_path)
+    finally:
+        locked.chmod(0o755)
+    # The unreadable subtree is excluded (not silently treated as readable) and
+    # the omission is logged rather than vanishing without a trace.
+    assert size == 100
+    assert any("Could not read" in record.message for record in caplog.records)
+
+
+def test_find_oversized_returns_nothing_under_limit(tmp_path: Path) -> None:
+    _write_file(tmp_path / "a.txt", 100)
+    targets = [_local_target(str(tmp_path))]
+    assert find_oversized_local_targets(targets, max_bytes=1000) == []
+
+
+def test_find_oversized_returns_target_over_limit(tmp_path: Path) -> None:
+    _write_file(tmp_path / "big.bin", 500)
+    targets = [_local_target(str(tmp_path))]
+    result = find_oversized_local_targets(targets, max_bytes=100)
+    assert result == [(str(tmp_path), 500)]
+
+
+def test_find_oversized_ignores_mounted_targets(tmp_path: Path) -> None:
+    _write_file(tmp_path / "big.bin", 500)
+    targets = [_local_target(str(tmp_path), mount=True)]
+    assert find_oversized_local_targets(targets, max_bytes=100) == []
+
+
+def test_find_oversized_ignores_non_local_targets() -> None:
+    targets = [{"type": "web_application", "details": {"target_url": "https://x"}}]
+    assert find_oversized_local_targets(targets, max_bytes=1) == []
+
+
+@pytest.mark.parametrize("disabled", [0, -1])
+def test_find_oversized_disabled_for_non_positive_limit(tmp_path: Path, disabled: int) -> None:
+    _write_file(tmp_path / "big.bin", 500)
+    targets = [_local_target(str(tmp_path))]
+    assert find_oversized_local_targets(targets, max_bytes=disabled) == []
+
+
+def test_collect_local_sources_propagates_mount_flag() -> None:
+    copied = _local_target("/copied")
+    copied["details"]["workspace_subdir"] = "copied"
+    mounted = _local_target("/mounted", mount=True)
+    mounted["details"]["workspace_subdir"] = "mounted"
+
+    sources = collect_local_sources([copied, mounted])
+
+    by_path = {s["source_path"]: s for s in sources}
+    assert by_path["/copied"]["mount"] is False
+    assert by_path["/mounted"]["mount"] is True
+
+
+def test_collect_local_sources_repository_is_never_mounted() -> None:
+    repo = {
+        "type": "repository",
+        "details": {"cloned_repo_path": "/clone", "workspace_subdir": "clone"},
+    }
+    sources = collect_local_sources([repo])
+    assert sources == [{"source_path": "/clone", "workspace_subdir": "clone", "mount": False}]
+
+
+def test_build_mount_targets_info_for_valid_dir(tmp_path: Path) -> None:
+    result = build_mount_targets_info([str(tmp_path)])
+    assert len(result) == 1
+    entry = result[0]
+    assert entry["type"] == "local_code"
+    assert entry["details"]["mount"] is True
+    assert entry["details"]["target_path"] == str(tmp_path.resolve())
+
+
+def test_build_mount_targets_info_rejects_missing_path(tmp_path: Path) -> None:
+    missing = tmp_path / "does-not-exist"
+    with pytest.raises(ValueError, match="not an existing directory"):
+        build_mount_targets_info([str(missing)])
+
+
+def test_build_mount_targets_info_rejects_file(tmp_path: Path) -> None:
+    file_path = tmp_path / "a-file.txt"
+    _write_file(file_path, 10)
+    with pytest.raises(ValueError, match="not an existing directory"):
+        build_mount_targets_info([str(file_path)])
+
+
+@pytest.mark.parametrize("empty", ["", "   "])
+def test_build_mount_targets_info_rejects_empty_path(empty: str) -> None:
+    # An empty path would otherwise resolve to the current working directory
+    # and silently bind-mount it into the sandbox.
+    with pytest.raises(ValueError, match="must not be empty"):
+        build_mount_targets_info([empty])
+
+
+def test_dedupe_keeps_distinct_targets_in_order() -> None:
+    targets = [
+        _local_target("/a"),
+        {"type": "web_application", "details": {"target_url": "https://x"}},
+        _local_target("/b", mount=True),
+    ]
+    assert dedupe_local_targets(targets) == targets
+
+
+def test_dedupe_mount_supersedes_copied_same_path() -> None:
+    copied = _local_target("/repo")
+    mounted = _local_target("/repo", mount=True)
+
+    # Copied first, then mounted: the single surviving entry is the mount.
+    result = dedupe_local_targets([copied, mounted])
+    assert len(result) == 1
+    assert result[0]["details"]["mount"] is True
+
+    # Order-independent: mounted first, copied second also yields the mount.
+    result_rev = dedupe_local_targets([mounted, copied])
+    assert len(result_rev) == 1
+    assert result_rev[0]["details"]["mount"] is True
+
+
+def test_dedupe_collapses_duplicate_mounts() -> None:
+    result = dedupe_local_targets(
+        [_local_target("/repo", mount=True), _local_target("/repo", mount=True)]
+    )
+    assert len(result) == 1

tests/test_runner_rate_limit.py

@@ -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

tests/test_session_entries.py

@@ -0,0 +1,67 @@
+"""Tests for build_session_entries: splitting copied vs bind-mounted sources."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from agents.sandbox.entries import LocalDir
+
+from strix.runtime.session_manager import build_session_entries
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _source(subdir: str, path: str, *, mount: bool = False) -> dict[str, Any]:
+    return {"source_path": path, "workspace_subdir": subdir, "mount": mount}
+
+
+def test_copied_source_becomes_localdir_entry(tmp_path: Path) -> None:
+    entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path))])
+
+    assert bind_mounts == []
+    assert isinstance(entries["repo"], LocalDir)
+    assert entries["repo"].src == tmp_path.resolve()
+
+
+def test_mounted_source_becomes_bind_mount(tmp_path: Path) -> None:
+    entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path), mount=True)])
+
+    assert entries == {}
+    assert bind_mounts == [
+        {
+            "source": str(tmp_path.resolve()),
+            "target": "/workspace/repo",
+            "read_only": True,
+        }
+    ]
+
+
+def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
+    copied = tmp_path / "copied"
+    mounted = tmp_path / "mounted"
+    copied.mkdir()
+    mounted.mkdir()
+
+    entries, bind_mounts = build_session_entries(
+        [
+            _source("copied", str(copied)),
+            _source("mounted", str(mounted), mount=True),
+        ]
+    )
+
+    assert list(entries) == ["copied"]
+    assert isinstance(entries["copied"], LocalDir)
+    assert [m["target"] for m in bind_mounts] == ["/workspace/mounted"]
+
+
+def test_incomplete_sources_are_skipped() -> None:
+    entries, bind_mounts = build_session_entries(
+        [
+            {"source_path": "", "workspace_subdir": "x"},
+            {"source_path": "/p", "workspace_subdir": ""},
+        ]
+    )
+    assert entries == {}
+    assert bind_mounts == []