chore: release v1.6.1

README.md

@@ -273,6 +273,7 @@ export LLM_API_KEY="your-api-key"
 
 # Optional
 export LLM_API_BASE="your-api-base-url"  # if using a local model, e.g. Ollama, LMStudio
+export PERPLEXITY_API_KEY="your-api-key"  # for search capabilities
 ```
 
 > [!NOTE]

docs/advanced/configuration.mdx

@@ -80,22 +80,6 @@ affecting the agents that do the actual testing.
   API key for Perplexity AI. Enables real-time web search during scans for OSINT and vulnerability research.
 </ParamField>
 
-<ParamField path="EXA_API_KEY" type="string">
-  API key for Exa. Enables real-time web search through the Exa `/search` endpoint. Exa also powers the `web_get_contents` tool, which fetches the full text of a page through the Exa `/contents` endpoint. This is the preferred web search provider.
-</ParamField>
-
-<ParamField path="STRIX_WEB_SEARCH_PROVIDER" default="auto" type="string">
-  Web search provider: `auto`, `perplexity`, or `exa`. With `auto`, Strix uses Exa when `EXA_API_KEY` is set, and Perplexity otherwise. Set an explicit provider to pin one when you configure both keys.
-</ParamField>
-
-<ParamField path="STRIX_EXA_SEARCH_TYPE" default="auto" type="string">
-  Exa search mode: `auto`, `fast`, `instant`, `deep-lite`, `deep`, or `deep-reasoning`. Lower modes return results faster. Higher modes plan across more steps and take more time. This setting applies only to the Exa provider.
-</ParamField>
-
-<ParamField path="STRIX_EXA_NUM_RESULTS" default="5" type="integer">
-  Number of Exa results to return, from `1` to `100`. Each result includes a title, a URL, and a short security-focused summary. To read a full page, the agent calls `web_get_contents` with the result URL. This setting applies only to the Exa provider.
-</ParamField>
-
 <ParamField path="POSTMAN_API_KEY" type="string">
   Postman API key (`PMAK-…`). Enables fetching Postman collections by id as a target (`postman://<collection-uid>`), and Postman environments (`postman://<collection-uid>?env=<environment-uid>`) to resolve collection variables. Not needed when passing a local collection export file.
 </ParamField>
@@ -175,8 +159,7 @@ strix --target ./app --config /path/to/config.json
 export STRIX_LLM="openrouter/z-ai/glm-5.3"
 export LLM_API_KEY="sk-..."
 
-# Optional: Enable web search (Exa preferred, Perplexity supported)
-export EXA_API_KEY="..."
+# Optional: Enable web search
 export PERPLEXITY_API_KEY="pplx-..."
 
 # Optional: Custom timeouts

docs/tools/overview.mdx

@@ -28,6 +28,6 @@ Strix agents use specialized tools to test your applications like a real penetra
 | -------------- | ---------------------------------------- |
 | Python Runtime | Write and execute custom exploit scripts |
 | File Editor    | Read and modify source code              |
-| Web Search     | Real-time OSINT with Exa or Perplexity   |
+| Web Search     | Real-time OSINT via Perplexity           |
 | Notes          | Document findings during the scan        |
 | Reporting      | Generate vulnerability reports with PoCs |

strix/agents/factory.py

@@ -69,7 +69,7 @@ from strix.tools.todo.tools import (
     mark_todo_pending,
     update_todo,
 )
-from strix.tools.web_search.tool import web_get_contents, web_search
+from strix.tools.web_search.tool import web_search
 
 
 if TYPE_CHECKING:
@@ -579,7 +579,6 @@ _BASE_TOOLS: tuple[Tool, ...] = (
     save_threat_model,
     amend_threat_model,
     web_search,
-    web_get_contents,
     create_vulnerability_report,
     create_dependency_report,
     update_vulnerability_report,

strix/config/settings.py

@@ -120,10 +120,6 @@ class TelemetrySettings(BaseSettings):
     enabled: bool = Field(default=True, alias="STRIX_TELEMETRY")
 
 
-WebSearchProvider = Literal["auto", "perplexity", "exa"]
-ExaSearchType = Literal["auto", "fast", "instant", "deep-lite", "deep", "deep-reasoning"]
-
-
 class IntegrationSettings(BaseSettings):
     model_config = _BASE_CONFIG
 
@@ -132,25 +128,6 @@ class IntegrationSettings(BaseSettings):
         alias="PERPLEXITY_API_KEY",
         repr=False,
     )
-    exa_api_key: str | None = Field(
-        default=None,
-        alias="EXA_API_KEY",
-        repr=False,
-    )
-    web_search_provider: WebSearchProvider = Field(
-        default="auto",
-        alias="STRIX_WEB_SEARCH_PROVIDER",
-    )
-    exa_search_type: ExaSearchType = Field(
-        default="auto",
-        alias="STRIX_EXA_SEARCH_TYPE",
-    )
-    exa_num_results: int = Field(
-        default=5,
-        ge=1,
-        le=100,
-        alias="STRIX_EXA_NUM_RESULTS",
-    )
     postman_api_key: str | None = Field(
         default=None,
         alias="POSTMAN_API_KEY",

strix/core/agents.py

@@ -24,8 +24,6 @@ logger = logging.getLogger(__name__)
 
 Status = Literal["running", "waiting", "completed", "stopped", "crashed", "failed", "budget_paused"]
 
-TERMINAL_STATUSES: frozenset[str] = frozenset({"completed", "stopped", "crashed", "failed"})
-
 # Why an agent parked. The user can message any agent, so this - not the agent's
 # position in the tree - decides whether waiting is bounded: only an agent waiting
 # on other agents is re-checked on a timer.
@@ -38,10 +36,6 @@ class AgentRuntime:
     task: asyncio.Task[Any] | None = None
     stream: Any | None = None
     interrupt_on_message: bool = False
-    # Whether the agent's loop parks after a terminal state and can be woken by a
-    # later message. A non-interactive loop returns instead, so once such an
-    # agent is terminal nothing will ever read its mailbox again.
-    resumable: bool = True
     wake: asyncio.Event = field(default_factory=asyncio.Event)
     mailbox: list[dict[str, Any]] = field(default_factory=list)
     user_wake_required: bool = False
@@ -181,7 +175,6 @@ class AgentCoordinator:
         session: Session | None = None,
         task: asyncio.Task[Any] | None = None,
         interrupt_on_message: bool | None = None,
-        resumable: bool | None = None,
     ) -> None:
         async with self._lock:
             runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
@@ -191,8 +184,6 @@ class AgentCoordinator:
                 runtime.task = task
             if interrupt_on_message is not None:
                 runtime.interrupt_on_message = interrupt_on_message
-            if resumable is not None:
-                runtime.resumable = resumable
 
     async def mark_running(self, agent_id: str) -> None:
         async with self._lock:
@@ -284,29 +275,10 @@ class AgentCoordinator:
             self._parent_notified.add(agent_id)
             return True
 
-    def _unreachable_locked(self, agent_id: str) -> bool:
-        """True when the agent is terminal and no loop will ever read its mailbox."""
-        if self.statuses.get(agent_id) not in TERMINAL_STATUSES:
-            return False
-        runtime = self.runtimes.get(agent_id)
-        return runtime is not None and not runtime.resumable
-
-    async def reachability(self, agent_id: str) -> tuple[bool, Status | None]:
-        """Whether a message to ``agent_id`` can still be acted on, plus its status."""
-        async with self._lock:
-            status = self.statuses.get(agent_id)
-            if status is None:
-                return False, None
-            return not self._unreachable_locked(agent_id), status
-
     async def send(
         self, target_agent_id: str, message: dict[str, Any], *, interrupt: bool = True
     ) -> bool:
-        """Queue a user/peer message in the target's mailbox and wake it.
-
-        Returns False when nothing will ever read the message: the target is
-        unknown, or it is terminal and its loop does not park for wake-ups.
-        """
+        """Queue a user/peer message in the target's mailbox and wake it."""
         from_user = message.get("from") == "user"
         if from_user and self._budget_paused:
             await self.resume_from_budget_pause(exclude=target_agent_id)
@@ -314,13 +286,6 @@ class AgentCoordinator:
             if target_agent_id not in self.statuses:
                 logger.debug("agent.send dropped unknown target=%s", target_agent_id)
                 return False
-            if self._unreachable_locked(target_agent_id):
-                logger.info(
-                    "agent.send dropped: target=%s is %s and cannot be woken",
-                    target_agent_id,
-                    self.statuses[target_agent_id],
-                )
-                return False
             runtime = self.runtimes.setdefault(target_agent_id, AgentRuntime())
             runtime.mailbox.append(dict(message))
             self.pending_counts[target_agent_id] = self.pending_counts.get(target_agent_id, 0) + 1

strix/core/execution.py

@@ -202,7 +202,6 @@ async def run_agent_loop(
         agent_id,
         session=session,
         interrupt_on_message=interactive,
-        resumable=interactive,
     )
     result: RunResultBase | None = None
 
@@ -1007,7 +1006,7 @@ async def _start_child_runner(
 ) -> None:
     session = open_agent_session(child_id, agents_db_path)
     sessions_to_close.append(session)
-    await coordinator.attach_runtime(child_id, session=session, resumable=interactive)
+    await coordinator.attach_runtime(child_id, session=session)
 
     child_ctx: dict[str, Any] = dict(parent_ctx)
     child_ctx["agent_id"] = child_id

strix/core/runner.py

@@ -45,7 +45,6 @@ from strix.core.paths import run_dir_for, runtime_state_dir
 from strix.core.sessions import open_agent_session
 from strix.report.state import get_global_report_state
 from strix.runtime import session_manager
-from strix.telemetry import set_scan_phase
 from strix.telemetry.logging import set_scan_id, setup_scan_logging
 from strix.tools.output_store import (
     WORKSPACE_SPILL_DIR,
@@ -117,13 +116,6 @@ def _record_mcp_connections(connections: list[ConnectedMcpServer]) -> None:
     report_state.record_mcp_connections([connection.name for connection in connections])
 
 
-def _note_exit_reason(reason: str) -> None:
-    """Record why the scan stopped so the end-of-scan beacon reports it."""
-    report_state = get_global_report_state()
-    if report_state is not None and report_state.scan_ended_exit_reason is None:
-        report_state.scan_ended_exit_reason = reason
-
-
 def _persist_mcp_status(roster: list[dict[str, Any]]) -> None:
     """Write the run's non-secret MCP connection status roster to run.json.
 
@@ -321,7 +313,6 @@ async def run_strix_scan(
         root_id = uuid.uuid4().hex[:8]
 
     logger.info("Bringing up sandbox session for scan %s", scan_id)
-    set_scan_phase("sandbox_init")
     bundle = await session_manager.create_or_reuse(
         scan_id,
         image=image,
@@ -331,7 +322,6 @@ async def run_strix_scan(
     )
     report("Waiting for the first model response")
     logger.info("Sandbox ready for scan %s", scan_id)
-    set_scan_phase("agent_setup")
 
     sandbox_session = bundle["session"]
 
@@ -583,7 +573,6 @@ async def run_strix_scan(
         async with coordinator._lock:
             root_status = coordinator.statuses.get(root_id)
 
-        set_scan_phase("agent_loop")
         result = await run_agent_loop(
             agent=root_agent,
             initial_input=initial_input,
@@ -621,7 +610,6 @@ async def run_strix_scan(
         return result  # noqa: TRY300
     except BudgetExceededError as exc:
         logger.info("Scan %s stopped: %s", scan_id, exc)
-        _note_exit_reason("budget_exceeded")
         if root_id is not None:
             with contextlib.suppress(Exception):
                 await coordinator.set_status(root_id, "stopped")
@@ -634,7 +622,6 @@ async def run_strix_scan(
             exc,
             scan_id,
         )
-        _note_exit_reason("rate_limited")
         if root_id is not None:
             with contextlib.suppress(Exception):
                 await coordinator.set_status(root_id, "stopped")

strix/interface/cloud/http.py

@@ -34,30 +34,13 @@ EXIT_AUTH = 4
 EXIT_PAYMENT = 5
 
 
-TOPUP_COMMAND = "strix cloud billing topup --credits <count>"
-BALANCE_COMMAND = "strix cloud billing credits"
-
-
 class CloudError(Exception):
-    """A failed cloud command. Carries the process exit code.
+    """A failed cloud command. Carries the process exit code."""
 
-    `next_step` is a short recovery instruction that the runner prints on its
-    own line after the error, so a person or an agent can act without reading
-    the docs.
-    """
-
-    def __init__(
-        self,
-        message: str,
-        *,
-        exit_code: int = EXIT_ERROR,
-        payload: Any = None,
-        next_step: str | None = None,
-    ) -> None:
+    def __init__(self, message: str, *, exit_code: int = EXIT_ERROR, payload: Any = None) -> None:
         super().__init__(message)
         self.exit_code = exit_code
         self.payload = payload
-        self.next_step = next_step
 
 
 class CloudTransportError(CloudError):
@@ -366,43 +349,13 @@ def check(response: requests.Response) -> Any:
             error_code = error_code or str(nested.get("code") or "")
             detail = str(nested.get("message") or detail)
     message = detail or f"HTTP {response.status_code}"
-    if error_code == "scan_credit_limit_reached" or response.status_code == 402:
-        raise payment_required_error(data, detail=detail)
+    if error_code == "scan_credit_limit_reached":
+        raise CloudError(message, exit_code=EXIT_PAYMENT, payload=data)
     if response.status_code in (401, 403):
         raise CloudError(message, exit_code=EXIT_AUTH, payload=data)
+    if response.status_code == 402:
+        hint = detail or (
+            "not enough credits. Run `strix cloud billing topup --credits N` to buy credits."
+        )
+        raise CloudError(hint, exit_code=EXIT_PAYMENT, payload=data)
     raise CloudError(message, exit_code=EXIT_ERROR, payload=data)
-
-
-def topup_url() -> str:
-    return f"{app_url()}/settings/billing"
-
-
-def topup_next_step(url: str | None = None) -> str:
-    return (
-        f"Buy credits with `{TOPUP_COMMAND}` or at {url or topup_url()}. "
-        f"Run `{BALANCE_COMMAND}` to see the balance. Then retry this command."
-    )
-
-
-def payment_required_error(data: Any, *, detail: str = "") -> CloudError:
-    """Build the error for an exhausted credit balance.
-
-    The platform sends the recovery instruction in `hint` and repeats it inside
-    `detail`. The CLI shows the instruction once, on its own line, and adds its
-    own instruction when the platform sends none.
-    """
-    server_hint = ""
-    server_url: str | None = None
-    if isinstance(data, dict):
-        raw = cast("dict[str, Any]", data)
-        server_hint = str(raw.get("hint") or "").strip()
-        raw_url = raw.get("topup_url")
-        if isinstance(raw_url, str) and raw_url.startswith("https://"):
-            server_url = raw_url
-    message = detail.strip()
-    if server_hint and message.endswith(server_hint):
-        message = message[: -len(server_hint)].strip()
-    if not message:
-        message = "Not enough credits to run this command."
-    next_step = server_hint or topup_next_step(server_url)
-    return CloudError(message, exit_code=EXIT_PAYMENT, payload=data, next_step=next_step)

strix/interface/cloud/runner.py

@@ -1035,14 +1035,10 @@ def _emit_error(
             payload = {"error": str(exc)}
             if exc.payload is not None:
                 payload["detail"] = exc.payload
-        if exc.next_step:
-            payload["next_step"] = exc.next_step
         sys.stdout.write(json.dumps(payload, indent=2, default=str) + "\n")
         return
     target = Console(stderr=True) if to_stderr else console
     target.print(f"[red]Error:[/] {escape(sanitize_terminal_text(exc))}")
-    if exc.next_step:
-        target.print(f"[yellow]Next step:[/] {escape(sanitize_terminal_text(exc.next_step))}")
 
 
 def _emit_interrupted(console: Console, *, as_json: bool, to_stderr: bool) -> None:

strix/interface/cloud/source_upload.py

@@ -203,17 +203,6 @@ def prepare_source(
     """Select safe source files and build a bounded temporary ZIP archive."""
     source = Path(value).expanduser().resolve()
     if not source.is_dir():
-        if source.is_file() and (
-            source.name.lower().endswith(_ARCHIVE_SUFFIXES) or _has_archive_magic(source)
-        ):
-            raise http.CloudError(
-                f"--source must be a directory, not an archive: {source}",
-                next_step=(
-                    "Extract the archive and pass the directory to --source. Strix packs the "
-                    "directory and excludes dependencies, build output, and secret-like files. "
-                    "Add --dry-run --show-files to review the selection first."
-                ),
-            )
         raise http.CloudError(f"--source must be a directory: {source}")
     manifest = select_source(
         source,
@@ -235,34 +224,14 @@ def prepare_source(
     archive_bytes = archive_path.stat().st_size
     if archive_bytes > MAX_ARCHIVE_BYTES:
         archive_path.unlink(missing_ok=True)
-        raise _archive_too_large_error(manifest, archive_bytes)
+        raise http.CloudError(
+            "source archive is larger than the 50 MB upload limit; narrow --source or "
+            "add --exclude patterns."
+        )
     digest = _sha256(archive_path)
     return SourceBundle(manifest, archive_path, archive_bytes, digest)
 
 
-_LARGEST_FILES_SHOWN = 5
-
-
-def _format_mib(size: int) -> str:
-    return f"{size / (1024 * 1024):.1f} MiB"
-
-
-def _archive_too_large_error(manifest: SourceManifest, archive_bytes: int) -> http.CloudError:
-    """Name the largest selected files so the user knows what to exclude."""
-    largest = sorted(manifest.files, key=lambda item: item.size, reverse=True)
-    listed = ", ".join(
-        f"{item.archive_name} ({_format_mib(item.size)})" for item in largest[:_LARGEST_FILES_SHOWN]
-    )
-    return http.CloudError(
-        f"the source archive is {_format_mib(archive_bytes)}, larger than the "
-        f"{_format_mib(MAX_ARCHIVE_BYTES)} upload limit. Largest files: {listed}.",
-        next_step=(
-            "Add --exclude patterns for large files or directories, or point --source at a "
-            "smaller directory. Run with --dry-run --show-files to review the selection."
-        ),
-    )
-
-
 def select_source(
     source: Path,
     *,

strix/interface/cloud/spec.py

@@ -1066,8 +1066,7 @@ SPEC: dict[str, dict[str, Cmd]] = {
         "request": Cmd(
             "POST",
             "/uploads/request",
-            "Request an upload URL. To scan local source, prefer `strix cloud scans start "
-            "--source DIR`, which packs, uploads, and starts the scan in one step.",
+            "Request an upload URL.",
             body=(
                 P("file_name", required=True, help="File name."),
                 P("file_size", "int", required=True, help="File size in bytes."),

strix/interface/environment.py

@@ -8,29 +8,17 @@ from rich.console import Console
 from rich.panel import Panel
 from rich.text import Text
 
-from strix.config import IntegrationSettings, codex, load_settings
+from strix.config import codex, load_settings
 from strix.interface.utils import (
     check_docker_connection,
     image_exists,
     process_pull_line,
 )
-from strix.telemetry import report_error
 
 
 logger = logging.getLogger(__name__)
 
 
-def _missing_web_search_vars(integrations: IntegrationSettings) -> list[str]:
-    """Mirror the web_search provider rules: which key(s) the selected provider needs."""
-    if integrations.web_search_provider == "exa":
-        return [] if integrations.exa_api_key else ["EXA_API_KEY"]
-    if integrations.web_search_provider == "perplexity":
-        return [] if integrations.perplexity_api_key else ["PERPLEXITY_API_KEY"]
-    if integrations.exa_api_key or integrations.perplexity_api_key:
-        return []
-    return ["EXA_API_KEY", "PERPLEXITY_API_KEY"]
-
-
 def validate_environment() -> None:
     logger.info("Validating environment")
     console = Console()
@@ -45,7 +33,6 @@ def validate_environment() -> None:
                 f"[red]STRIX_LLM={settings.llm.model} uses your ChatGPT subscription, "
                 "but you're not signed in.[/] Run [cyan]strix auth login chatgpt[/] first."
             )
-            report_error("subscription_not_signed_in")
             sys.exit(1)
         logger.info("Environment OK (ChatGPT subscription)")
         return
@@ -59,7 +46,8 @@ def validate_environment() -> None:
     if not settings.llm.api_base:
         missing_optional_vars.append("LLM_API_BASE")
 
-    missing_optional_vars.extend(_missing_web_search_vars(settings.integrations))
+    if not settings.integrations.perplexity_api_key:
+        missing_optional_vars.append("PERPLEXITY_API_KEY")
 
     if missing_required_vars:
         error_text = Text()
@@ -101,14 +89,7 @@ def validate_environment() -> None:
                     error_text.append("• ", style="white")
                     error_text.append("PERPLEXITY_API_KEY", style="bold cyan")
                     error_text.append(
-                        " - API key for Perplexity AI web search (alternative to Exa)\n",
-                        style="white",
-                    )
-                elif var == "EXA_API_KEY":
-                    error_text.append("• ", style="white")
-                    error_text.append("EXA_API_KEY", style="bold cyan")
-                    error_text.append(
-                        " - API key for Exa web search (enables real-time research)\n",
+                        " - API key for Perplexity AI web search (enables real-time research)\n",
                         style="white",
                     )
                 elif var == "STRIX_REASONING_EFFORT":
@@ -135,8 +116,6 @@ def validate_environment() -> None:
                     error_text.append(
                         "export PERPLEXITY_API_KEY='your-perplexity-key-here'\n", style="dim white"
                     )
-                elif var == "EXA_API_KEY":
-                    error_text.append("export EXA_API_KEY='your-exa-key-here'\n", style="dim white")
                 elif var == "STRIX_REASONING_EFFORT":
                     error_text.append(
                         "export STRIX_REASONING_EFFORT='high'\n",
@@ -155,7 +134,6 @@ def validate_environment() -> None:
         console.print("\n")
         console.print(panel)
         console.print()
-        report_error("missing_required_config")
         sys.exit(1)
     logger.info(
         "Environment OK (optional missing: %s)",
@@ -183,7 +161,6 @@ def check_docker_installed() -> None:
             padding=(1, 2),
         )
         console.print("\n", panel, "\n")
-        report_error("docker_not_installed")
         sys.exit(1)
     logger.debug("Docker CLI present")
 
@@ -231,7 +208,6 @@ def pull_docker_image() -> None:
                 padding=(1, 2),
             )
             console.print(panel, "\n")
-            report_error("image_pull_failed", e)
             sys.exit(1)
 
     logger.info("Docker image %s ready", image)

strix/interface/main.py

@@ -41,8 +41,7 @@ from strix.interface.update_check import (
 from strix.interface.utils import (
     build_final_stats_text,
 )
-from strix.llm.warmup import start_import_warmup, wait_for_import_warmup
-from strix.telemetry import posthog, report_error, scarf, set_scan_phase
+from strix.telemetry import posthog, scarf
 from strix.telemetry.logging import configure_dependency_logging
 
 
@@ -396,18 +395,15 @@ def _bootstrap_scan(args: argparse.Namespace) -> None:
     happen inside the TUI so the interface paints immediately instead of
     waiting on a model round trip.
     """
-    set_scan_phase("preflight")
     try:
         asyncio.run(warm_up_llm(show_model_warning=True))
     except ModelConnectionError as exc:
-        report_error("model_connection_failed", exc)
         _print_model_connection_error(exc, exc.model_name)
         sys.exit(1)
     persist_current()
     try:
         prepare_run(args)
     except ValueError as e:
-        report_error("scan_preparation_failed", e)
         _print_error_panel("SCAN PREPARATION FAILED", str(e))
         sys.exit(1)
     telemetry_start(args)
@@ -454,6 +450,8 @@ def main() -> None:
 
         sys.exit(run_cloud(sys.argv[2:]))
 
+    from strix.llm.warmup import start_import_warmup
+
     start_import_warmup()
 
     args = parse_arguments()
@@ -468,9 +466,6 @@ def main() -> None:
     pull_docker_image()
     validate_environment()
 
-    # Everything below imports the scan engine; do not race the warm-up thread.
-    wait_for_import_warmup()
-
     if args.non_interactive:
         _bootstrap_scan(args)
 
@@ -482,21 +477,18 @@ def main() -> None:
             from strix.interface.cli import run_cli
 
             asyncio.run(run_cli(args))
-            # Headless runs have no user to quit: the agent either finished
-            # (already beaconed as finished_by_tool) or stopped on its own.
-            exit_reason = "agent_stopped"
         else:
             asyncio.run(run_tui(args))
     except InteractiveSetupUnavailableError as exc:
         exit_reason = "error"
-        report_error("interactive_setup_unavailable", exc)
         _print_error_panel("INTERACTIVE SETUP UNAVAILABLE", str(exc))
         sys.exit(1)
     except KeyboardInterrupt:
         exit_reason = "interrupted"
-    except Exception as exc:
+    except Exception:
         exit_reason = "error"
-        report_error("unhandled_exception", exc)
+        posthog.error("unhandled_exception")
+        scarf.error("unhandled_exception")
         raise
     finally:
         report_state = get_global_report_state()

strix/interface/tui/runtime.py

@@ -37,7 +37,6 @@ from strix.interface.tui.sidecar import (
 )
 from strix.interface.utils import read_workspace_files
 from strix.report.state import ReportState, set_global_report_state
-from strix.telemetry import report_error, set_scan_phase
 from strix.utils.resource_paths import get_strix_resource_path
 
 
@@ -139,13 +138,11 @@ class GoTuiRuntime:
             await self._preflight_model()
         except Exception as exc:
             logger.exception("Go TUI setup model preflight failed")
-            report_error("model_connection_failed", exc)
             raise RuntimeError(f"Model connection failed: {exc}") from exc
 
     async def _preflight_model(self) -> None:
         model = (load_settings().llm.model or "").strip()
         self.controller.add_message("Verifying model connection...")
-        set_scan_phase("preflight")
         await preflight_model_connection(model)
         self.model_verified = True
 
@@ -184,11 +181,7 @@ class GoTuiRuntime:
             candidate.target = list(self.controller.targets)
             candidate.target_list = []
             build_targets_info(candidate)
-        try:
-            prepare_run(candidate)
-        except Exception as exc:
-            report_error("scan_preparation_failed", exc)
-            raise
+        prepare_run(candidate)
         telemetry_start(candidate)
 
         vars(self.args).update(vars(candidate))
@@ -202,21 +195,13 @@ class GoTuiRuntime:
         launch so the interface appears immediately.
         """
         model = (load_settings().llm.model or "").strip()
-        set_scan_phase("preflight")
         try:
             await preflight_model_connection(model)
-        except Exception as exc:
-            logger.exception("Go TUI scan preparation failed")
-            report_error("model_connection_failed", exc)
-            self.controller.fail_preparation(str(exc))
-            return
-        try:
             persist_current()
             prepare_run(self.args)
             telemetry_start(self.args)
         except Exception as exc:
             logger.exception("Go TUI scan preparation failed")
-            report_error("scan_preparation_failed", exc)
             self.controller.fail_preparation(str(exc))
             return
         self.controller.scan_state = "running"
@@ -255,9 +240,6 @@ class GoTuiRuntime:
             self.controller.scan_state = "completed" if report_status == "completed" else "stopped"
         except Exception as exc:
             logger.exception("Go TUI scan failed")
-            report_error("unhandled_exception", exc)
-            if self.report_state is not None and self.report_state.scan_ended_exit_reason is None:
-                self.report_state.scan_ended_exit_reason = "error"
             self.scan_error = exc
             self.controller.error = str(exc)
             self.controller.scan_state = "failed"

strix/interface/utils.py

@@ -19,7 +19,6 @@ from rich.panel import Panel
 from rich.text import Text
 
 from strix.config import load_settings
-from strix.telemetry import report_error
 from strix.utils.api_spec import detect_spec_format
 
 
@@ -1603,8 +1602,7 @@ def check_docker_connection() -> Any:
 
     try:
         return docker.from_env()
-    except DockerException as exc:
-        report_error("docker_unavailable", exc)
+    except DockerException:
         console = Console()
         error_text = Text()
         error_text.append("DOCKER NOT AVAILABLE", style="bold red")

strix/interface/viewer/cli.py

@@ -87,7 +87,7 @@ def run_view(argv: list[str]) -> None:
 
     posthog.viewer_opened(source="cli", live=live)
 
-    state_label = _state_label(summary)
+    state_label = "[#eab308]live[/]" if live else "[#22c55e]finished[/]"
     console.print()
     console.print(f"Serving [bold white]{run_name}[/] ({state_label}) at:")
     # Print the URL alone on its own line with soft_wrap so Rich never inserts a
@@ -107,18 +107,6 @@ def run_view(argv: list[str]) -> None:
         httpd.server_close()
 
 
-def _state_label(summary: dict[str, object]) -> str:
-    if not summary.get("finished", False):
-        return "[#eab308]live[/]"
-
-    status = summary.get("status")
-    if status == "failed":
-        return "[#ef4444]failed[/]"
-    if status in {"stopped", "interrupted"}:
-        return f"[#eab308]{status}[/]"
-    return "[#22c55e]finished[/]"
-
-
 def _resolve_run_dir(run: str | None, console: Console) -> Path:
     if run:
         run_dir = run_dir_for(run)

strix/llm/warmup.py

@@ -1,22 +1,20 @@
 """Background pre-import of the heavy scan dependencies.
 
 The scan engine's import graph (the agents SDK, OpenAI client, LiteLLM, the
-Caido SDK) costs seconds to import cold, but none of it is needed until a scan
-actually starts. Importing it on a daemon thread at CLI entry overlaps that
-cost with the I/O-bound startup work that always precedes a scan (argument
-parsing, Docker checks, image pull, TUI setup). The Docker SDK is not on the
-list: the Docker checks import it on the main thread during that same window.
-
-The main thread must call :func:`wait_for_import_warmup` before its first
-import from that graph. Two threads that enter the same package graph from
-different modules hold each other's import locks, and CPython breaks the cycle
-by failing one of the imports.
+Caido SDK, the Docker SDK) costs seconds to import cold, but none of it is
+needed until a scan actually starts. Importing it on a daemon thread at CLI
+entry overlaps that cost with the I/O-bound startup work that always precedes
+a scan (argument parsing, Docker checks, image pull, TUI setup), so by the
+time the scan begins the modules are already in ``sys.modules``. Any thread
+that needs one of them before the warm-up finishes just blocks on the normal
+import lock, so behaviour is unchanged either way.
 """
 
 from __future__ import annotations
 
 import importlib
 import logging
+import sys
 import threading
 
 
@@ -26,17 +24,45 @@ WARMUP_MODULES = (
     "strix.core.runner",
     "litellm",
     "caido_sdk_client",
+    "docker",
 )
 
+_lock = threading.Lock()
 _thread: threading.Thread | None = None
 
 
+def _purge_orphaned_modules(before: frozenset[str]) -> None:
+    """Remove submodules stranded by an import attempt that just failed.
+
+    When a package import fails partway (for example CPython's import-lock
+    deadlock avoidance breaking a cross-thread cycle), the failed package is
+    removed from ``sys.modules`` but submodules it already finished stay
+    behind. A later import of one of those submodules then short-circuits on
+    the cached entry without re-importing its parent, and re-entering the
+    parent from inside a submodule crashes with "partially initialized
+    module". Dropping the orphans (cached submodules whose ancestor package is
+    gone) restores a clean slate, and touches nothing another thread imported
+    successfully.
+    """
+    added = set(sys.modules) - before
+    for name in added:
+        parent = name.rpartition(".")[0]
+        while parent:
+            if parent not in sys.modules:
+                sys.modules.pop(name, None)
+                logger.debug("Import warm-up purged orphaned module %r", name)
+                break
+            parent = parent.rpartition(".")[0]
+
+
 def _warm(modules: tuple[str, ...]) -> None:
     for name in modules:
+        before = frozenset(sys.modules)
         try:
             importlib.import_module(name)
         except Exception:  # noqa: BLE001 - a failed warm-up must never fail the run.
             logger.debug("Import warm-up for %r failed", name, exc_info=True)
+            _purge_orphaned_modules(before)
 
 
 def start_import_warmup(modules: tuple[str, ...] = WARMUP_MODULES) -> threading.Thread:
@@ -46,15 +72,11 @@ def start_import_warmup(modules: tuple[str, ...] = WARMUP_MODULES) -> threading.
     runtime that has no local Docker) warm a narrower set.
     """
     global _thread  # noqa: PLW0603
-    if _thread is None:
+    with _lock:
+        if _thread is not None:
+            return _thread
         _thread = threading.Thread(
             target=_warm, args=(modules,), name="strix-import-warmup", daemon=True
         )
         _thread.start()
-    return _thread
-
-
-def wait_for_import_warmup() -> None:
-    """Block until the warm-up thread has finished, if one was started."""
-    if _thread is not None:
-        _thread.join()
+        return _thread

strix/skills/__init__.py

@@ -1,5 +1,6 @@
 import logging
 import re
+import threading
 from collections import Counter
 from collections.abc import Iterator
 from pathlib import Path
@@ -7,6 +8,7 @@ from typing import TypeGuard
 
 import yaml
 
+from strix.telemetry import posthog, scarf
 from strix.utils.resource_paths import get_strix_resource_path
 
 
@@ -239,19 +241,16 @@ def validate_requested_skills(skill_list: list[str], max_skills: int = 5) -> str
     return None
 
 
-_LOADED_SKILLS: set[str] = set()
-
-
 def _track_skill_loaded(skill_name: str, file_path: Path) -> None:
     builtin = get_strix_resource_path("skills")
     if not file_path.is_relative_to(builtin):
         skill_name = "custom"
-    _LOADED_SKILLS.add(skill_name)
 
+    def _send() -> None:
+        posthog.skill_loaded(skill_name)
+        scarf.skill_loaded(skill_name)
 
-def get_loaded_skill_names() -> list[str]:
-    """Distinct skills loaded so far in this process (custom skills collapse to ``"custom"``)."""
-    return sorted(_LOADED_SKILLS)
+    threading.Thread(target=_send, daemon=True).start()
 
 
 def _candidate_skill_files(skill_name: str) -> list[Path]:

strix/telemetry/README.md

@@ -12,11 +12,11 @@ Privacy is our priority. All collected data is anonymized by default. Each sessi
 
 We collect only very **basic** usage data including:
 
-**Session Errors:** Duration, the failure category, the scan phase, and the exception class name (not messages or stack traces)\
+**Session Errors:** Duration and error types (not messages or stack traces)\
 **System Context:** OS type, architecture, Strix version\
 **Scan Context:** Scan mode (quick/standard/deep), scan type (whitebox/blackbox)\
 **Model Usage:** Which LLM model is being used and whether it runs via an API key or a model subscription (not prompts or responses)\
-**Feature Usage:** Which built-in skills were used during a scan (reported once, at scan end)\
+**Feature Usage:** Which built-in skills are loaded\
 **Aggregate Metrics:** Vulnerability counts by severity and weakness category (CWE)
 
 ### What We **Never** Collect

strix/telemetry/__init__.py

@@ -1,19 +1,7 @@
 from . import posthog, scarf
-from ._common import set_scan_phase
-
-
-def report_error(error_type: str, exc: BaseException | None = None) -> None:
-    """Beacon a failure category, plus the exception class when one is given.
-
-    Only class names travel: never the message, arguments, or traceback.
-    """
-    posthog.error(error_type, exc)
-    scarf.error(error_type, exc)
 
 
 __all__ = [
     "posthog",
-    "report_error",
     "scarf",
-    "set_scan_phase",
 ]

strix/telemetry/_common.py

@@ -5,7 +5,7 @@ import platform
 import sys
 from importlib.metadata import PackageNotFoundError, version
 from pathlib import Path
-from typing import Any, cast
+from typing import Any
 from uuid import uuid4
 
 
@@ -54,43 +54,3 @@ def base_props() -> dict[str, Any]:
         "python": f"{sys.version_info.major}.{sys.version_info.minor}",
         "strix_version": get_version(),
     }
-
-
-# Coarse stage of the current run, attached to ``error`` beacons so a failure
-# can be placed without a message or trace. Process-local, like the rest of the
-# CLI telemetry: one process runs one scan.
-_scan_phase = "startup"
-
-
-def set_scan_phase(phase: str) -> None:
-    global _scan_phase  # noqa: PLW0603
-    _scan_phase = phase
-
-
-def get_scan_phase() -> str:
-    return _scan_phase
-
-
-def _exception_name(exc: BaseException) -> str:
-    cls = type(exc)
-    package = cls.__module__.split(".")[0]
-    return cls.__name__ if package == "builtins" else f"{package}.{cls.__name__}"
-
-
-def _unwrap_group(exc: BaseException) -> BaseException:
-    if not isinstance(exc, BaseExceptionGroup):
-        return exc
-    group = cast("BaseExceptionGroup[BaseException]", exc)
-    return group.exceptions[0] if group.exceptions else group
-
-
-def exception_props(exc: BaseException) -> dict[str, str]:
-    """Class names only. Messages, arguments, and tracebacks never leave the machine."""
-    exc = _unwrap_group(exc)
-    props = {"exception_type": _exception_name(exc)}
-    cause = exc.__cause__
-    if cause is None and not exc.__suppress_context__:
-        cause = exc.__context__
-    if cause is not None:
-        props["exception_cause"] = _exception_name(cause)
-    return props

strix/telemetry/posthog.py

@@ -4,14 +4,10 @@ from typing import TYPE_CHECKING, Any
 import requests
 
 from strix.config import load_settings
-from strix.skills import get_loaded_skill_names
 from strix.telemetry._common import (
     SEND_TIMEOUT,
     SESSION_ID,
     base_props,
-    exception_props,
-    get_scan_phase,
-    get_version,
     is_first_run,
 )
 
@@ -39,12 +35,7 @@ def _send(event: str, properties: dict[str, Any]) -> bool:
             "api_key": _POSTHOG_PUBLIC_API_KEY,
             "event": event,
             "distinct_id": SESSION_ID,
-            "properties": {
-                **properties,
-                "$lib": "strix-cli",
-                "$lib_version": get_version(),
-                "$process_person_profile": False,
-            },
+            "properties": properties,
         }
         with requests.post(f"{_POSTHOG_HOST}/capture/", json=payload, timeout=SEND_TIMEOUT):
             pass
@@ -91,6 +82,16 @@ def finding(severity: str, cwe: str | None = None, is_cve: bool = False) -> None
     )
 
 
+def skill_loaded(skill_name: str) -> None:
+    _send(
+        "skill_loaded",
+        {
+            **base_props(),
+            "skill": skill_name,
+        },
+    )
+
+
 def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
     if report_state.posthog_scan_ended_sent:
         return
@@ -129,7 +130,6 @@ def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
             "vulnerabilities_total": len(report_state.vulnerability_reports),
             **{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()},
             **llm_props,
-            "skills": get_loaded_skill_names(),
         },
     )
 
@@ -180,12 +180,6 @@ def viewer_agent_steered() -> None:
     _send("viewer_agent_steered", {**base_props()})
 
 
-def error(error_type: str, exc: BaseException | None = None) -> None:
-    props: dict[str, Any] = {
-        **base_props(),
-        "error_type": error_type,
-        "phase": get_scan_phase(),
-    }
-    if exc is not None:
-        props.update(exception_props(exc))
+def error(error_type: str) -> None:
+    props = {**base_props(), "error_type": error_type}
     _send("error", props)

strix/telemetry/scarf.py

@@ -7,13 +7,10 @@ from typing import TYPE_CHECKING, Any
 import requests
 
 from strix.config import load_settings
-from strix.skills import get_loaded_skill_names
 from strix.telemetry._common import (
     SEND_TIMEOUT,
     SESSION_ID,
     base_props,
-    exception_props,
-    get_scan_phase,
     get_version,
     is_first_run,
 )
@@ -93,6 +90,17 @@ def finding(severity: str, cwe: str | None = None, is_cve: bool = False) -> None
     )
 
 
+def skill_loaded(skill_name: str) -> None:
+    _send(
+        "skill_loaded",
+        {
+            **base_props(),
+            "session": SESSION_ID,
+            "skill": skill_name,
+        },
+    )
+
+
 def end(report_state: ReportState, exit_reason: str = "completed") -> None:
     if report_state.scarf_scan_ended_sent:
         return
@@ -132,18 +140,14 @@ def end(report_state: ReportState, exit_reason: str = "completed") -> None:
             "vulnerabilities_total": len(report_state.vulnerability_reports),
             **{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()},
             **llm_props,
-            "skills": ",".join(get_loaded_skill_names()),
         },
     )
 
 
-def error(error_type: str, exc: BaseException | None = None) -> None:
+def error(error_type: str) -> None:
     props: dict[str, Any] = {
         **base_props(),
         "session": SESSION_ID,
         "error_type": error_type,
-        "phase": get_scan_phase(),
     }
-    if exc is not None:
-        props.update(exception_props(exc))
     _send("error", props)

strix/tools/agents_graph/tools.py

@@ -15,7 +15,6 @@ from agents import RunContextWrapper, function_tool
 from strix.core.agents import Status, coordinator_from_context
 from strix.core.execution import notify_parent_on_terminal
 from strix.core.hooks import LLM_TURN_KEY
-from strix.report.state import get_global_report_state
 from strix.skills import validate_requested_skills
 
 
@@ -29,40 +28,6 @@ def _ctx(ctx: RunContextWrapper) -> dict[str, Any]:
     return ctx.context if isinstance(ctx.context, dict) else {}
 
 
-def _filed_reports_by(agent_id: str) -> list[dict[str, Any]]:
-    """Vulnerability reports the agent actually filed, from report state.
-
-    The narrative ``findings`` an agent hands to ``agent_finish`` is prose; a
-    parent that wants to act on a child's work needs the report ids. Read them
-    from the report state rather than trusting the child's description.
-    """
-    state = get_global_report_state()
-    if state is None:
-        return []
-    filed: list[dict[str, Any]] = []
-    seen: set[str] = set()
-    for report in state.get_existing_vulnerabilities():
-        if report.get("agent_id") != agent_id:
-            continue
-        report_id = str(report.get("id") or "")
-        if not report_id or report_id in seen:
-            continue
-        seen.add(report_id)
-        filed.append(report)
-    return filed
-
-
-def _render_filed_report(report: dict[str, Any]) -> str:
-    line = f"- {report.get('id')}"
-    severity = report.get("severity")
-    if severity:
-        line += f" [{str(severity).upper()}]"
-    title = report.get("title")
-    if title:
-        line += f" {title}"
-    return line
-
-
 def _render_completion_report(
     *,
     agent_name: str,
@@ -73,7 +38,6 @@ def _render_completion_report(
     findings: list[str],
     recommendations: list[str],
     open_items: list[str],
-    filed_reports: list[dict[str, Any]] | None = None,
 ) -> str:
     """Render a child's completion report as plain structured text.
 
@@ -99,12 +63,6 @@ def _render_completion_report(
         lines.append("Findings:")
         lines.extend(f"- {f}" for f in findings)
     lines.append("")
-    lines.append("Vulnerability reports filed by this agent (authoritative; use these ids):")
-    if filed_reports:
-        lines.extend(_render_filed_report(r) for r in filed_reports)
-    else:
-        lines.append("- (none)")
-    lines.append("")
     lines.append("Open items (unresolved, need follow-up):")
     if open_items:
         lines.extend(f"- {o}" for o in open_items)
@@ -191,11 +149,8 @@ async def send_message_to_agent(
     **Don't** use for routine "hello/status" pings, for context the
     target already has (children inherit parent history), or when
     parent/child completion via ``agent_finish`` already covers the
-    flow. In interactive runs a message wakes the target regardless of
+    flow. Messages to any registered agent wake it, regardless of
     status, so a follow-up can restart a completed/stopped/failed agent.
-    In non-interactive runs a finished agent is gone for good: the call
-    fails with the target's status, and you should read its filed
-    reports (``list_reports``) or spawn a new agent instead of waiting.
 
     Args:
         target_agent_id: Recipient's 8-char id.
@@ -240,23 +195,10 @@ async def send_message_to_agent(
         },
     )
     if not delivered:
-        _, status = await coordinator.reachability(target_agent_id)
-        if status is None:
-            error = f"Target agent '{target_agent_id}' not found"
-        else:
-            error = (
-                f"Target agent '{target_agent_id}' is '{status}' and cannot be woken in "
-                "this run; it will never read this message. Its filed reports are in "
-                "list_reports / get_report. Do not wait_for_agents on it - spawn a new "
-                "agent if more work is needed."
-            )
         return json.dumps(
             {
                 "success": False,
-                "error": error,
-                "target_agent_id": target_agent_id,
-                "target_status": status,
-                "delivery_status": "not_delivered",
+                "error": f"Target agent '{target_agent_id}' not found or message delivery failed",
             },
             ensure_ascii=False,
             default=str,
@@ -422,31 +364,6 @@ async def wait_for_agents(  # noqa: PLR0911
             default=str,
         )
 
-    # Non-interactive agents cannot be woken once terminal, so with nobody
-    # running or waiting there is no message left to wait for.
-    if not await coordinator.active_agents_except(me):
-        _, statuses, names, _ = await coordinator.graph_snapshot()
-        return json.dumps(
-            {
-                "success": True,
-                "wait_outcome": "no_active_agents",
-                "reason": reason,
-                "agents": [
-                    {"agent_id": aid, "name": names.get(aid, aid), "status": status}
-                    for aid, status in statuses.items()
-                    if aid != me
-                ],
-                "note": (
-                    "No other agent is running or waiting, so no message can arrive. "
-                    "Finished agents' results are in list_reports / get_report and their "
-                    "completion reports are already in your history. Continue your own "
-                    "work, spawn a new agent, or finish."
-                ),
-            },
-            ensure_ascii=False,
-            default=str,
-        )
-
     await coordinator.park_waiting(me, wait_kind="agents")
     try:
         await asyncio.wait_for(coordinator.wait_for_message(me), timeout_seconds)
@@ -693,9 +610,6 @@ async def agent_finish(
             default=str,
         )
 
-    filed_reports = _filed_reports_by(me)
-    filed_report_ids = [str(r.get("id")) for r in filed_reports]
-
     parent_notified = False
     if report_to_parent and await coordinator.claim_parent_notice(me):
         async with coordinator._lock:
@@ -709,7 +623,6 @@ async def agent_finish(
             findings=list(findings or []),
             recommendations=list(final_recommendations or []),
             open_items=list(open_items or []),
-            filed_reports=filed_reports,
         )
         await coordinator.send(
             parent_id,
@@ -719,7 +632,6 @@ async def agent_finish(
                 "content": report,
                 "type": "completion",
                 "priority": "high",
-                "filed_report_ids": filed_report_ids,
             },
         )
         parent_notified = True
@@ -730,11 +642,10 @@ async def agent_finish(
         await notify_parent_on_terminal(coordinator, me, "completed")
 
     logger.info(
-        "agent_finish: %s success=%s findings=%d filed_reports=%d parent_notified=%s",
+        "agent_finish: %s success=%s findings=%d parent_notified=%s",
         me,
         success,
         len(findings or []),
-        len(filed_report_ids),
         parent_notified,
     )
 
@@ -745,7 +656,6 @@ async def agent_finish(
             "parent_notified": parent_notified,
             "agent_id": me,
             "summary": result_summary,
-            "filed_report_ids": filed_report_ids,
             "findings_count": len(findings or []),
             "open_items_count": len(open_items or []),
             "has_recommendations": bool(final_recommendations),

strix/tools/web_search/tool.py

@@ -1,12 +1,11 @@
-"""Security-focused web research tools (Exa or Perplexity)."""
+"""``web_search`` — Perplexity-backed security-focused web search."""
 
 from __future__ import annotations
 
 import asyncio
 import json
 import logging
-from typing import TYPE_CHECKING, Any, cast
-from urllib.parse import urlsplit, urlunsplit
+from typing import Any
 
 import requests
 from agents import RunContextWrapper, function_tool
@@ -14,10 +13,6 @@ from agents import RunContextWrapper, function_tool
 from strix.config import load_settings
 
 
-if TYPE_CHECKING:
-    from collections.abc import Callable
-
-
 logger = logging.getLogger(__name__)
 
 
@@ -46,7 +41,22 @@ Structure your response to be comprehensive yet concise, emphasizing the most cr
 security implications and details."""
 
 
-def _perplexity_content(api_key: str, query: str) -> str:
+def _do_search(query: str) -> dict[str, Any]:  # noqa: PLR0911 - each error class needs its own sanitized return
+    if not query or not query.strip():
+        return {"success": False, "error": "Query cannot be empty"}
+
+    api_key = load_settings().integrations.perplexity_api_key
+    if not api_key:
+        logger.warning("web_search invoked without PERPLEXITY_API_KEY configured")
+        return {
+            "success": False,
+            "error": (
+                "Web search is not configured for this scan "
+                "(operator needs to set PERPLEXITY_API_KEY). Proceed without it"
+            ),
+        }
+    logger.info("web_search query (len=%d): %s", len(query), query[:120])
+
     url = "https://api.perplexity.ai/chat/completions"
     headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
     payload = {
@@ -56,269 +66,61 @@ def _perplexity_content(api_key: str, query: str) -> str:
             {"role": "user", "content": query},
         ],
     }
-    with requests.post(url, headers=headers, json=payload, timeout=300) as response:
-        response.raise_for_status()
-        return str(response.json()["choices"][0]["message"]["content"])
 
-
-_EXA_PAGE_MAX_CHARS = 20000
-_EXA_MAX_CONTENT_URLS = 10
-_EXA_SUMMARY_PROMPT = (
-    "Summarize this page for a penetration tester. Keep concrete technical detail: "
-    "affected products and exact versions, CVE and CWE identifiers, CVSS scores, "
-    "exploitation preconditions, payloads or commands, and mitigations. "
-    "Leave out marketing copy and navigation text."
-)
-
-
-def _exa_result_block(result: dict[str, Any]) -> str | None:
-    result_url = str(result.get("url") or result.get("id") or "")
-    if not result_url:
-        return None
-    title = str(result.get("title") or result_url)
-    parts = [f"### {title}\n{result_url}"]
-    summary = str(result.get("summary") or "").strip()
-    if summary:
-        parts.append(summary)
-    return "\n".join(parts)
-
-
-def _exa_page_block(result: dict[str, Any]) -> str | None:
-    result_url = str(result.get("url") or result.get("id") or "")
-    text = str(result.get("text") or "").strip()
-    if not result_url or not text:
-        return None
-    if len(text) > _EXA_PAGE_MAX_CHARS:
-        text = f"{text[:_EXA_PAGE_MAX_CHARS]}\n[truncated at {_EXA_PAGE_MAX_CHARS} characters]"
-    title = str(result.get("title") or result_url)
-    return f"### {title}\n{result_url}\n\n{text}"
-
-
-def _exa_blocks(
-    results: list[Any],
-    render: Callable[[dict[str, Any]], str | None],
-) -> list[str]:
-    blocks: list[str] = []
-    for result in results:
-        if not isinstance(result, dict):
-            continue
-        block = render(cast("dict[str, Any]", result))
-        if block:
-            blocks.append(block)
-    return blocks
-
-
-def _exa_post(api_key: str, endpoint: str, payload: dict[str, Any]) -> dict[str, Any]:
-    headers = {"x-api-key": api_key, "Content-Type": "application/json"}
-    with requests.post(endpoint, headers=headers, json=payload, timeout=300) as response:
-        response.raise_for_status()
-        body: dict[str, Any] = response.json()
-    return body
-
-
-def _exa_content(api_key: str, query: str, search_type: str, num_results: int) -> str:
-    body = _exa_post(
-        api_key,
-        "https://api.exa.ai/search",
-        {
-            "query": f"{_SYSTEM_PROMPT}\n\n{query}",
-            "type": search_type,
-            "numResults": num_results,
-            "contents": {"summary": {"query": _EXA_SUMMARY_PROMPT}},
-        },
-    )
-    blocks = _exa_blocks(body.get("results") or [], _exa_result_block)
-    if not blocks:
-        raise ValueError("Exa response has no results")
-    return "\n\n".join(blocks)
-
-
-def _normalize_url(url: str) -> str:
-    """Canonical form for matching: case-fold scheme and host only, drop a trailing slash."""
-    parts = urlsplit(url.strip())
-    return urlunsplit(
-        (parts.scheme.lower(), parts.netloc.lower(), parts.path.rstrip("/"), parts.query, "")
-    )
-
-
-def _exa_page_text(api_key: str, urls: list[str]) -> tuple[str, set[str]]:
-    """Fetch page text and report which of the requested URLs Exa returned."""
-    body = _exa_post(api_key, "https://api.exa.ai/contents", {"urls": urls, "text": True})
-    blocks: list[str] = []
-    fetched: set[str] = set()
-    results: list[Any] = body.get("results") or []
-    for result in results:
-        if not isinstance(result, dict):
-            continue
-        page = cast("dict[str, Any]", result)
-        block = _exa_page_block(page)
-        if not block:
-            continue
-        blocks.append(block)
-        fetched.add(_normalize_url(str(page.get("url") or page.get("id") or "")))
-    if not blocks:
-        raise ValueError("Exa returned no page contents")
-    return "\n\n".join(blocks), fetched
-
-
-def _resolve_provider(  # noqa: PLR0911 - each provider/missing-key case needs its own return
-    integrations: Any,
-) -> tuple[str, str] | dict[str, Any]:
-    """Pick the search provider and its key, or return a sanitized error dict."""
-    provider = integrations.web_search_provider
-    perplexity_key = integrations.perplexity_api_key
-    exa_key = integrations.exa_api_key
-
-    if provider == "perplexity":
-        if not perplexity_key:
-            return _not_configured_error("PERPLEXITY_API_KEY")
-        return ("perplexity", perplexity_key)
-    if provider == "exa":
-        if not exa_key:
-            return _not_configured_error("EXA_API_KEY")
-        return ("exa", exa_key)
-
-    if exa_key:
-        return ("exa", exa_key)
-    if perplexity_key:
-        return ("perplexity", perplexity_key)
-    return _not_configured_error("EXA_API_KEY or PERPLEXITY_API_KEY")
-
-
-def _not_configured_error(missing: str) -> dict[str, Any]:
-    logger.warning("web_search invoked without %s configured", missing)
-    return {
-        "success": False,
-        "error": (
-            "Web search is not configured for this scan "
-            f"(operator needs to set {missing}). Proceed without it"
-        ),
-    }
-
-
-def _guarded_call[T](  # noqa: PLR0911 - each error class needs its own sanitized return
-    tool: str,
-    rejected_hint: str,
-    fetch: Callable[[], T],
-) -> T | dict[str, Any]:
-    """Run a provider call and translate any failure into a sanitized error dict."""
     try:
-        return fetch()
+        with requests.post(url, headers=headers, json=payload, timeout=300) as response:
+            response.raise_for_status()
+            content = response.json()["choices"][0]["message"]["content"]
     except requests.exceptions.Timeout:
-        logger.warning("%s timed out", tool)
-        return {"success": False, "error": f"{tool} timed out. Try again or narrow the request"}
+        logger.warning("web_search timed out")
+        return {
+            "success": False,
+            "error": "Web search timed out. Try again or shorten the query",
+        }
     except requests.exceptions.HTTPError as exc:
         status = exc.response.status_code if exc.response is not None else None
-        logger.exception("%s HTTP error status=%s", tool, status)
+        logger.exception("web_search HTTP error status=%s", status)
         if status is not None and 400 <= status < 500:
-            return {"success": False, "error": rejected_hint}
-        return {"success": False, "error": f"{tool} service is unavailable. Try again later"}
+            return {
+                "success": False,
+                "error": (
+                    "Web search rejected the query. Refine it "
+                    "(more specific, shorter, no unusual characters) and retry"
+                ),
+            }
+        return {
+            "success": False,
+            "error": "Web search service is unavailable. Try again later",
+        }
     except requests.exceptions.RequestException:
-        logger.exception("%s network error", tool)
-        return {"success": False, "error": f"{tool} network error. Try again later"}
+        logger.exception("web_search network error")
+        return {
+            "success": False,
+            "error": "Web search network error. Try again later",
+        }
     except (KeyError, IndexError, ValueError):
-        logger.exception("%s response shape unexpected", tool)
-        return {"success": False, "error": f"{tool} returned an unexpected response. Try again"}
+        logger.exception("web_search response shape unexpected")
+        return {
+            "success": False,
+            "error": "Web search returned an unexpected response. Try again",
+        }
     except Exception:
-        logger.exception("%s failed", tool)
-        return {"success": False, "error": f"{tool} failed unexpectedly"}
-
-
-def _do_search(query: str) -> dict[str, Any]:
-    if not query or not query.strip():
-        return {"success": False, "error": "Query cannot be empty"}
-
-    integrations = load_settings().integrations
-    resolved = _resolve_provider(integrations)
-    if isinstance(resolved, dict):
-        return resolved
-    provider, api_key = resolved
-    logger.info("web_search provider=%s query (len=%d): %s", provider, len(query), query[:120])
-
-    def fetch() -> str:
-        if provider == "exa":
-            return _exa_content(
-                api_key,
-                query,
-                integrations.exa_search_type,
-                integrations.exa_num_results,
-            )
-        return _perplexity_content(api_key, query)
-
-    outcome = _guarded_call(
-        "Web search",
-        (
-            "Web search rejected the query. Refine it "
-            "(more specific, shorter, no unusual characters) and retry"
-        ),
-        fetch,
-    )
-    if isinstance(outcome, dict):
-        return outcome
-    return {
-        "success": True,
-        "query": query,
-        "provider": provider,
-        "content": outcome,
-    }
-
-
-def _do_get_contents(urls: list[str]) -> dict[str, Any]:
-    cleaned = [url.strip() for url in urls if url and url.strip()]
-    if not cleaned:
-        return {"success": False, "error": "Provide at least one URL"}
-    if len(cleaned) > _EXA_MAX_CONTENT_URLS:
+        logger.exception("web_search failed")
         return {
             "success": False,
-            "error": f"Too many URLs. Pass at most {_EXA_MAX_CONTENT_URLS} per call",
+            "error": "Web search failed unexpectedly",
         }
-
-    integrations = load_settings().integrations
-    api_key = integrations.exa_api_key
-    if not api_key:
-        return _not_configured_error("EXA_API_KEY")
-    if integrations.web_search_provider == "perplexity":
-        logger.warning("web_get_contents invoked while the provider is pinned to Perplexity")
+    else:
         return {
-            "success": False,
-            "error": (
-                "Page fetching needs the Exa provider "
-                "(operator pinned STRIX_WEB_SEARCH_PROVIDER to perplexity). "
-                "Use web_search instead"
-            ),
+            "success": True,
+            "query": query,
+            "content": content,
         }
 
-    logger.info("web_get_contents urls=%d", len(cleaned))
-    outcome = _guarded_call(
-        "Page fetch",
-        "Page fetch was rejected. Check the URLs are complete, public, and correctly formed",
-        lambda: _exa_page_text(api_key, cleaned),
-    )
-    if isinstance(outcome, dict):
-        return outcome
-    content, fetched = outcome
-    missing = [url for url in cleaned if _normalize_url(url) not in fetched]
-    result: dict[str, Any] = {
-        "success": True,
-        "urls": [url for url in cleaned if url not in missing],
-        "provider": "exa",
-        "content": content,
-    }
-    if missing:
-        logger.warning(
-            "web_get_contents returned %d of %d pages", len(cleaned) - len(missing), len(cleaned)
-        )
-        result["failed_urls"] = missing
-        result["warning"] = (
-            f"Exa returned no content for {len(missing)} of {len(cleaned)} requested URLs. "
-            "Those pages are missing from the content below"
-        )
-    return result
-
 
 @function_tool(timeout=330)
 async def web_search(ctx: RunContextWrapper, query: str) -> str:
-    """Real-time web search (Exa or Perplexity) — your primary research tool.
+    """Real-time web search via Perplexity — your primary research tool.
 
     Use it liberally for anything that's not in your training data:
 
@@ -348,12 +150,6 @@ async def web_search(ctx: RunContextWrapper, query: str) -> str:
     exploits, Kali-compatible tooling, and concrete code/command
     examples.
 
-    With the Exa provider you get a ranked list of results, each with a
-    title, URL, and a short security-focused summary. Read the result
-    you need, then call ``web_get_contents`` with its URL to pull the
-    full page text when a summary is not enough. With Perplexity you get
-    a single synthesized cited answer.
-
     **Good example queries** (each is a full sentence, names a
     version/product, and asks one concrete thing):
 
@@ -381,33 +177,3 @@ async def web_search(ctx: RunContextWrapper, query: str) -> str:
     """
     result = await asyncio.to_thread(_do_search, query)
     return json.dumps(result, ensure_ascii=False, default=str)
-
-
-@function_tool(timeout=330)
-async def web_get_contents(ctx: RunContextWrapper, urls: list[str]) -> str:
-    """Fetch the full, cleaned text of specific web pages (Exa only).
-
-    Use this as the drill-down step after ``web_search``: when a result's
-    summary is not enough, pass that result's URL here to read the whole
-    page. Good for reading a full advisory, a CVE writeup,
-    an exploit proof-of-concept, or vendor documentation end to end.
-
-    Prefer ``web_search`` first to find the right pages, then fetch only
-    the few URLs worth reading in full — each page can be large, so avoid
-    fetching many pages you do not need.
-
-    This tool needs the Exa provider (``EXA_API_KEY``). When the operator
-    pins the provider to Perplexity, it returns an error and you should
-    use ``web_search`` instead.
-
-    Some pages block extraction. When a page returns no content, the
-    result lists it under ``failed_urls`` and the ``content`` field holds
-    only the pages that came back. Check ``failed_urls`` before you
-    conclude that a page had nothing useful.
-
-    Args:
-        urls: The page URLs to fetch, at most 10 per call. Use complete,
-            public URLs (for example the ones returned by ``web_search``).
-    """
-    result = await asyncio.to_thread(_do_get_contents, urls)
-    return json.dumps(result, ensure_ascii=False, default=str)

tests/test_agent_graph_coordination.py

@@ -1,258 +0,0 @@
-"""Tests for parent/child coordination once a non-interactive child has finished.
-
-A non-interactive agent's loop returns after its terminal state, so nothing will
-ever read a message sent to it afterwards. Messaging it must say so instead of
-reporting delivery, waiting on it must return at once, and its completion report
-must carry the ids of the reports it actually filed so the parent does not have
-to go asking.
-"""
-
-from __future__ import annotations
-
-import json
-from typing import TYPE_CHECKING, Any, cast
-
-import pytest
-from agents.tool_context import ToolContext
-
-from strix.core.agents import AgentCoordinator
-from strix.report.state import ReportState, set_global_report_state
-from strix.tools.agents_graph.tools import agent_finish, send_message_to_agent, wait_for_agents
-
-
-if TYPE_CHECKING:
-    from collections.abc import Iterator
-    from pathlib import Path
-
-
-@pytest.fixture
-def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[ReportState]:
-    monkeypatch.chdir(tmp_path)
-    state = ReportState(run_name="test-run")
-    set_global_report_state(state)
-    yield state
-    set_global_report_state(None)
-
-
-async def _graph(*, interactive: bool) -> AgentCoordinator:
-    coordinator = AgentCoordinator()
-    await coordinator.register("root", "strix", parent_id=None)
-    await coordinator.register("child", "Validator", parent_id="root")
-    await coordinator.attach_runtime("root", resumable=interactive)
-    await coordinator.attach_runtime("child", resumable=interactive)
-    return coordinator
-
-
-async def _call(
-    tool: Any, coordinator: AgentCoordinator, agent_id: str, args: dict[str, Any], **extra: Any
-) -> dict[str, Any]:
-    ctx = ToolContext(
-        context={"coordinator": coordinator, "agent_id": agent_id, **extra},
-        tool_name=tool.name,
-        tool_call_id="call-1",
-        tool_arguments="{}",
-    )
-    raw: str = await tool.on_invoke_tool(ctx, json.dumps(args))
-    return cast("dict[str, Any]", json.loads(raw))
-
-
-# --- send_message_to_agent -------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_message_to_finished_non_interactive_child_is_not_delivered() -> None:
-    coordinator = await _graph(interactive=False)
-    await coordinator.set_status("child", "completed")
-
-    result = await _call(
-        send_message_to_agent,
-        coordinator,
-        "root",
-        {"target_agent_id": "child", "message": "did you file it?", "message_type": "query"},
-    )
-
-    assert result["success"] is False
-    assert result["delivery_status"] == "not_delivered"
-    assert result["target_status"] == "completed"
-    assert "list_reports" in result["error"]
-    assert coordinator.pending_counts.get("child", 0) == 0
-    assert coordinator.runtimes["child"].mailbox == []
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize("status", ["stopped", "failed", "crashed"])
-async def test_every_terminal_non_interactive_status_is_unreachable(status: str) -> None:
-    coordinator = await _graph(interactive=False)
-    await coordinator.set_status("child", status)
-
-    assert await coordinator.send("child", {"from": "root", "content": "hi"}) is False
-    assert await coordinator.reachability("child") == (False, status)
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize("status", ["running", "waiting"])
-async def test_message_to_live_child_is_delivered(status: str) -> None:
-    coordinator = await _graph(interactive=False)
-    await coordinator.set_status("child", status)
-
-    result = await _call(
-        send_message_to_agent,
-        coordinator,
-        "root",
-        {"target_agent_id": "child", "message": "wrap up"},
-    )
-
-    assert result["success"] is True
-    assert result["delivery_status"] == "delivered"
-    assert coordinator.pending_counts["child"] == 1
-
-
-@pytest.mark.asyncio
-async def test_message_to_finished_interactive_child_still_wakes_it() -> None:
-    # An interactive loop parks after finishing and resumes on a message.
-    coordinator = await _graph(interactive=True)
-    await coordinator.set_status("child", "completed")
-
-    result = await _call(
-        send_message_to_agent,
-        coordinator,
-        "root",
-        {"target_agent_id": "child", "message": "one more thing"},
-    )
-
-    assert result["success"] is True
-    assert coordinator.pending_counts["child"] == 1
-
-
-@pytest.mark.asyncio
-async def test_unknown_target_is_reported_as_not_found() -> None:
-    coordinator = await _graph(interactive=False)
-
-    result = await _call(
-        send_message_to_agent,
-        coordinator,
-        "root",
-        {"target_agent_id": "ghost", "message": "hello"},
-    )
-
-    assert result["success"] is False
-    assert result["target_status"] is None
-    assert "not found" in result["error"]
-
-
-# --- wait_for_agents -------------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_wait_returns_at_once_when_no_child_can_answer() -> None:
-    coordinator = await _graph(interactive=False)
-    await coordinator.set_status("child", "completed")
-    # The completion report was already consumed in an earlier turn.
-
-    result = await _call(
-        wait_for_agents,
-        coordinator,
-        "root",
-        {"reason": "waiting for validator", "timeout_seconds": 240},
-    )
-
-    assert result["wait_outcome"] == "no_active_agents"
-    assert result["agents"] == [{"agent_id": "child", "name": "Validator", "status": "completed"}]
-    assert coordinator.statuses["root"] == "running"
-
-
-@pytest.mark.asyncio
-async def test_wait_delivers_a_pending_report_before_checking_liveness() -> None:
-    coordinator = await _graph(interactive=False)
-    await coordinator.send("root", {"from": "child", "type": "completion", "content": "done"})
-    await coordinator.set_status("child", "completed")
-
-    result = await _call(wait_for_agents, coordinator, "root", {"timeout_seconds": 5})
-
-    assert result["wait_outcome"] == "message_arrived"
-    assert result["pending_messages"] == 1
-
-
-@pytest.mark.asyncio
-async def test_wait_still_parks_while_a_child_is_running() -> None:
-    coordinator = await _graph(interactive=False)
-
-    result = await _call(wait_for_agents, coordinator, "root", {"timeout_seconds": 1})
-
-    assert result["wait_outcome"] == "timeout"
-
-
-@pytest.mark.asyncio
-async def test_interactive_wait_parks_even_without_active_children() -> None:
-    # In an interactive run a finished child can be woken later, so parking is
-    # legitimate; the run loop's own auto-resume bounds the wait.
-    coordinator = await _graph(interactive=True)
-    await coordinator.set_status("child", "completed")
-
-    result = await _call(
-        wait_for_agents, coordinator, "root", {"timeout_seconds": 5}, interactive=True
-    )
-
-    assert result["wait_outcome"] == "waiting"
-
-
-# --- agent_finish ----------------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_agent_finish_lists_the_reports_the_child_filed(report_state: ReportState) -> None:
-    coordinator = await _graph(interactive=False)
-    mine = report_state.add_vulnerability_report(
-        title="IDOR on /api/audits", severity="high", agent_id="child", agent_name="Validator"
-    )
-    report_state.add_vulnerability_report(title="Root's own", severity="low", agent_id="root")
-
-    result = await _call(
-        agent_finish,
-        coordinator,
-        "child",
-        {"result_summary": "confirmed", "findings": ["IDOR confirmed"]},
-        parent_id="root",
-    )
-
-    assert result["filed_report_ids"] == [mine]
-    delivered = coordinator.runtimes["root"].mailbox
-    assert len(delivered) == 1
-    assert delivered[0]["filed_report_ids"] == [mine]
-    body = delivered[0]["content"]
-    assert f"- {mine} [HIGH] IDOR on /api/audits" in body
-    assert "Root's own" not in body
-
-
-@pytest.mark.asyncio
-async def test_agent_finish_states_explicitly_when_nothing_was_filed(
-    report_state: ReportState,
-) -> None:
-    coordinator = await _graph(interactive=False)
-    report_state.add_vulnerability_report(title="Someone else's", severity="low", agent_id="root")
-
-    result = await _call(
-        agent_finish,
-        coordinator,
-        "child",
-        {"result_summary": "nothing exploitable", "findings": ["ruled out X"]},
-        parent_id="root",
-    )
-
-    assert result["filed_report_ids"] == []
-    body = coordinator.runtimes["root"].mailbox[0]["content"]
-    assert "Vulnerability reports filed by this agent" in body
-    assert body.index("filed by this agent") < body.index("- (none)")
-
-
-@pytest.mark.asyncio
-async def test_agent_finish_without_report_state_still_completes() -> None:
-    set_global_report_state(None)
-    coordinator = await _graph(interactive=False)
-
-    result = await _call(
-        agent_finish, coordinator, "child", {"result_summary": "done"}, parent_id="root"
-    )
-
-    assert result["success"] is True
-    assert result["filed_report_ids"] == []

tests/test_cloud_cli.py

@@ -534,69 +534,6 @@ def test_insufficient_credits_exits_with_payment_code(monkeypatch: pytest.Monkey
     assert cloud.run_cloud(["scans", "start", "--domain-ids", "d1"]) == http.EXIT_PAYMENT
 
 
-def test_insufficient_credits_always_prints_topup_instruction(
-    monkeypatch: pytest.MonkeyPatch, capsys: Any
-) -> None:
-    monkeypatch.setattr(
-        http,
-        "request",
-        lambda *_a, **_k: FakeResponse(
-            status_code=402,
-            payload={"detail": "Out of credits.", "code": "scan_credit_limit_reached"},
-        ),
-    )
-    monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
-    argv = ["scans", "start", "--domain-ids", "d1", "--app-url", "https://app.strix.ai"]
-    assert cloud.run_cloud(argv) == http.EXIT_PAYMENT
-    output = " ".join(capsys.readouterr().out.split())
-    assert "Error: Out of credits." in output
-    assert "Next step:" in output
-    assert "strix cloud billing topup --credits <count>" in output
-    assert "https://app.strix.ai/settings/billing" in output
-    assert "strix cloud billing credits" in output
-
-
-def test_insufficient_credits_shows_platform_hint_once(
-    monkeypatch: pytest.MonkeyPatch, capsys: Any
-) -> None:
-    hint = "Buy credits at https://app.strix.ai/settings/billing. Then retry this request."
-    payload = {
-        "detail": f"Out of credits. {hint}",
-        "code": "scan_credit_limit_reached",
-        "hint": hint,
-        "topup_url": "https://app.strix.ai/settings/billing",
-    }
-    monkeypatch.setattr(
-        http, "request", lambda *_a, **_k: FakeResponse(status_code=402, payload=payload)
-    )
-    assert cloud.run_cloud(["scans", "start", "--domain-ids", "d1", "--json"]) == http.EXIT_PAYMENT
-    result = json.loads(capsys.readouterr().out)
-    assert result["error"] == "Out of credits."
-    assert result["next_step"] == hint
-    assert result["topup_url"] == "https://app.strix.ai/settings/billing"
-
-    monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
-    assert cloud.run_cloud(["scans", "start", "--domain-ids", "d1"]) == http.EXIT_PAYMENT
-    output = " ".join(capsys.readouterr().out.split())
-    assert output.count(hint) == 1
-    assert "Error: Out of credits." in output
-    assert f"Next step: {hint}" in output
-
-
-def test_payment_required_without_body_names_the_topup_command(
-    monkeypatch: pytest.MonkeyPatch, capsys: Any
-) -> None:
-    monkeypatch.setattr(
-        http, "request", lambda *_a, **_k: FakeResponse(status_code=402, payload={})
-    )
-    argv = ["scans", "start", "--domain-ids", "d1", "--json", "--app-url", "https://app.strix.ai"]
-    assert cloud.run_cloud(argv) == http.EXIT_PAYMENT
-    result = json.loads(capsys.readouterr().out)
-    assert result["error"] == "Not enough credits to run this command."
-    assert "strix cloud billing topup --credits <count>" in result["next_step"]
-    assert "https://app.strix.ai/settings/billing" in result["next_step"]
-
-
 def test_data_rejects_non_object() -> None:
     assert cloud.run_cloud(["scans", "start", "--data", "[1,2]"]) == http.EXIT_USAGE
     assert cloud.run_cloud(["scans", "start", "--data", "not json"]) == http.EXIT_USAGE

tests/test_cloud_source_upload.py

@@ -655,44 +655,3 @@ def test_incomplete_upload_credentials_delete_the_reserved_upload(
     monkeypatch.setattr(http, "request", fake_request)
     assert cloud.run_cloud(["scans", "start", "--source", str(tmp_path), "--yes", "--json"]) == 1
     assert ("DELETE", "/uploads/upload-incomplete") in paths
-
-
-def test_archive_source_is_rejected_with_directory_guidance(tmp_path: Path) -> None:
-    archive = tmp_path / "backend.zip"
-    with zipfile.ZipFile(archive, "w") as bundle:
-        bundle.writestr("app.py", "print('safe')\n")
-
-    with pytest.raises(http.CloudError, match="not an archive") as raised:
-        source_upload.prepare_source(
-            str(archive),
-            include_hidden=False,
-            include_sensitive=False,
-            include_archives=False,
-            exclude=[],
-        )
-    assert raised.value.next_step is not None
-    assert "--source" in raised.value.next_step
-    assert "--dry-run --show-files" in raised.value.next_step
-
-
-def test_oversize_archive_names_largest_files_and_exclude_guidance(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    (tmp_path / "app.py").write_text("print('safe')\n", encoding="utf-8")
-    (tmp_path / "big.bin").write_bytes(os.urandom(4096))
-    monkeypatch.setattr(source_upload, "MAX_ARCHIVE_BYTES", 1024)
-
-    with pytest.raises(http.CloudError, match=r"larger than the 0\.0 MiB upload limit") as raised:
-        source_upload.prepare_source(
-            str(tmp_path),
-            include_hidden=False,
-            include_sensitive=False,
-            include_archives=False,
-            exclude=[],
-        )
-    message = str(raised.value)
-    assert message.index("big.bin") < message.index("app.py")
-    assert raised.value.next_step is not None
-    assert "--exclude" in raised.value.next_step
-    assert "--dry-run --show-files" in raised.value.next_step
-    assert not list(tmp_path.glob("strix-source-*.zip"))

tests/test_config_loader.py

@@ -30,8 +30,6 @@ _LLM_ENV_KEYS = [
     "STRIX_FORCE_REQUIRED_TOOL_CHOICE",
     "LLM_TIMEOUT",
     "PERPLEXITY_API_KEY",
-    "EXA_API_KEY",
-    "STRIX_WEB_SEARCH_PROVIDER",
     # RuntimeSettings
     "STRIX_IMAGE",
     "STRIX_RUNTIME_BACKEND",
@@ -82,17 +80,6 @@ def test_read_json_overrides_maps_to_nested_settings(tmp_path: Path) -> None:
     }
 
 
-def test_read_json_overrides_maps_exa_and_provider(tmp_path: Path) -> None:
-    path = tmp_path / "cli-config.json"
-    path.write_text(
-        json.dumps({"env": {"EXA_API_KEY": "exa-key", "STRIX_WEB_SEARCH_PROVIDER": "exa"}}),
-        encoding="utf-8",
-    )
-    assert loader._read_json_overrides(path) == {
-        "integrations": {"exa_api_key": "exa-key", "web_search_provider": "exa"},
-    }
-
-
 def test_read_json_overrides_skips_keys_already_in_environ(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/test_import_warmup.py

@@ -1,10 +1,12 @@
-"""The import warm-up thread must never race the main thread into the engine.
+"""The import warm-up thread must never leave the import system poisoned.
 
-Two threads that enter the same package graph from different modules hold
-each other's import locks (warm-up: ``strix.core.runner`` -> ``agents``;
-main: ``agents.models.interface``). CPython breaks such a cycle by failing one
-of the imports, so the main thread waits for the warm-up before its first
-engine import.
+Field failure: the warm-up thread's ``strix.core.runner`` import and the main
+thread's ``strix.report`` import both walked the agents SDK graph, and the two
+held each other's import locks (report -> dedupe -> agents while runner ->
+hooks -> report.state). CPython's deadlock avoidance breaks such a cycle by
+failing one import, which strands finished submodules in ``sys.modules`` with
+their parent package gone — and the next import of one of those submodules
+crashes with "partially initialized module".
 """
 
 from __future__ import annotations
@@ -12,16 +14,10 @@ from __future__ import annotations
 import subprocess
 import sys
 import textwrap
-import threading
-from typing import TYPE_CHECKING
 
 from strix.llm import warmup
 
 
-if TYPE_CHECKING:
-    import pytest
-
-
 def _run(code: str) -> subprocess.CompletedProcess[str]:
     return subprocess.run(  # noqa: S603
         [sys.executable, "-c", textwrap.dedent(code)],
@@ -60,49 +56,44 @@ def test_check_duplicate_resolves_lazily() -> None:
     assert result.returncode == 0, result.stderr
 
 
-def test_wait_for_import_warmup_lets_main_thread_import_the_agents_graph() -> None:
+def test_failed_warm_import_purges_orphaned_submodules() -> None:
     result = _run(
         """
         import sys
 
-        from strix.llm.warmup import start_import_warmup, wait_for_import_warmup
+        from strix.llm.warmup import _warm
 
-        # Same shape as the CLI: warm-up starts, then the main thread needs a
-        # module from the middle of the agents graph.
-        start_import_warmup()
-        wait_for_import_warmup()
+        # A package whose import fails after a submodule already completed:
+        # CPython removes the package but leaves the submodule stranded.
+        import pathlib
+        import tempfile
 
-        from agents.models.interface import ModelTracing  # noqa: F401
+        root = pathlib.Path(tempfile.mkdtemp())
+        pkg = root / "stranded_pkg"
+        pkg.mkdir()
+        (pkg / "ok.py").write_text("VALUE = 1")
+        (pkg / "__init__.py").write_text("from . import ok\\nraise RuntimeError('boom')")
+        sys.path.insert(0, str(root))
 
-        assert "agents" in sys.modules
-        assert "agents.models" in sys.modules
-        assert "strix.core.runner" in sys.modules
+        _warm(("stranded_pkg",))
+
+        assert "stranded_pkg" not in sys.modules
+        assert "stranded_pkg.ok" not in sys.modules, "orphan survived the purge"
+
+        # And the subtree imports cleanly afterwards up to the real error.
+        try:
+            import stranded_pkg  # noqa: F401
+        except RuntimeError:
+            pass
+        else:
+            raise AssertionError("expected the package's own error")
         """
     )
     assert result.returncode == 0, result.stderr
 
 
-def test_wait_for_import_warmup_blocks_until_the_thread_finishes(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    release = threading.Event()
-    monkeypatch.setattr(warmup, "_warm", lambda _modules: release.wait())
-    monkeypatch.setattr(warmup, "_thread", None)
-    warmup.start_import_warmup(())
-
-    waiter = threading.Thread(target=warmup.wait_for_import_warmup)
-    waiter.start()
-    waiter.join(0.2)
-    assert waiter.is_alive(), "returned before the warm-up finished"
-
-    release.set()
-    waiter.join(5)
-    assert not waiter.is_alive()
-
-
-def test_failed_warm_import_does_not_raise() -> None:
-    warmup._warm(("strix_no_such_module_for_warmup_test",))
-
-
-def test_wait_for_import_warmup_is_a_no_op_without_a_thread() -> None:
-    warmup.wait_for_import_warmup()
+def test_purge_does_not_touch_preexisting_or_healthy_modules() -> None:
+    before = frozenset(sys.modules) - {"strix.llm.warmup"}
+    warmup._purge_orphaned_modules(before)
+    assert "strix.llm.warmup" in sys.modules  # parent chain intact -> kept
+    assert "strix" in sys.modules

tests/test_telemetry_errors.py

@@ -1,125 +0,0 @@
-"""Error beacons carry a category, phase, and exception class — never a message."""
-
-from __future__ import annotations
-
-from typing import Any
-
-import pytest
-import requests
-
-from strix.report.state import ReportState
-from strix.telemetry import posthog, report_error, scarf, set_scan_phase
-from strix.telemetry._common import exception_props
-
-
-PRIVATE_MESSAGE = "private message that must stay on the machine"
-
-
-def _capture(sent: list[dict[str, Any]], event: str, props: dict[str, Any]) -> bool:
-    sent.append({"event": event, **props})
-    return True
-
-
-def test_exception_props_uses_bare_name_for_builtins() -> None:
-    assert exception_props(ValueError(PRIVATE_MESSAGE)) == {"exception_type": "ValueError"}
-
-
-def test_exception_props_prefixes_third_party_top_level_package() -> None:
-    props = exception_props(requests.exceptions.ConnectTimeout(PRIVATE_MESSAGE))
-    assert props == {"exception_type": "requests.ConnectTimeout"}
-
-
-def _chained(cause: BaseException | None, *, explicit: bool) -> RuntimeError:
-    exc = RuntimeError("wrapped")
-    if explicit:
-        exc.__cause__ = cause
-        exc.__suppress_context__ = True
-    else:
-        exc.__context__ = cause
-    return exc
-
-
-def test_exception_props_reports_explicit_cause() -> None:
-    props = exception_props(_chained(ConnectionError(PRIVATE_MESSAGE), explicit=True))
-    assert props == {"exception_type": "RuntimeError", "exception_cause": "ConnectionError"}
-
-
-def test_exception_props_reports_implicit_context() -> None:
-    props = exception_props(_chained(KeyError("k"), explicit=False))
-    assert props["exception_cause"] == "KeyError"
-
-
-def test_exception_props_ignores_suppressed_context() -> None:
-    exc = _chained(None, explicit=True)
-    exc.__context__ = KeyError("k")
-    assert exception_props(exc) == {"exception_type": "RuntimeError"}
-
-
-def test_exception_props_unwraps_exception_group() -> None:
-    group = ExceptionGroup("tasks", [TimeoutError("t"), ValueError("v")])
-    assert exception_props(group) == {"exception_type": "TimeoutError"}
-
-
-@pytest.mark.parametrize("telemetry", [posthog, scarf])
-def test_error_event_carries_phase_and_class_but_no_message(
-    telemetry: Any,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    sent: list[dict[str, Any]] = []
-    monkeypatch.setattr(telemetry, "_send", lambda event, props: _capture(sent, event, props))
-    set_scan_phase("sandbox_init")
-
-    telemetry.error("scan_failed", RuntimeError(PRIVATE_MESSAGE))
-
-    assert len(sent) == 1
-    event = sent[0]
-    assert event["event"] == "error"
-    assert event["error_type"] == "scan_failed"
-    assert event["phase"] == "sandbox_init"
-    assert event["exception_type"] == "RuntimeError"
-    assert PRIVATE_MESSAGE not in repr(event)
-
-
-@pytest.mark.parametrize("telemetry", [posthog, scarf])
-def test_error_event_without_exception_omits_exception_fields(
-    telemetry: Any,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    sent: list[dict[str, Any]] = []
-    monkeypatch.setattr(telemetry, "_send", lambda event, props: _capture(sent, event, props))
-    set_scan_phase("startup")
-
-    telemetry.error("docker_not_installed")
-
-    assert sent[0]["error_type"] == "docker_not_installed"
-    assert sent[0]["phase"] == "startup"
-    assert "exception_type" not in sent[0]
-    assert "exception_cause" not in sent[0]
-
-
-def test_report_error_fans_out_to_both_backends(monkeypatch: pytest.MonkeyPatch) -> None:
-    sent: list[dict[str, Any]] = []
-    monkeypatch.setattr(posthog, "_send", lambda event, props: _capture(sent, event, props))
-    monkeypatch.setattr(scarf, "_send", lambda event, props: _capture(sent, event, props))
-
-    report_error("model_connection_failed", TimeoutError(PRIVATE_MESSAGE))
-
-    assert len(sent) == 2
-    assert {e["error_type"] for e in sent} == {"model_connection_failed"}
-    assert {e["exception_type"] for e in sent} == {"TimeoutError"}
-
-
-@pytest.mark.parametrize("telemetry", [posthog, scarf])
-def test_scan_ended_prefers_recorded_exit_reason(
-    telemetry: Any,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    state = ReportState()
-    state.scan_ended_exit_reason = "budget_exceeded"
-    sent: list[dict[str, Any]] = []
-    monkeypatch.setattr(telemetry, "_send", lambda event, props: _capture(sent, event, props))
-
-    telemetry.end(state, exit_reason="user_exit")
-
-    assert sent[0]["event"] == "scan_ended"
-    assert sent[0]["exit_reason"] == "budget_exceeded"

tests/test_viewer.py

@@ -11,7 +11,7 @@ from typing import TYPE_CHECKING
 from urllib.parse import urlsplit
 
 from strix.core.paths import latest_run_dir, runs_base_dir
-from strix.interface.viewer.cli import _state_label, run_view
+from strix.interface.viewer.cli import run_view
 from strix.interface.viewer.server import serve
 from strix.interface.viewer.transcript import (
     build_run_state,
@@ -115,14 +115,6 @@ def test_read_run_summary_surfaces_mcp_connection_status(tmp_path: Path) -> None
     assert read_run_summary(run_dir)["mcp_connection_status"] == roster
 
 
-def test_viewer_cli_labels_terminal_statuses() -> None:
-    assert _state_label({"status": "completed", "finished": True}) == "[#22c55e]finished[/]"
-    assert _state_label({"status": "stopped", "finished": True}) == "[#eab308]stopped[/]"
-    assert _state_label({"status": "interrupted", "finished": True}) == "[#eab308]interrupted[/]"
-    assert _state_label({"status": "failed", "finished": True}) == "[#ef4444]failed[/]"
-    assert _state_label({"status": "running", "finished": False}) == "[#eab308]live[/]"
-
-
 def test_read_missing_artifacts_return_defaults(tmp_path: Path) -> None:
     run_dir = _make_run(tmp_path, "empty", status="running", end_time=None)
     assert read_vulnerabilities(run_dir) == []

tests/test_wait_dedupe.py

@@ -41,8 +41,6 @@ def _fast_wait(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
 async def _context() -> dict[str, Any]:
     coordinator = AgentCoordinator()
     await coordinator.register("root", "strix", parent_id=None)
-    # A live child keeps the wait genuine: with nobody to hear from it returns at once.
-    await coordinator.register("child", "recon", parent_id="root")
     return {"agent_id": "root", "coordinator": coordinator}
 
 

tests/test_web_search.py

@@ -1,416 +0,0 @@
-"""Tests for web_search/web_get_contents provider selection and the Exa backend."""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING, Any
-
-import pytest
-import requests
-
-from strix.config.settings import IntegrationSettings
-from strix.interface.environment import _missing_web_search_vars
-from strix.tools.web_search import tool
-
-
-if TYPE_CHECKING:
-    from typing import Self
-
-
-class _FakeResponse:
-    def __init__(self, body: dict[str, Any]) -> None:
-        self._body = body
-        self.headers: dict[str, str] = {}
-
-    def __enter__(self) -> Self:
-        return self
-
-    def __exit__(self, *_exc: object) -> None:
-        return None
-
-    def raise_for_status(self) -> None:
-        return None
-
-    def json(self) -> dict[str, Any]:
-        return self._body
-
-
-def test_auto_prefers_exa_when_both_keys_set() -> None:
-    integrations = IntegrationSettings(PERPLEXITY_API_KEY="pk", EXA_API_KEY="ek")
-    assert tool._resolve_provider(integrations) == ("exa", "ek")
-
-
-def test_auto_falls_back_to_perplexity_when_only_perplexity_is_set() -> None:
-    integrations = IntegrationSettings(PERPLEXITY_API_KEY="pk")
-    assert tool._resolve_provider(integrations) == ("perplexity", "pk")
-
-
-def test_explicit_exa_ignores_a_configured_perplexity_key() -> None:
-    integrations = IntegrationSettings(
-        PERPLEXITY_API_KEY="pk",
-        EXA_API_KEY="ek",
-        STRIX_WEB_SEARCH_PROVIDER="exa",
-    )
-    assert tool._resolve_provider(integrations) == ("exa", "ek")
-
-
-def test_explicit_perplexity_ignores_a_configured_exa_key() -> None:
-    integrations = IntegrationSettings(
-        PERPLEXITY_API_KEY="pk",
-        EXA_API_KEY="ek",
-        STRIX_WEB_SEARCH_PROVIDER="perplexity",
-    )
-    assert tool._resolve_provider(integrations) == ("perplexity", "pk")
-
-
-def test_explicit_exa_without_a_key_names_only_exa() -> None:
-    integrations = IntegrationSettings(
-        PERPLEXITY_API_KEY="pk",
-        STRIX_WEB_SEARCH_PROVIDER="exa",
-    )
-    resolved = tool._resolve_provider(integrations)
-    assert isinstance(resolved, dict)
-    assert resolved["success"] is False
-    assert "EXA_API_KEY" in resolved["error"]
-    assert "PERPLEXITY_API_KEY" not in resolved["error"]
-
-
-def test_no_keys_names_both_providers() -> None:
-    resolved = tool._resolve_provider(IntegrationSettings())
-    assert isinstance(resolved, dict)
-    assert "EXA_API_KEY or PERPLEXITY_API_KEY" in resolved["error"]
-
-
-def test_exa_content_requests_summaries_and_renders_results(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    captured: dict[str, Any] = {}
-
-    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
-        captured["url"] = url
-        captured["headers"] = kwargs["headers"]
-        captured["json"] = kwargs["json"]
-        return _FakeResponse(
-            {
-                "results": [
-                    {
-                        "url": "https://nvd.example/cve",
-                        "title": "NVD entry",
-                        "summary": "  CVE-2024-0001 is a heap overflow.  ",
-                    },
-                    {"id": "https://blog.example/post"},
-                    "not-a-dict",
-                    {"title": "no url"},
-                ],
-            }
-        )
-
-    monkeypatch.setattr(requests, "post", fake_post)
-
-    content = tool._exa_content("ek", "OpenSSH 7.4 RCE?", "auto", 5)
-
-    assert captured["url"] == "https://api.exa.ai/search"
-    assert captured["headers"]["x-api-key"] == "ek"
-    assert "OpenSSH 7.4 RCE?" in captured["json"]["query"]
-    assert captured["json"]["type"] == "auto"
-    assert captured["json"]["numResults"] == 5
-    assert captured["json"]["contents"] == {"summary": {"query": tool._EXA_SUMMARY_PROMPT}}
-    assert content == (
-        "### NVD entry\nhttps://nvd.example/cve\nCVE-2024-0001 is a heap overflow.\n\n"
-        "### https://blog.example/post\nhttps://blog.example/post"
-    )
-
-
-def test_exa_content_renders_a_result_without_contents(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    monkeypatch.setattr(
-        requests,
-        "post",
-        lambda *_a, **_kw: _FakeResponse(
-            {"results": [{"url": "https://ex.example", "title": "Ex"}]}
-        ),
-    )
-    assert tool._exa_content("ek", "q", "auto", 5) == "### Ex\nhttps://ex.example"
-
-
-@pytest.mark.parametrize("body", [{}, {"results": None}, {"results": []}, {"results": ["x"]}])
-def test_exa_content_rejects_empty_results(
-    monkeypatch: pytest.MonkeyPatch, body: dict[str, Any]
-) -> None:
-    monkeypatch.setattr(requests, "post", lambda *_a, **_kw: _FakeResponse(body))
-    with pytest.raises(ValueError, match="no results"):
-        tool._exa_content("ek", "q", "auto", 5)
-
-
-def test_do_search_reports_empty_exa_results_as_unexpected(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    class _Settings:
-        integrations = IntegrationSettings(EXA_API_KEY="ek")
-
-    monkeypatch.setattr(tool, "load_settings", _Settings)
-    monkeypatch.setattr(requests, "post", lambda *_a, **_kw: _FakeResponse({}))
-
-    result = tool._do_search("q")
-
-    assert result["success"] is False
-    assert "unexpected response" in result["error"]
-
-
-@pytest.mark.parametrize(
-    ("env", "expected"),
-    [
-        ({}, ["EXA_API_KEY", "PERPLEXITY_API_KEY"]),
-        ({"EXA_API_KEY": "ek"}, []),
-        ({"PERPLEXITY_API_KEY": "pk"}, []),
-        ({"STRIX_WEB_SEARCH_PROVIDER": "exa", "PERPLEXITY_API_KEY": "pk"}, ["EXA_API_KEY"]),
-        ({"STRIX_WEB_SEARCH_PROVIDER": "exa", "EXA_API_KEY": "ek"}, []),
-        ({"STRIX_WEB_SEARCH_PROVIDER": "perplexity", "EXA_API_KEY": "ek"}, ["PERPLEXITY_API_KEY"]),
-        ({"STRIX_WEB_SEARCH_PROVIDER": "perplexity", "PERPLEXITY_API_KEY": "pk"}, []),
-    ],
-)
-def test_environment_validation_follows_provider_rules(
-    env: dict[str, str], expected: list[str]
-) -> None:
-    integrations = IntegrationSettings.model_validate(env)
-    assert _missing_web_search_vars(integrations) == expected
-
-
-def test_exa_search_type_and_num_results_are_configurable(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    captured: dict[str, Any] = {}
-
-    class _Settings:
-        integrations = IntegrationSettings(
-            EXA_API_KEY="ek",
-            STRIX_EXA_SEARCH_TYPE="deep-reasoning",
-            STRIX_EXA_NUM_RESULTS=3,
-        )
-
-    def fake_post(_url: str, **kwargs: Any) -> _FakeResponse:
-        captured["json"] = kwargs["json"]
-        return _FakeResponse({"results": [{"url": "https://ex.example", "title": "Ex"}]})
-
-    monkeypatch.setattr(tool, "load_settings", _Settings)
-    monkeypatch.setattr(requests, "post", fake_post)
-
-    assert tool._do_search("q")["success"] is True
-    assert captured["json"]["type"] == "deep-reasoning"
-    assert captured["json"]["numResults"] == 3
-
-
-def test_exa_page_text_requests_full_text_and_renders_pages(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    captured: dict[str, Any] = {}
-
-    def fake_post(url: str, **kwargs: Any) -> _FakeResponse:
-        captured["url"] = url
-        captured["headers"] = kwargs["headers"]
-        captured["json"] = kwargs["json"]
-        return _FakeResponse(
-            {
-                "results": [
-                    {
-                        "url": "https://nvd.example/cve",
-                        "title": "NVD entry",
-                        "text": "  Full advisory body.  ",
-                    },
-                    {"url": "https://empty.example", "text": "   "},
-                    "not-a-dict",
-                    {"text": "no url"},
-                ],
-            }
-        )
-
-    monkeypatch.setattr(requests, "post", fake_post)
-
-    content, fetched = tool._exa_page_text("ek", ["https://nvd.example/cve"])
-
-    assert captured["url"] == "https://api.exa.ai/contents"
-    assert captured["headers"]["x-api-key"] == "ek"
-    assert captured["json"] == {"urls": ["https://nvd.example/cve"], "text": True}
-    assert content == "### NVD entry\nhttps://nvd.example/cve\n\nFull advisory body."
-    assert fetched == {"https://nvd.example/cve"}
-
-
-def test_exa_page_text_truncates_a_long_page(monkeypatch: pytest.MonkeyPatch) -> None:
-    body = "A" * (tool._EXA_PAGE_MAX_CHARS + 500)
-    monkeypatch.setattr(
-        requests,
-        "post",
-        lambda *_a, **_kw: _FakeResponse(
-            {"results": [{"url": "https://ex.example", "text": body}]}
-        ),
-    )
-    content, _fetched = tool._exa_page_text("ek", ["https://ex.example"])
-    assert "truncated at" in content
-    assert content.count("A") == tool._EXA_PAGE_MAX_CHARS
-
-
-@pytest.mark.parametrize("body", [{}, {"results": []}, {"results": [{"url": "u"}]}])
-def test_exa_page_text_rejects_pages_without_text(
-    monkeypatch: pytest.MonkeyPatch, body: dict[str, Any]
-) -> None:
-    monkeypatch.setattr(requests, "post", lambda *_a, **_kw: _FakeResponse(body))
-    with pytest.raises(ValueError, match="no page contents"):
-        tool._exa_page_text("ek", ["https://ex.example"])
-
-
-@pytest.mark.parametrize("urls", [[], ["", "   "]])
-def test_do_get_contents_requires_a_url(urls: list[str]) -> None:
-    result = tool._do_get_contents(urls)
-    assert result["success"] is False
-    assert "at least one URL" in result["error"]
-
-
-def test_do_get_contents_caps_the_url_count() -> None:
-    urls = [f"https://ex{index}.example" for index in range(tool._EXA_MAX_CONTENT_URLS + 1)]
-    result = tool._do_get_contents(urls)
-    assert result["success"] is False
-    assert "Too many URLs" in result["error"]
-
-
-def test_do_get_contents_needs_an_exa_key(monkeypatch: pytest.MonkeyPatch) -> None:
-    class _Settings:
-        integrations = IntegrationSettings(PERPLEXITY_API_KEY="pk")
-
-    monkeypatch.setattr(tool, "load_settings", _Settings)
-
-    result = tool._do_get_contents(["https://ex.example"])
-
-    assert result["success"] is False
-    assert "EXA_API_KEY" in result["error"]
-
-
-def test_do_get_contents_refuses_a_perplexity_pinned_provider(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    class _Settings:
-        integrations = IntegrationSettings(
-            EXA_API_KEY="ek",
-            PERPLEXITY_API_KEY="pk",
-            STRIX_WEB_SEARCH_PROVIDER="perplexity",
-        )
-
-    monkeypatch.setattr(tool, "load_settings", _Settings)
-
-    result = tool._do_get_contents(["https://ex.example"])
-
-    assert result["success"] is False
-    assert "web_search" in result["error"]
-
-
-def test_do_get_contents_returns_page_text(monkeypatch: pytest.MonkeyPatch) -> None:
-    class _Settings:
-        integrations = IntegrationSettings(EXA_API_KEY="ek")
-
-    monkeypatch.setattr(tool, "load_settings", _Settings)
-    monkeypatch.setattr(tool, "_exa_page_text", lambda *_a: ("page", {"https://ex.example"}))
-
-    result = tool._do_get_contents([" https://ex.example "])
-
-    assert result == {
-        "success": True,
-        "urls": ["https://ex.example"],
-        "provider": "exa",
-        "content": "page",
-    }
-
-
-def test_do_get_contents_reports_urls_exa_did_not_return(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    class _Settings:
-        integrations = IntegrationSettings(EXA_API_KEY="ek")
-
-    monkeypatch.setattr(tool, "load_settings", _Settings)
-    monkeypatch.setattr(
-        requests,
-        "post",
-        lambda *_a, **_kw: _FakeResponse(
-            {"results": [{"url": "https://ok.example/", "text": "Body."}]}
-        ),
-    )
-
-    result = tool._do_get_contents(["https://ok.example", "https://blocked.example"])
-
-    assert result["success"] is True
-    assert result["urls"] == ["https://ok.example"]
-    assert result["failed_urls"] == ["https://blocked.example"]
-    assert "1 of 2" in result["warning"]
-    assert "blocked.example" not in result["content"]
-
-
-def test_normalize_url_folds_only_scheme_and_host() -> None:
-    assert tool._normalize_url("HTTPS://Ex.Example/Path/") == tool._normalize_url(
-        "https://ex.example/Path"
-    )
-    assert tool._normalize_url("https://ex.example/Path") != tool._normalize_url(
-        "https://ex.example/path"
-    )
-    assert tool._normalize_url("https://ex.example/p?Q=A") != tool._normalize_url(
-        "https://ex.example/p?q=a"
-    )
-
-
-def test_do_get_contents_omits_the_warning_when_every_page_returns(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    class _Settings:
-        integrations = IntegrationSettings(EXA_API_KEY="ek")
-
-    monkeypatch.setattr(tool, "load_settings", _Settings)
-    monkeypatch.setattr(
-        requests,
-        "post",
-        lambda *_a, **_kw: _FakeResponse(
-            {
-                "results": [
-                    {"url": "https://a.example", "text": "A."},
-                    {"url": "https://b.example", "text": "B."},
-                ]
-            }
-        ),
-    )
-
-    result = tool._do_get_contents(["https://a.example", "https://b.example"])
-
-    assert result["urls"] == ["https://a.example", "https://b.example"]
-    assert "failed_urls" not in result
-    assert "warning" not in result
-
-
-def test_do_get_contents_sanitizes_a_network_error(monkeypatch: pytest.MonkeyPatch) -> None:
-    class _Settings:
-        integrations = IntegrationSettings(EXA_API_KEY="ek")
-
-    def boom(*_args: Any, **_kwargs: Any) -> None:
-        raise requests.exceptions.ConnectionError
-
-    monkeypatch.setattr(tool, "load_settings", _Settings)
-    monkeypatch.setattr(requests, "post", boom)
-
-    result = tool._do_get_contents(["https://ex.example"])
-
-    assert result["success"] is False
-    assert "network error" in result["error"]
-    assert "ek" not in result["error"]
-
-
-def test_do_search_reports_the_provider_it_used(monkeypatch: pytest.MonkeyPatch) -> None:
-    class _Settings:
-        integrations = IntegrationSettings(EXA_API_KEY="ek")
-
-    monkeypatch.setattr(tool, "load_settings", _Settings)
-    monkeypatch.setattr(tool, "_exa_content", lambda *_a: "answer")
-
-    result = tool._do_search("OpenSSH 7.4 RCE?")
-
-    assert result == {
-        "success": True,
-        "query": "OpenSSH 7.4 RCE?",
-        "provider": "exa",
-        "content": "answer",
-    }