From 18a36520a731fc048266463fb66fd7e809b1adfd Mon Sep 17 00:00:00 2001
From: maxi-maxima <240684166@qq.com>
Date: Tue, 11 Aug 2026 10:44:09 +0800
Subject: [PATCH] Wait for Caido readiness by deadline
---
docs/advanced/configuration.mdx | 333 +++++++++++-----------
strix/config/settings.py | 313 ++++++++++-----------
strix/runtime/caido_bootstrap.py | 246 +++++++++-------
strix/runtime/session_manager.py | 468 ++++++++++++++++---------------
tests/test_caido_bootstrap.py | 104 +++++++
tests/test_config_loader.py | 443 +++++++++++++++--------------
6 files changed, 1027 insertions(+), 880 deletions(-)
create mode 100644 tests/test_caido_bootstrap.py
diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx
index f1542b752..d5678478e 100644
--- a/docs/advanced/configuration.mdx
+++ b/docs/advanced/configuration.mdx
@@ -1,169 +1,174 @@
----
-title: "Configuration"
-description: "Environment variables for Strix"
----
-
-Configure Strix using environment variables or a config file.
-
-## LLM Configuration
-
-
- Model name in LiteLLM format (e.g., `openai/gpt-5.4`, `anthropic/claude-sonnet-4-6`).
-
-
-
- API key for your LLM provider. Not required for local models or cloud provider auth (Vertex AI, AWS Bedrock).
-
-
-
- Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`.
-
-
-
- Extra HTTP headers sent on every LLM request, as a JSON object (e.g.
- `{"X-Feature-Key":"value","X-Tenant":"acme"}`). Useful for OpenAI-compatible
- gateways that require attribution or routing headers in addition to the bearer
- token. The bearer token itself still comes from `LLM_API_KEY`. Applies to both
- the LiteLLM and native OpenAI routing paths.
-
-
-
- Request timeout in seconds for LLM calls.
-
-
-
- Maximum number of retries for LLM API calls on transient failures.
-
-
-
- Control thinking effort for reasoning models. Valid values: `none`, `minimal`, `low`, `medium`, `high`, `xhigh`, `max`. Defaults to `medium` for quick scan mode.
-
-
-
- Timeout in seconds for memory compression operations (context summarization).
-
-
-### Dedicated deduplication model
-
-Finding deduplication is a cheap, structured classification task. By default it
-runs on the main model, but you can route it to a smaller/cheaper model without
-affecting the agents that do the actual testing.
-
-
- Model used to judge whether a candidate finding duplicates an existing report.
- Falls back to `STRIX_LLM` when unset.
-
-
-
- Optional provider key for the deduplication model.
-
-
-
- Optional custom API base URL for the deduplication model. Use when the dedupe
- model runs on a different endpoint than the main model.
-
-
-
- Optional JSON object of extra HTTP headers sent on every deduplication-model
- request, e.g. `{"X-Feature-Key":"value"}`. A dedicated dedupe model never
- inherits `LLM_EXTRA_HEADERS`; set this when its endpoint needs custom headers.
-
-
-
- Reasoning effort for the deduplication model. Defaults to the model's own
- baseline when unset.
-
-
-## Optional Features
-
-
- API key for Perplexity AI. Enables real-time web search during scans for OSINT and vulnerability research.
-
-
-
- Postman API key (`PMAK-…`). Enables fetching Postman collections by id as a target (`postman://`), and Postman environments (`postman://?env=`) to resolve collection variables. Not needed when passing a local collection export file.
-
-
-
- Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL).
-
-
-
- OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only.
-
-
-
- API key used for remote trace export. Remote export is enabled only when both `TRACELOOP_BASE_URL` and `TRACELOOP_API_KEY` are set.
-
-
-
- Optional custom OTEL headers (JSON object or `key=value,key2=value2`). Useful for Langfuse or custom/self-hosted OTLP gateways.
-
-
-When remote OTEL vars are not set, Strix still writes complete run telemetry locally to:
-
-```bash
-strix_runs//events.jsonl
-```
-
-When remote vars are set, Strix dual-writes telemetry to both local JSONL and the remote OTEL endpoint.
-
-## Docker Configuration
-
-
- Docker image to use for the sandbox container.
-
-
-
- Docker daemon socket path. Use for remote Docker hosts or custom configurations.
-
-
+---
+title: "Configuration"
+description: "Environment variables for Strix"
+---
+
+Configure Strix using environment variables or a config file.
+
+## LLM Configuration
+
+
+ Model name in LiteLLM format (e.g., `openai/gpt-5.4`, `anthropic/claude-sonnet-4-6`).
+
+
+
+ API key for your LLM provider. Not required for local models or cloud provider auth (Vertex AI, AWS Bedrock).
+
+
+
+ Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`.
+
+
+
+ Extra HTTP headers sent on every LLM request, as a JSON object (e.g.
+ `{"X-Feature-Key":"value","X-Tenant":"acme"}`). Useful for OpenAI-compatible
+ gateways that require attribution or routing headers in addition to the bearer
+ token. The bearer token itself still comes from `LLM_API_KEY`. Applies to both
+ the LiteLLM and native OpenAI routing paths.
+
+
+
+ Request timeout in seconds for LLM calls.
+
+
+
+ Maximum number of retries for LLM API calls on transient failures.
+
+
+
+ Control thinking effort for reasoning models. Valid values: `none`, `minimal`, `low`, `medium`, `high`, `xhigh`, `max`. Defaults to `medium` for quick scan mode.
+
+
+
+ Timeout in seconds for memory compression operations (context summarization).
+
+
+### Dedicated deduplication model
+
+Finding deduplication is a cheap, structured classification task. By default it
+runs on the main model, but you can route it to a smaller/cheaper model without
+affecting the agents that do the actual testing.
+
+
+ Model used to judge whether a candidate finding duplicates an existing report.
+ Falls back to `STRIX_LLM` when unset.
+
+
+
+ Optional provider key for the deduplication model.
+
+
+
+ Optional custom API base URL for the deduplication model. Use when the dedupe
+ model runs on a different endpoint than the main model.
+
+
+
+ Optional JSON object of extra HTTP headers sent on every deduplication-model
+ request, e.g. `{"X-Feature-Key":"value"}`. A dedicated dedupe model never
+ inherits `LLM_EXTRA_HEADERS`; set this when its endpoint needs custom headers.
+
+
+
+ Reasoning effort for the deduplication model. Defaults to the model's own
+ baseline when unset.
+
+
+## Optional Features
+
+
+ API key for Perplexity AI. Enables real-time web search during scans for OSINT and vulnerability research.
+
+
+
+ Postman API key (`PMAK-…`). Enables fetching Postman collections by id as a target (`postman://`), and Postman environments (`postman://?env=`) to resolve collection variables. Not needed when passing a local collection export file.
+
+
+
+ Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL).
+
+
+
+ OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only.
+
+
+
+ API key used for remote trace export. Remote export is enabled only when both `TRACELOOP_BASE_URL` and `TRACELOOP_API_KEY` are set.
+
+
+
+ Optional custom OTEL headers (JSON object or `key=value,key2=value2`). Useful for Langfuse or custom/self-hosted OTLP gateways.
+
+
+When remote OTEL vars are not set, Strix still writes complete run telemetry locally to:
+
+```bash
+strix_runs//events.jsonl
+```
+
+When remote vars are set, Strix dual-writes telemetry to both local JSONL and the remote OTEL endpoint.
+
+## Docker Configuration
+
+
+ Docker image to use for the sandbox container.
+
+
+
+ Docker daemon socket path. Use for remote Docker hosts or custom configurations.
+
+
Runtime backend for the sandbox environment.
+
+ Maximum time in seconds to wait for the in-sandbox Caido proxy to become ready.
+ Increase this on slow or heavily loaded container hosts.
+
+
## Sandbox Configuration
-
-
- Maximum execution time in seconds for sandbox operations.
-
-
-
- Timeout in seconds for connecting to the sandbox container.
-
-
-## Config File
-
-Strix stores configuration in `~/.strix/cli-config.json`. You can also specify a custom config file:
-
-```bash
-strix --target ./app --config /path/to/config.json
-```
-
-**Config file format:**
-
-```json
-{
- "env": {
- "STRIX_LLM": "openai/gpt-5.4",
- "LLM_API_KEY": "sk-...",
- "STRIX_REASONING_EFFORT": "high"
- }
-}
-```
-
-## Example Setup
-
-```bash
-# Required
-export STRIX_LLM="openai/gpt-5.4"
-export LLM_API_KEY="sk-..."
-
-# Optional: Enable web search
-export PERPLEXITY_API_KEY="pplx-..."
-
-# Optional: Custom timeouts
-export LLM_TIMEOUT="600"
-export STRIX_SANDBOX_EXECUTION_TIMEOUT="300"
-
-```
+
+
+ Maximum execution time in seconds for sandbox operations.
+
+
+
+ Timeout in seconds for connecting to the sandbox container.
+
+
+## Config File
+
+Strix stores configuration in `~/.strix/cli-config.json`. You can also specify a custom config file:
+
+```bash
+strix --target ./app --config /path/to/config.json
+```
+
+**Config file format:**
+
+```json
+{
+ "env": {
+ "STRIX_LLM": "openai/gpt-5.4",
+ "LLM_API_KEY": "sk-...",
+ "STRIX_REASONING_EFFORT": "high"
+ }
+}
+```
+
+## Example Setup
+
+```bash
+# Required
+export STRIX_LLM="openai/gpt-5.4"
+export LLM_API_KEY="sk-..."
+
+# Optional: Enable web search
+export PERPLEXITY_API_KEY="pplx-..."
+
+# Optional: Custom timeouts
+export LLM_TIMEOUT="600"
+export STRIX_SANDBOX_EXECUTION_TIMEOUT="300"
+
+```
diff --git a/strix/config/settings.py b/strix/config/settings.py
index 42a2c97ea..dda0ebf37 100644
--- a/strix/config/settings.py
+++ b/strix/config/settings.py
@@ -1,156 +1,157 @@
-"""Strix application settings — pydantic-settings powered."""
-
-from __future__ import annotations
-
-from typing import Literal
-
-from pydantic import AliasChoices, Field
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-ReasoningEffort = Literal["none", "minimal", "low", "medium", "high", "xhigh", "max"]
-
-DEFAULT_MAX_TURNS = 500
-
-_BASE_CONFIG = SettingsConfigDict(
- case_sensitive=False,
- populate_by_name=True,
- extra="ignore",
-)
-
-
-class LlmSettings(BaseSettings):
- model_config = _BASE_CONFIG
-
- model: str | None = Field(default=None, alias="STRIX_LLM")
- api_key: str | None = Field(
- default=None,
- validation_alias=AliasChoices("LLM_API_KEY", "OPENAI_API_KEY"),
- repr=False,
- )
- api_base: str | None = Field(
- default=None,
- validation_alias=AliasChoices(
- "LLM_API_BASE",
- "OPENAI_API_BASE",
- "OPENAI_BASE_URL",
- "LITELLM_BASE_URL",
- "OLLAMA_API_BASE",
- ),
- )
- extra_headers: dict[str, str] | None = Field(
- default=None,
- alias="LLM_EXTRA_HEADERS",
- repr=False,
- )
- reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
- force_required_tool_choice: bool = Field(
- default=False,
- alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
- )
- prompt_cache: bool = Field(
- default=True,
- alias="STRIX_PROMPT_CACHE",
- )
- disable_streaming: bool = Field(
- default=False,
- alias="LLM_DISABLE_STREAMING",
- )
- timeout: int = Field(default=300, alias="LLM_TIMEOUT")
- stream_idle_timeout: int = Field(default=300, ge=0, alias="LLM_STREAM_IDLE_TIMEOUT")
- max_tool_calls_per_turn: int = Field(
- default=32,
- ge=0,
- alias="LLM_MAX_TOOL_CALLS_PER_TURN",
- )
-
-
-class DedupeSettings(BaseSettings):
- model_config = _BASE_CONFIG
-
- model: str | None = Field(default=None, alias="STRIX_DEDUPE_MODEL")
- reasoning_effort: ReasoningEffort | None = Field(
- default=None,
- alias="STRIX_DEDUPE_REASONING_EFFORT",
- )
- api_key: str | None = Field(default=None, alias="DEDUPE_LLM_API_KEY", repr=False)
- api_base: str | None = Field(default=None, alias="DEDUPE_LLM_API_BASE")
- extra_headers: dict[str, str] | None = Field(
- default=None,
- alias="DEDUPE_LLM_EXTRA_HEADERS",
- repr=False,
- )
-
-
-class ContextSettings(BaseSettings):
- """Context-window management: per-tool-output caps and history compaction."""
-
- model_config = _BASE_CONFIG
-
- auto_compact: bool = Field(default=True, alias="STRIX_CONTEXT_AUTO_COMPACT")
- compact_buffer_tokens: int = Field(default=20_000, gt=0, alias="STRIX_CONTEXT_BUFFER_TOKENS")
- keep_tokens: int = Field(default=8_000, gt=0, alias="STRIX_CONTEXT_KEEP_TOKENS")
- fallback_context_tokens: int = Field(
- default=200_000, gt=0, alias="STRIX_CONTEXT_FALLBACK_TOKENS"
- )
- summary_max_tokens: int = Field(default=4_096, gt=0, alias="STRIX_CONTEXT_SUMMARY_TOKENS")
- tool_output_max_tokens: int = Field(default=8_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_TOKENS")
- tool_output_max_lines: int = Field(default=2_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_LINES")
- # Floor above the truncation-notice size so a preview always fits.
- tool_output_max_bytes: int = Field(
- default=50 * 1024, ge=1024, alias="STRIX_TOOL_OUTPUT_MAX_BYTES"
- )
-
-
-class RuntimeSettings(BaseSettings):
- model_config = _BASE_CONFIG
-
- image: str = Field(
- default="ghcr.io/usestrix/strix-sandbox:1.3.0",
- alias="STRIX_IMAGE",
- )
- backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
- # Max screenshot/image tool outputs kept live per agent context (0 = none).
- max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES")
-
-
-class TelemetrySettings(BaseSettings):
- model_config = _BASE_CONFIG
-
- enabled: bool = Field(default=True, alias="STRIX_TELEMETRY")
-
-
-class IntegrationSettings(BaseSettings):
- model_config = _BASE_CONFIG
-
- perplexity_api_key: str | None = Field(
- default=None,
- alias="PERPLEXITY_API_KEY",
- repr=False,
- )
- postman_api_key: str | None = Field(
- default=None,
- alias="POSTMAN_API_KEY",
- repr=False,
- )
-
-
-class ViewerSettings(BaseSettings):
- model_config = _BASE_CONFIG
-
- # Base URL of the Strix relay the local viewer proxies to for email
- # verification and encrypted report delivery. The browser never talks to
- # the relay directly; the local server is the only caller.
- app_url: str = Field(default="https://app.strix.ai", alias="STRIX_APP_URL")
-
-
-class Settings(BaseSettings):
- model_config = _BASE_CONFIG
-
- llm: LlmSettings = Field(default_factory=LlmSettings)
- dedupe: DedupeSettings = Field(default_factory=DedupeSettings)
- runtime: RuntimeSettings = Field(default_factory=RuntimeSettings)
- context: ContextSettings = Field(default_factory=ContextSettings)
- telemetry: TelemetrySettings = Field(default_factory=TelemetrySettings)
- integrations: IntegrationSettings = Field(default_factory=IntegrationSettings)
- viewer: ViewerSettings = Field(default_factory=ViewerSettings)
+"""Strix application settings — pydantic-settings powered."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import AliasChoices, Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+ReasoningEffort = Literal["none", "minimal", "low", "medium", "high", "xhigh", "max"]
+
+DEFAULT_MAX_TURNS = 500
+
+_BASE_CONFIG = SettingsConfigDict(
+ case_sensitive=False,
+ populate_by_name=True,
+ extra="ignore",
+)
+
+
+class LlmSettings(BaseSettings):
+ model_config = _BASE_CONFIG
+
+ model: str | None = Field(default=None, alias="STRIX_LLM")
+ api_key: str | None = Field(
+ default=None,
+ validation_alias=AliasChoices("LLM_API_KEY", "OPENAI_API_KEY"),
+ repr=False,
+ )
+ api_base: str | None = Field(
+ default=None,
+ validation_alias=AliasChoices(
+ "LLM_API_BASE",
+ "OPENAI_API_BASE",
+ "OPENAI_BASE_URL",
+ "LITELLM_BASE_URL",
+ "OLLAMA_API_BASE",
+ ),
+ )
+ extra_headers: dict[str, str] | None = Field(
+ default=None,
+ alias="LLM_EXTRA_HEADERS",
+ repr=False,
+ )
+ reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
+ force_required_tool_choice: bool = Field(
+ default=False,
+ alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
+ )
+ prompt_cache: bool = Field(
+ default=True,
+ alias="STRIX_PROMPT_CACHE",
+ )
+ disable_streaming: bool = Field(
+ default=False,
+ alias="LLM_DISABLE_STREAMING",
+ )
+ timeout: int = Field(default=300, alias="LLM_TIMEOUT")
+ stream_idle_timeout: int = Field(default=300, ge=0, alias="LLM_STREAM_IDLE_TIMEOUT")
+ max_tool_calls_per_turn: int = Field(
+ default=32,
+ ge=0,
+ alias="LLM_MAX_TOOL_CALLS_PER_TURN",
+ )
+
+
+class DedupeSettings(BaseSettings):
+ model_config = _BASE_CONFIG
+
+ model: str | None = Field(default=None, alias="STRIX_DEDUPE_MODEL")
+ reasoning_effort: ReasoningEffort | None = Field(
+ default=None,
+ alias="STRIX_DEDUPE_REASONING_EFFORT",
+ )
+ api_key: str | None = Field(default=None, alias="DEDUPE_LLM_API_KEY", repr=False)
+ api_base: str | None = Field(default=None, alias="DEDUPE_LLM_API_BASE")
+ extra_headers: dict[str, str] | None = Field(
+ default=None,
+ alias="DEDUPE_LLM_EXTRA_HEADERS",
+ repr=False,
+ )
+
+
+class ContextSettings(BaseSettings):
+ """Context-window management: per-tool-output caps and history compaction."""
+
+ model_config = _BASE_CONFIG
+
+ auto_compact: bool = Field(default=True, alias="STRIX_CONTEXT_AUTO_COMPACT")
+ compact_buffer_tokens: int = Field(default=20_000, gt=0, alias="STRIX_CONTEXT_BUFFER_TOKENS")
+ keep_tokens: int = Field(default=8_000, gt=0, alias="STRIX_CONTEXT_KEEP_TOKENS")
+ fallback_context_tokens: int = Field(
+ default=200_000, gt=0, alias="STRIX_CONTEXT_FALLBACK_TOKENS"
+ )
+ summary_max_tokens: int = Field(default=4_096, gt=0, alias="STRIX_CONTEXT_SUMMARY_TOKENS")
+ tool_output_max_tokens: int = Field(default=8_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_TOKENS")
+ tool_output_max_lines: int = Field(default=2_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_LINES")
+ # Floor above the truncation-notice size so a preview always fits.
+ tool_output_max_bytes: int = Field(
+ default=50 * 1024, ge=1024, alias="STRIX_TOOL_OUTPUT_MAX_BYTES"
+ )
+
+
+class RuntimeSettings(BaseSettings):
+ model_config = _BASE_CONFIG
+
+ image: str = Field(
+ default="ghcr.io/usestrix/strix-sandbox:1.3.0",
+ alias="STRIX_IMAGE",
+ )
+ backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
+ caido_boot_wait_s: int = Field(default=300, gt=0, alias="STRIX_CAIDO_BOOT_WAIT_S")
+ # Max screenshot/image tool outputs kept live per agent context (0 = none).
+ max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES")
+
+
+class TelemetrySettings(BaseSettings):
+ model_config = _BASE_CONFIG
+
+ enabled: bool = Field(default=True, alias="STRIX_TELEMETRY")
+
+
+class IntegrationSettings(BaseSettings):
+ model_config = _BASE_CONFIG
+
+ perplexity_api_key: str | None = Field(
+ default=None,
+ alias="PERPLEXITY_API_KEY",
+ repr=False,
+ )
+ postman_api_key: str | None = Field(
+ default=None,
+ alias="POSTMAN_API_KEY",
+ repr=False,
+ )
+
+
+class ViewerSettings(BaseSettings):
+ model_config = _BASE_CONFIG
+
+ # Base URL of the Strix relay the local viewer proxies to for email
+ # verification and encrypted report delivery. The browser never talks to
+ # the relay directly; the local server is the only caller.
+ app_url: str = Field(default="https://app.strix.ai", alias="STRIX_APP_URL")
+
+
+class Settings(BaseSettings):
+ model_config = _BASE_CONFIG
+
+ llm: LlmSettings = Field(default_factory=LlmSettings)
+ dedupe: DedupeSettings = Field(default_factory=DedupeSettings)
+ runtime: RuntimeSettings = Field(default_factory=RuntimeSettings)
+ context: ContextSettings = Field(default_factory=ContextSettings)
+ telemetry: TelemetrySettings = Field(default_factory=TelemetrySettings)
+ integrations: IntegrationSettings = Field(default_factory=IntegrationSettings)
+ viewer: ViewerSettings = Field(default_factory=ViewerSettings)
diff --git a/strix/runtime/caido_bootstrap.py b/strix/runtime/caido_bootstrap.py
index 0b9ad5b17..545c29596 100644
--- a/strix/runtime/caido_bootstrap.py
+++ b/strix/runtime/caido_bootstrap.py
@@ -1,109 +1,137 @@
-"""Caido client bootstrap.
-
-The Caido CLI runs as an in-container sidecar listening on
-``127.0.0.1:48080`` *inside* the sandbox. We grab a guest token by
-``session.exec()``-ing curl from inside the container, then construct
-a host-side :class:`caido_sdk_client.Client` against the runtime's
-exposed-port URL for all subsequent SDK calls.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import json
-import logging
-from typing import TYPE_CHECKING
-
-from caido_sdk_client import Client, TokenAuthOptions
-from caido_sdk_client.types import CreateProjectOptions
-
-
-if TYPE_CHECKING:
- from agents.sandbox.session import BaseSandboxSession
-
-
-logger = logging.getLogger(__name__)
-
-
-_LOGIN_AS_GUEST_BODY = (
- '{"query":"mutation LoginAsGuest { loginAsGuest { token { accessToken } } }"}'
-)
-
-
-async def _login_as_guest(
- session: BaseSandboxSession,
- *,
- container_url: str,
- attempts: int = 10,
-) -> str:
- """``session.exec`` curl to fetch a guest token; retry until ready.
-
- Caido's GraphQL listener may not be up the instant the container
- starts. The retry loop also doubles as the Caido readiness probe —
- no separate TCP healthcheck needed.
- """
- last_err: str | None = None
- for i in range(1, attempts + 1):
- result = await session.exec(
- "curl",
- "-fsS",
- "-X",
- "POST",
- "-H",
- "Content-Type: application/json",
- "-d",
- _LOGIN_AS_GUEST_BODY,
- f"{container_url}/graphql",
- timeout=15,
- )
- if result.ok():
- try:
- payload = json.loads(result.stdout)
- token = (
- payload.get("data", {})
- .get("loginAsGuest", {})
- .get("token", {})
- .get("accessToken")
- )
- if token:
- return str(token)
- last_err = f"loginAsGuest returned no token: {payload}"
- except json.JSONDecodeError as exc:
- last_err = f"unparseable response: {exc}: {result.stdout!r}"
- else:
- stderr = result.stderr.decode("utf-8", errors="replace")[:200]
- last_err = f"curl exit {result.exit_code}: {stderr}"
- logger.debug("loginAsGuest attempt %d/%d failed: %s", i, attempts, last_err)
- await asyncio.sleep(min(2.0 * i, 8.0))
-
- raise RuntimeError(f"loginAsGuest failed after {attempts} attempts: {last_err}")
-
-
-async def bootstrap_caido(
- session: BaseSandboxSession,
- *,
- host_url: str,
- container_url: str,
-) -> Client:
- """Connect to the in-container Caido sidecar and select a fresh project."""
- logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url)
-
- access_token = await _login_as_guest(session, container_url=container_url)
-
- client = Client(host_url, auth=TokenAuthOptions(token=access_token))
- await client.connect()
-
- try:
- project = await client.project.create(
- CreateProjectOptions(name="sandbox", temporary=True),
- )
- await client.project.select(project.id)
- except BaseException:
- # The connected client never reaches the session bundle if project
- # setup fails, so close it here to avoid leaking the transport.
- with contextlib.suppress(Exception):
- await client.aclose()
- raise
- logger.info("Caido project selected: %s", project.id)
- return client
+"""Caido client bootstrap.
+
+The Caido CLI runs as an in-container sidecar listening on
+``127.0.0.1:48080`` *inside* the sandbox. We grab a guest token by
+``session.exec()``-ing curl from inside the container, then construct
+a host-side :class:`caido_sdk_client.Client` against the runtime's
+exposed-port URL for all subsequent SDK calls.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import json
+import logging
+import math
+import time
+from typing import TYPE_CHECKING
+
+from caido_sdk_client import Client, TokenAuthOptions
+from caido_sdk_client.types import CreateProjectOptions
+
+
+if TYPE_CHECKING:
+ from agents.sandbox.session import BaseSandboxSession
+
+
+logger = logging.getLogger(__name__)
+
+
+_LOGIN_AS_GUEST_BODY = (
+ '{"query":"mutation LoginAsGuest { loginAsGuest { token { accessToken } } }"}'
+)
+
+
+async def _login_as_guest(
+ session: BaseSandboxSession,
+ *,
+ container_url: str,
+ wait_timeout_s: float,
+) -> str:
+ """``session.exec`` curl to fetch a guest token until the deadline.
+
+ Caido's GraphQL listener may not be up the instant the container
+ starts. The retry loop also doubles as the Caido readiness probe —
+ no separate TCP healthcheck needed.
+ """
+ started_at = time.monotonic()
+ deadline = started_at + wait_timeout_s
+ last_err: str | None = None
+ attempts = 0
+
+ while attempts == 0 or time.monotonic() < deadline:
+ attempts += 1
+ remaining = deadline - time.monotonic()
+ exec_timeout = max(1, min(15, math.ceil(remaining)))
+ result = await session.exec(
+ "curl",
+ "-fsS",
+ "-X",
+ "POST",
+ "-H",
+ "Content-Type: application/json",
+ "-d",
+ _LOGIN_AS_GUEST_BODY,
+ f"{container_url}/graphql",
+ timeout=exec_timeout,
+ )
+ if result.ok():
+ try:
+ payload = json.loads(result.stdout)
+ token = (
+ payload.get("data", {})
+ .get("loginAsGuest", {})
+ .get("token", {})
+ .get("accessToken")
+ )
+ if token:
+ return str(token)
+ last_err = f"loginAsGuest returned no token: {payload}"
+ except json.JSONDecodeError as exc:
+ last_err = f"unparseable response: {exc}: {result.stdout!r}"
+ else:
+ stderr = result.stderr.decode("utf-8", errors="replace")[:200]
+ last_err = f"curl exit {result.exit_code}: {stderr}"
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ break
+ retry_delay = min(2.0 * attempts, 8.0, remaining)
+ logger.debug(
+ "loginAsGuest attempt %d failed after %.1fs; retrying in %.1fs (%.1fs remaining): %s",
+ attempts,
+ time.monotonic() - started_at,
+ retry_delay,
+ remaining,
+ last_err,
+ )
+ await asyncio.sleep(retry_delay)
+
+ raise RuntimeError(
+ f"loginAsGuest not ready after {float(wait_timeout_s):g} seconds "
+ f"({attempts} attempts): {last_err}"
+ )
+
+
+async def bootstrap_caido(
+ session: BaseSandboxSession,
+ *,
+ host_url: str,
+ container_url: str,
+ wait_timeout_s: float,
+) -> Client:
+ """Connect to the in-container Caido sidecar and select a fresh project."""
+ logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url)
+
+ access_token = await _login_as_guest(
+ session,
+ container_url=container_url,
+ wait_timeout_s=wait_timeout_s,
+ )
+
+ client = Client(host_url, auth=TokenAuthOptions(token=access_token))
+ await client.connect()
+
+ try:
+ project = await client.project.create(
+ CreateProjectOptions(name="sandbox", temporary=True),
+ )
+ await client.project.select(project.id)
+ except BaseException:
+ # The connected client never reaches the session bundle if project
+ # setup fails, so close it here to avoid leaking the transport.
+ with contextlib.suppress(Exception):
+ await client.aclose()
+ raise
+ logger.info("Caido project selected: %s", project.id)
+ return client
diff --git a/strix/runtime/session_manager.py b/strix/runtime/session_manager.py
index 4b61d7351..d3bcd08d7 100644
--- a/strix/runtime/session_manager.py
+++ b/strix/runtime/session_manager.py
@@ -1,233 +1,235 @@
-"""Per-scan sandbox session lifecycle."""
-
-from __future__ import annotations
-
-import logging
-import os
-import sys
-from pathlib import Path
-from typing import TYPE_CHECKING, Any
-
-from agents.sandbox.entries import BaseEntry, LocalDir
-from agents.sandbox.manifest import Environment, Manifest
-
-from strix.config import load_settings
-from strix.runtime.backends import backend_supports_bind_mounts, get_backend
-from strix.runtime.caido_bootstrap import bootstrap_caido
-
-
-if TYPE_CHECKING:
- from strix.runtime.status import StatusSink
-
-
-logger = logging.getLogger(__name__)
-
-
-# In-container Caido sidecar port (matches the image's caido-cli bind).
-_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"
-
-_PROTECTED_METADATA_NAMES = (".git", ".agents", ".codex")
-
-
-def _host_identity_env() -> dict[str, str]:
- # Read the platform through a local so it is not narrowed to whichever OS is
- # type-checking: comparing sys.platform directly makes one of these branches
- # statically dead, and which one flips between Linux and macOS.
- platform_name: str = sys.platform
- if platform_name != "linux":
- return {}
- # Bind-mount ownership only needs mapping on Linux, where the container uid
- # must match the host's.
- return {"STRIX_HOST_UID": str(os.getuid()), "STRIX_HOST_GID": str(os.getgid())}
-
-
-def build_bind_mounts(local_sources: list[dict[str, Any]]) -> list[dict[str, Any]]:
- 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()
- target = f"{_WORKSPACE_ROOT}/{ws_subdir}"
- bind_mounts.append({"source": str(resolved), "target": target, "read_only": False})
- if src.get("protect_metadata"):
- bind_mounts.extend(_metadata_mounts(resolved, target))
- return bind_mounts
-
-
-def build_manifest_entries(local_sources: list[dict[str, Any]]) -> dict[str | Path, BaseEntry]:
- 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())
- return entries
-
-
-def _metadata_mounts(tree: Path, target: str) -> list[dict[str, Any]]:
- mounts: list[dict[str, Any]] = []
- for name in _PROTECTED_METADATA_NAMES:
- metadata = tree / name
- if not metadata.is_dir() and not metadata.is_file():
- continue
- if not metadata.resolve().is_relative_to(tree):
- continue
- mounts.append({"source": str(metadata), "target": f"{target}/{name}", "read_only": True})
- gitdir = _gitdir_from_pointer(metadata) if metadata.is_file() else None
- if gitdir is not None and gitdir.exists() and gitdir.is_relative_to(tree):
- relative = gitdir.relative_to(tree).as_posix()
- mounts.append(
- {"source": str(gitdir), "target": f"{target}/{relative}", "read_only": True}
- )
- return mounts
-
-
-def _gitdir_from_pointer(git_file: Path) -> Path | None:
- try:
- content = git_file.read_text(encoding="utf-8", errors="replace")
- except OSError:
- return None
- for line in content.splitlines():
- prefix, _, value = line.partition(":")
- if prefix.strip() == "gitdir" and value.strip():
- candidate = Path(value.strip()).expanduser()
- if not candidate.is_absolute():
- candidate = git_file.parent / candidate
- return candidate.resolve()
- return None
-
-
-async def create_or_reuse(
- scan_id: str,
- *,
- image: str,
- local_sources: list[dict[str, Any]],
- status_sink: StatusSink | None = None,
-) -> dict[str, Any]:
- """Return the existing session bundle for ``scan_id`` or create a new one.
-
- Each ``local_sources`` entry exposes its host ``source_path`` at
- ``/workspace/`` inside the container.
- """
-
- def report(phase: str) -> None:
- if status_sink is not None:
- status_sink(phase)
-
- cached = _SESSION_CACHE.get(scan_id)
- if cached is not None:
- logger.info("Reusing existing sandbox session for scan %s", scan_id)
- return cached
-
- backend_name = load_settings().runtime.backend
- backend = get_backend(backend_name)
-
- if backend_supports_bind_mounts(backend_name):
- bind_mounts = build_bind_mounts(local_sources)
- entries: dict[str | Path, BaseEntry] = {}
- else:
- bind_mounts = []
- entries = build_manifest_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.)
- # picks up these env vars automatically. ``NO_PROXY`` keeps the
- # agent-browser CDP daemon's localhost traffic from looping back
- # through Caido.
- container_caido_url = f"http://127.0.0.1:{_CONTAINER_CAIDO_PORT}"
- manifest = Manifest(
- entries=entries,
- environment=Environment(
- value={
- "PYTHONUNBUFFERED": "1",
- "HOST_GATEWAY": "host.docker.internal",
- **_host_identity_env(),
- "http_proxy": container_caido_url,
- "https_proxy": container_caido_url,
- "ALL_PROXY": container_caido_url,
- "NO_PROXY": "localhost,127.0.0.1",
- },
- ),
- )
-
- logger.info(
- "Creating sandbox session for scan %s (backend=%s, image=%s)",
- scan_id,
- backend_name,
- image,
- )
- report("Starting sandbox container")
- client, session = await backend(
- image=image,
- manifest=manifest,
- exposed_ports=(_CONTAINER_CAIDO_PORT,),
- bind_mounts=bind_mounts,
- )
-
- report("Setting up the proxy")
- caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
- scheme = "https" if caido_endpoint.tls else "http"
- host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
- logger.debug("Caido host endpoint resolved: %s", host_caido_url)
-
- caido_client = await bootstrap_caido(
- session,
- host_url=host_caido_url,
- container_url=container_caido_url,
- )
-
- bundle = {
- "client": client,
- "session": session,
- "caido_client": caido_client,
- }
- _SESSION_CACHE[scan_id] = bundle
- logger.info("Sandbox session for scan %s ready and cached", scan_id)
- return bundle
-
-
-async def cleanup(scan_id: str) -> None:
- """Tear down ``scan_id``'s container and drop its cache entry.
-
- Best-effort: any error during ``client.delete`` is logged and
- swallowed. We never want a cleanup failure to prevent the next
- scan from starting; the worst case is a stranded container that
- Docker's normal reaping will catch on next ``docker prune``.
- """
- bundle = _SESSION_CACHE.pop(scan_id, None)
- if bundle is None:
- logger.debug("cleanup(%s): no cached session", scan_id)
- return
-
- caido_client = bundle.get("caido_client")
- if caido_client is not None:
- try:
- await caido_client.aclose()
- except Exception: # noqa: BLE001
- logger.debug("cleanup(%s): caido_client.aclose() raised", scan_id, exc_info=True)
-
- client = bundle["client"]
- try:
- await client.delete(bundle["session"])
- logger.info("Cleaned up sandbox session for scan %s", scan_id)
- except Exception:
- logger.exception(
- "cleanup(%s): client.delete raised; container may need manual reaping",
- scan_id,
- )
-
- docker_client = getattr(client, "docker_client", None)
- if docker_client is not None:
- try:
- docker_client.close()
- except Exception: # noqa: BLE001
- logger.debug("cleanup(%s): docker_client.close() raised", scan_id, exc_info=True)
+"""Per-scan sandbox session lifecycle."""
+
+from __future__ import annotations
+
+import logging
+import os
+import sys
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from agents.sandbox.entries import BaseEntry, LocalDir
+from agents.sandbox.manifest import Environment, Manifest
+
+from strix.config import load_settings
+from strix.runtime.backends import backend_supports_bind_mounts, get_backend
+from strix.runtime.caido_bootstrap import bootstrap_caido
+
+
+if TYPE_CHECKING:
+ from strix.runtime.status import StatusSink
+
+
+logger = logging.getLogger(__name__)
+
+
+# In-container Caido sidecar port (matches the image's caido-cli bind).
+_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"
+
+_PROTECTED_METADATA_NAMES = (".git", ".agents", ".codex")
+
+
+def _host_identity_env() -> dict[str, str]:
+ # Read the platform through a local so it is not narrowed to whichever OS is
+ # type-checking: comparing sys.platform directly makes one of these branches
+ # statically dead, and which one flips between Linux and macOS.
+ platform_name: str = sys.platform
+ if platform_name != "linux":
+ return {}
+ # Bind-mount ownership only needs mapping on Linux, where the container uid
+ # must match the host's.
+ return {"STRIX_HOST_UID": str(os.getuid()), "STRIX_HOST_GID": str(os.getgid())}
+
+
+def build_bind_mounts(local_sources: list[dict[str, Any]]) -> list[dict[str, Any]]:
+ 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()
+ target = f"{_WORKSPACE_ROOT}/{ws_subdir}"
+ bind_mounts.append({"source": str(resolved), "target": target, "read_only": False})
+ if src.get("protect_metadata"):
+ bind_mounts.extend(_metadata_mounts(resolved, target))
+ return bind_mounts
+
+
+def build_manifest_entries(local_sources: list[dict[str, Any]]) -> dict[str | Path, BaseEntry]:
+ 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())
+ return entries
+
+
+def _metadata_mounts(tree: Path, target: str) -> list[dict[str, Any]]:
+ mounts: list[dict[str, Any]] = []
+ for name in _PROTECTED_METADATA_NAMES:
+ metadata = tree / name
+ if not metadata.is_dir() and not metadata.is_file():
+ continue
+ if not metadata.resolve().is_relative_to(tree):
+ continue
+ mounts.append({"source": str(metadata), "target": f"{target}/{name}", "read_only": True})
+ gitdir = _gitdir_from_pointer(metadata) if metadata.is_file() else None
+ if gitdir is not None and gitdir.exists() and gitdir.is_relative_to(tree):
+ relative = gitdir.relative_to(tree).as_posix()
+ mounts.append(
+ {"source": str(gitdir), "target": f"{target}/{relative}", "read_only": True}
+ )
+ return mounts
+
+
+def _gitdir_from_pointer(git_file: Path) -> Path | None:
+ try:
+ content = git_file.read_text(encoding="utf-8", errors="replace")
+ except OSError:
+ return None
+ for line in content.splitlines():
+ prefix, _, value = line.partition(":")
+ if prefix.strip() == "gitdir" and value.strip():
+ candidate = Path(value.strip()).expanduser()
+ if not candidate.is_absolute():
+ candidate = git_file.parent / candidate
+ return candidate.resolve()
+ return None
+
+
+async def create_or_reuse(
+ scan_id: str,
+ *,
+ image: str,
+ local_sources: list[dict[str, Any]],
+ status_sink: StatusSink | None = None,
+) -> dict[str, Any]:
+ """Return the existing session bundle for ``scan_id`` or create a new one.
+
+ Each ``local_sources`` entry exposes its host ``source_path`` at
+ ``/workspace/`` inside the container.
+ """
+
+ def report(phase: str) -> None:
+ if status_sink is not None:
+ status_sink(phase)
+
+ cached = _SESSION_CACHE.get(scan_id)
+ if cached is not None:
+ logger.info("Reusing existing sandbox session for scan %s", scan_id)
+ return cached
+
+ runtime_settings = load_settings().runtime
+ backend_name = runtime_settings.backend
+ backend = get_backend(backend_name)
+
+ if backend_supports_bind_mounts(backend_name):
+ bind_mounts = build_bind_mounts(local_sources)
+ entries: dict[str | Path, BaseEntry] = {}
+ else:
+ bind_mounts = []
+ entries = build_manifest_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.)
+ # picks up these env vars automatically. ``NO_PROXY`` keeps the
+ # agent-browser CDP daemon's localhost traffic from looping back
+ # through Caido.
+ container_caido_url = f"http://127.0.0.1:{_CONTAINER_CAIDO_PORT}"
+ manifest = Manifest(
+ entries=entries,
+ environment=Environment(
+ value={
+ "PYTHONUNBUFFERED": "1",
+ "HOST_GATEWAY": "host.docker.internal",
+ **_host_identity_env(),
+ "http_proxy": container_caido_url,
+ "https_proxy": container_caido_url,
+ "ALL_PROXY": container_caido_url,
+ "NO_PROXY": "localhost,127.0.0.1",
+ },
+ ),
+ )
+
+ logger.info(
+ "Creating sandbox session for scan %s (backend=%s, image=%s)",
+ scan_id,
+ backend_name,
+ image,
+ )
+ report("Starting sandbox container")
+ client, session = await backend(
+ image=image,
+ manifest=manifest,
+ exposed_ports=(_CONTAINER_CAIDO_PORT,),
+ bind_mounts=bind_mounts,
+ )
+
+ report("Setting up the proxy")
+ caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
+ scheme = "https" if caido_endpoint.tls else "http"
+ host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
+ logger.debug("Caido host endpoint resolved: %s", host_caido_url)
+
+ caido_client = await bootstrap_caido(
+ session,
+ host_url=host_caido_url,
+ container_url=container_caido_url,
+ wait_timeout_s=runtime_settings.caido_boot_wait_s,
+ )
+
+ bundle = {
+ "client": client,
+ "session": session,
+ "caido_client": caido_client,
+ }
+ _SESSION_CACHE[scan_id] = bundle
+ logger.info("Sandbox session for scan %s ready and cached", scan_id)
+ return bundle
+
+
+async def cleanup(scan_id: str) -> None:
+ """Tear down ``scan_id``'s container and drop its cache entry.
+
+ Best-effort: any error during ``client.delete`` is logged and
+ swallowed. We never want a cleanup failure to prevent the next
+ scan from starting; the worst case is a stranded container that
+ Docker's normal reaping will catch on next ``docker prune``.
+ """
+ bundle = _SESSION_CACHE.pop(scan_id, None)
+ if bundle is None:
+ logger.debug("cleanup(%s): no cached session", scan_id)
+ return
+
+ caido_client = bundle.get("caido_client")
+ if caido_client is not None:
+ try:
+ await caido_client.aclose()
+ except Exception: # noqa: BLE001
+ logger.debug("cleanup(%s): caido_client.aclose() raised", scan_id, exc_info=True)
+
+ client = bundle["client"]
+ try:
+ await client.delete(bundle["session"])
+ logger.info("Cleaned up sandbox session for scan %s", scan_id)
+ except Exception:
+ logger.exception(
+ "cleanup(%s): client.delete raised; container may need manual reaping",
+ scan_id,
+ )
+
+ docker_client = getattr(client, "docker_client", None)
+ if docker_client is not None:
+ try:
+ docker_client.close()
+ except Exception: # noqa: BLE001
+ logger.debug("cleanup(%s): docker_client.close() raised", scan_id, exc_info=True)
diff --git a/tests/test_caido_bootstrap.py b/tests/test_caido_bootstrap.py
new file mode 100644
index 000000000..5e63e18b3
--- /dev/null
+++ b/tests/test_caido_bootstrap.py
@@ -0,0 +1,104 @@
+"""Tests for the Caido readiness deadline."""
+
+# pyright: reportPrivateUsage=false
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, cast
+
+import pytest
+
+from strix.runtime import caido_bootstrap
+
+
+if TYPE_CHECKING:
+ from agents.sandbox.session import BaseSandboxSession
+
+
+@dataclass
+class _ExecResult:
+ success: bool
+ stdout: str = ""
+ stderr: bytes = b"connection refused"
+ exit_code: int = 7
+
+ def ok(self) -> bool:
+ return self.success
+
+
+class _Session:
+ def __init__(self, results: list[_ExecResult]) -> None:
+ self._results = iter(results)
+ self.calls: list[tuple[tuple[str, ...], dict[str, Any]]] = []
+
+ async def exec(self, *args: str, **kwargs: Any) -> _ExecResult:
+ self.calls.append((args, kwargs))
+ return next(self._results)
+
+
+class _Clock:
+ def __init__(self) -> None:
+ self.now = 0.0
+ self.sleeps: list[float] = []
+
+ def monotonic(self) -> float:
+ return self.now
+
+ async def sleep(self, delay: float) -> None:
+ self.sleeps.append(delay)
+ self.now += delay
+
+
+def _install_clock(monkeypatch: pytest.MonkeyPatch) -> _Clock:
+ clock = _Clock()
+ monkeypatch.setattr("strix.runtime.caido_bootstrap.time.monotonic", clock.monotonic)
+ monkeypatch.setattr("strix.runtime.caido_bootstrap.asyncio.sleep", clock.sleep)
+ return clock
+
+
+@pytest.mark.asyncio
+async def test_login_as_guest_retries_until_ready(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ clock = _install_clock(monkeypatch)
+ session = _Session(
+ [
+ _ExecResult(success=False),
+ _ExecResult(
+ success=True,
+ stdout='{"data":{"loginAsGuest":{"token":{"accessToken":"token"}}}}',
+ ),
+ ]
+ )
+
+ token = await caido_bootstrap._login_as_guest(
+ cast("BaseSandboxSession", session),
+ container_url="http://127.0.0.1:48080",
+ wait_timeout_s=10,
+ )
+
+ assert token == "token" # noqa: S105 - synthetic token fixture
+ assert clock.sleeps == [2.0]
+ assert [call[1]["timeout"] for call in session.calls] == [10, 8]
+
+
+@pytest.mark.asyncio
+async def test_login_as_guest_stops_at_deadline(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ clock = _install_clock(monkeypatch)
+ session = _Session([_ExecResult(success=False), _ExecResult(success=False)])
+
+ with pytest.raises(
+ RuntimeError,
+ match=r"loginAsGuest not ready after 5 seconds \(2 attempts\)",
+ ):
+ await caido_bootstrap._login_as_guest(
+ cast("BaseSandboxSession", session),
+ container_url="http://127.0.0.1:48080",
+ wait_timeout_s=5,
+ )
+
+ assert clock.sleeps == [2.0, 3.0]
+ assert [call[1]["timeout"] for call in session.calls] == [5, 3]
diff --git a/tests/test_config_loader.py b/tests/test_config_loader.py
index e83ab1192..53510e9ef 100644
--- a/tests/test_config_loader.py
+++ b/tests/test_config_loader.py
@@ -1,218 +1,225 @@
-"""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, ValidationError
-from pydantic.fields import FieldInfo
-
-from strix.config import loader
-from strix.config.settings import ContextSettings
-
-
-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",
- "STRIX_FORCE_REQUIRED_TOOL_CHOICE",
- "LLM_TIMEOUT",
- "PERPLEXITY_API_KEY",
- # RuntimeSettings
- "STRIX_IMAGE",
- "STRIX_RUNTIME_BACKEND",
- # 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) == {}
-
-
-def test_read_json_overrides_env_wins_across_field_aliases(
- tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
- # api_key resolves from AliasChoices("LLM_API_KEY", "OPENAI_API_KEY"). The env
- # sets one alias while the persisted file holds another. Env must still win, so
- # the stale file value must not be surfaced as an init kwarg (which outranks env).
- monkeypatch.setenv("OPENAI_API_KEY", "sk-env")
- path = tmp_path / "cli-config.json"
- path.write_text(json.dumps({"env": {"LLM_API_KEY": "sk-file"}}), encoding="utf-8")
- assert loader._read_json_overrides(path) == {}
-
-
-def test_read_json_overrides_env_wins_case_insensitively(
- tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
- # Settings use case_sensitive=False, so a lowercase env var also counts as set.
- 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")
- assert loader._read_json_overrides(path) == {}
-
-
-def test_read_json_overrides_uses_json_when_no_alias_in_environ(tmp_path: Path) -> None:
- # No alias of api_key is set in the environment -> the file value is used, even
- # when it is stored under a non-first alias.
- path = tmp_path / "cli-config.json"
- path.write_text(json.dumps({"env": {"OPENAI_API_KEY": "sk-file"}}), encoding="utf-8")
- assert loader._read_json_overrides(path) == {"llm": {"api_key": "sk-file"}}
-
-
-def test_tool_output_max_bytes_rejects_sub_notice_values() -> None:
- with pytest.raises(ValidationError):
- ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=64)
-
-
-def test_tool_output_max_bytes_accepts_floor() -> None:
- assert ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=1024).tool_output_max_bytes == 1024
-
-
-# --------------------------------------------------------------------------- #
-# _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 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, ValidationError
+from pydantic.fields import FieldInfo
+
+from strix.config import loader
+from strix.config.settings import ContextSettings
+
+
+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",
+ "STRIX_FORCE_REQUIRED_TOOL_CHOICE",
+ "LLM_TIMEOUT",
+ "PERPLEXITY_API_KEY",
+ # RuntimeSettings
+ "STRIX_IMAGE",
+ "STRIX_RUNTIME_BACKEND",
+ "STRIX_CAIDO_BOOT_WAIT_S",
+ # 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) == {}
+
+
+def test_read_json_overrides_env_wins_across_field_aliases(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ # api_key resolves from AliasChoices("LLM_API_KEY", "OPENAI_API_KEY"). The env
+ # sets one alias while the persisted file holds another. Env must still win, so
+ # the stale file value must not be surfaced as an init kwarg (which outranks env).
+ monkeypatch.setenv("OPENAI_API_KEY", "sk-env")
+ path = tmp_path / "cli-config.json"
+ path.write_text(json.dumps({"env": {"LLM_API_KEY": "sk-file"}}), encoding="utf-8")
+ assert loader._read_json_overrides(path) == {}
+
+
+def test_read_json_overrides_env_wins_case_insensitively(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ # Settings use case_sensitive=False, so a lowercase env var also counts as set.
+ 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")
+ assert loader._read_json_overrides(path) == {}
+
+
+def test_read_json_overrides_uses_json_when_no_alias_in_environ(tmp_path: Path) -> None:
+ # No alias of api_key is set in the environment -> the file value is used, even
+ # when it is stored under a non-first alias.
+ path = tmp_path / "cli-config.json"
+ path.write_text(json.dumps({"env": {"OPENAI_API_KEY": "sk-file"}}), encoding="utf-8")
+ assert loader._read_json_overrides(path) == {"llm": {"api_key": "sk-file"}}
+
+
+def test_tool_output_max_bytes_rejects_sub_notice_values() -> None:
+ with pytest.raises(ValidationError):
+ ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=64)
+
+
+def test_tool_output_max_bytes_accepts_floor() -> None:
+ assert ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=1024).tool_output_max_bytes == 1024
+
+
+def test_runtime_caido_boot_wait_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
+ monkeypatch.setenv("STRIX_CAIDO_BOOT_WAIT_S", "900")
+
+ assert loader.load_settings().runtime.caido_boot_wait_s == 900
+
+
+# --------------------------------------------------------------------------- #
+# _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