fix(telemetry): classify setup-mode TUI preflight and preparation failures

error events now carry phase (startup/preflight/sandbox_init/agent_setup/
agent_loop) and the exception class name (plus its cause), never the message
or trace. Startup and preflight failures that exit(1) before the scan starts
are beaconed with a stable error_type instead of vanishing. scan_ended
distinguishes budget_exceeded, rate_limited, and headless agent_stopped
from user_exit.

README.md

@@ -273,7 +273,6 @@ 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,6 +80,22 @@ 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>
@@ -159,7 +175,8 @@ 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
+# Optional: Enable web search (Exa preferred, Perplexity supported)
+export EXA_API_KEY="..."
 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 via Perplexity           |
+| Web Search     | Real-time OSINT with Exa or 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_search
+from strix.tools.web_search.tool import web_get_contents, web_search
 
 
 if TYPE_CHECKING:
@@ -579,6 +579,7 @@ _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,6 +120,10 @@ 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
 
@@ -128,6 +132,25 @@ 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/runner.py

@@ -45,6 +45,7 @@ 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,
@@ -116,6 +117,13 @@ 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.
 
@@ -313,6 +321,7 @@ 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,
@@ -322,6 +331,7 @@ 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"]
 
@@ -573,6 +583,7 @@ 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,
@@ -610,6 +621,7 @@ 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")
@@ -622,6 +634,7 @@ 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/environment.py

@@ -8,17 +8,29 @@ from rich.console import Console
 from rich.panel import Panel
 from rich.text import Text
 
-from strix.config import codex, load_settings
+from strix.config import IntegrationSettings, 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()
@@ -33,6 +45,7 @@ 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
@@ -46,8 +59,7 @@ def validate_environment() -> None:
     if not settings.llm.api_base:
         missing_optional_vars.append("LLM_API_BASE")
 
-    if not settings.integrations.perplexity_api_key:
-        missing_optional_vars.append("PERPLEXITY_API_KEY")
+    missing_optional_vars.extend(_missing_web_search_vars(settings.integrations))
 
     if missing_required_vars:
         error_text = Text()
@@ -89,7 +101,14 @@ 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 (enables real-time research)\n",
+                        " - 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",
                         style="white",
                     )
                 elif var == "STRIX_REASONING_EFFORT":
@@ -116,6 +135,8 @@ 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",
@@ -134,6 +155,7 @@ 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)",
@@ -161,6 +183,7 @@ 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")
 
@@ -208,6 +231,7 @@ 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,7 +41,8 @@ from strix.interface.update_check import (
 from strix.interface.utils import (
     build_final_stats_text,
 )
-from strix.telemetry import posthog, scarf
+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.logging import configure_dependency_logging
 
 
@@ -395,15 +396,18 @@ 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)
@@ -450,8 +454,6 @@ def main() -> None:
 
         sys.exit(run_cloud(sys.argv[2:]))
 
-    from strix.llm.warmup import start_import_warmup
-
     start_import_warmup()
 
     args = parse_arguments()
@@ -466,6 +468,9 @@ 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)
 
@@ -477,18 +482,21 @@ 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:
+    except Exception as exc:
         exit_reason = "error"
-        posthog.error("unhandled_exception")
-        scarf.error("unhandled_exception")
+        report_error("unhandled_exception", exc)
         raise
     finally:
         report_state = get_global_report_state()

strix/interface/tui/runtime.py

@@ -37,6 +37,7 @@ 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
 
 
@@ -138,11 +139,13 @@ 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
 
@@ -181,7 +184,11 @@ class GoTuiRuntime:
             candidate.target = list(self.controller.targets)
             candidate.target_list = []
             build_targets_info(candidate)
-        prepare_run(candidate)
+        try:
+            prepare_run(candidate)
+        except Exception as exc:
+            report_error("scan_preparation_failed", exc)
+            raise
         telemetry_start(candidate)
 
         vars(self.args).update(vars(candidate))
@@ -195,13 +202,21 @@ 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"
@@ -240,6 +255,9 @@ 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,6 +19,7 @@ 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
 
 
@@ -1602,7 +1603,8 @@ def check_docker_connection() -> Any:
 
     try:
         return docker.from_env()
-    except DockerException:
+    except DockerException as exc:
+        report_error("docker_unavailable", exc)
         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 = "[#eab308]live[/]" if live else "[#22c55e]finished[/]"
+    state_label = _state_label(summary)
     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,6 +107,18 @@ 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,20 +1,22 @@
 """Background pre-import of the heavy scan dependencies.
 
 The scan engine's import graph (the agents SDK, OpenAI client, LiteLLM, the
-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.
+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.
 """
 
 from __future__ import annotations
 
 import importlib
 import logging
-import sys
 import threading
 
 
@@ -24,45 +26,17 @@ 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:
@@ -72,11 +46,15 @@ def start_import_warmup(modules: tuple[str, ...] = WARMUP_MODULES) -> threading.
     runtime that has no local Docker) warm a narrower set.
     """
     global _thread  # noqa: PLW0603
-    with _lock:
-        if _thread is not None:
-            return _thread
+    if _thread is None:
         _thread = threading.Thread(
             target=_warm, args=(modules,), name="strix-import-warmup", daemon=True
         )
         _thread.start()
-        return _thread
+    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()

strix/skills/__init__.py

@@ -1,6 +1,5 @@
 import logging
 import re
-import threading
 from collections import Counter
 from collections.abc import Iterator
 from pathlib import Path
@@ -8,7 +7,6 @@ from typing import TypeGuard
 
 import yaml
 
-from strix.telemetry import posthog, scarf
 from strix.utils.resource_paths import get_strix_resource_path
 
 
@@ -241,16 +239,19 @@ 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)
 
-    threading.Thread(target=_send, daemon=True).start()
+def get_loaded_skill_names() -> list[str]:
+    """Distinct skills loaded so far in this process (custom skills collapse to ``"custom"``)."""
+    return sorted(_LOADED_SKILLS)
 
 
 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 and error types (not messages or stack traces)\
+**Session Errors:** Duration, the failure category, the scan phase, and the exception class name (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 are loaded\
+**Feature Usage:** Which built-in skills were used during a scan (reported once, at scan end)\
 **Aggregate Metrics:** Vulnerability counts by severity and weakness category (CWE)
 
 ### What We **Never** Collect

strix/telemetry/__init__.py

@@ -1,7 +1,19 @@
 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
+from typing import Any, cast
 from uuid import uuid4
 
 
@@ -54,3 +54,43 @@ 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,10 +4,14 @@ 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,
 )
 
@@ -35,7 +39,12 @@ def _send(event: str, properties: dict[str, Any]) -> bool:
             "api_key": _POSTHOG_PUBLIC_API_KEY,
             "event": event,
             "distinct_id": SESSION_ID,
-            "properties": properties,
+            "properties": {
+                **properties,
+                "$lib": "strix-cli",
+                "$lib_version": get_version(),
+                "$process_person_profile": False,
+            },
         }
         with requests.post(f"{_POSTHOG_HOST}/capture/", json=payload, timeout=SEND_TIMEOUT):
             pass
@@ -82,16 +91,6 @@ 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
@@ -130,6 +129,7 @@ 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,6 +180,12 @@ def viewer_agent_steered() -> None:
     _send("viewer_agent_steered", {**base_props()})
 
 
-def error(error_type: str) -> None:
-    props = {**base_props(), "error_type": error_type}
+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))
     _send("error", props)

strix/telemetry/scarf.py

@@ -7,10 +7,13 @@ 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,
 )
@@ -90,17 +93,6 @@ 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
@@ -140,14 +132,18 @@ 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) -> None:
+def error(error_type: str, exc: BaseException | None = None) -> 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/web_search/tool.py

@@ -1,11 +1,12 @@
-"""``web_search`` — Perplexity-backed security-focused web search."""
+"""Security-focused web research tools (Exa or Perplexity)."""
 
 from __future__ import annotations
 
 import asyncio
 import json
 import logging
-from typing import Any
+from typing import TYPE_CHECKING, Any, cast
+from urllib.parse import urlsplit, urlunsplit
 
 import requests
 from agents import RunContextWrapper, function_tool
@@ -13,6 +14,10 @@ from agents import RunContextWrapper, function_tool
 from strix.config import load_settings
 
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -41,22 +46,7 @@ Structure your response to be comprehensive yet concise, emphasizing the most cr
 security implications and details."""
 
 
-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])
-
+def _perplexity_content(api_key: str, query: str) -> str:
     url = "https://api.perplexity.ai/chat/completions"
     headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
     payload = {
@@ -66,61 +56,269 @@ def _do_search(query: str) -> dict[str, Any]:  # noqa: PLR0911 - each error clas
             {"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:
-        with requests.post(url, headers=headers, json=payload, timeout=300) as response:
-            response.raise_for_status()
-            content = response.json()["choices"][0]["message"]["content"]
+        return fetch()
     except requests.exceptions.Timeout:
-        logger.warning("web_search timed out")
-        return {
-            "success": False,
-            "error": "Web search timed out. Try again or shorten the query",
-        }
+        logger.warning("%s timed out", tool)
+        return {"success": False, "error": f"{tool} timed out. Try again or narrow the request"}
     except requests.exceptions.HTTPError as exc:
         status = exc.response.status_code if exc.response is not None else None
-        logger.exception("web_search HTTP error status=%s", status)
+        logger.exception("%s HTTP error status=%s", tool, status)
         if status is not None and 400 <= status < 500:
-            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",
-        }
+            return {"success": False, "error": rejected_hint}
+        return {"success": False, "error": f"{tool} service is unavailable. Try again later"}
     except requests.exceptions.RequestException:
-        logger.exception("web_search network error")
-        return {
-            "success": False,
-            "error": "Web search network error. Try again later",
-        }
+        logger.exception("%s network error", tool)
+        return {"success": False, "error": f"{tool} network error. Try again later"}
     except (KeyError, IndexError, ValueError):
-        logger.exception("web_search response shape unexpected")
-        return {
-            "success": False,
-            "error": "Web search returned an unexpected response. Try again",
-        }
+        logger.exception("%s response shape unexpected", tool)
+        return {"success": False, "error": f"{tool} returned an unexpected response. Try again"}
     except Exception:
-        logger.exception("web_search failed")
+        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:
         return {
             "success": False,
-            "error": "Web search failed unexpectedly",
+            "error": f"Too many URLs. Pass at most {_EXA_MAX_CONTENT_URLS} per call",
         }
-    else:
+
+    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")
         return {
-            "success": True,
-            "query": query,
-            "content": content,
+            "success": False,
+            "error": (
+                "Page fetching needs the Exa provider "
+                "(operator pinned STRIX_WEB_SEARCH_PROVIDER to perplexity). "
+                "Use web_search instead"
+            ),
         }
 
+    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 via Perplexity — your primary research tool.
+    """Real-time web search (Exa or Perplexity) — your primary research tool.
 
     Use it liberally for anything that's not in your training data:
 
@@ -150,6 +348,12 @@ 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):
 
@@ -177,3 +381,33 @@ 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_config_loader.py

@@ -30,6 +30,8 @@ _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",
@@ -80,6 +82,17 @@ 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,12 +1,10 @@
-"""The import warm-up thread must never leave the import system poisoned.
+"""The import warm-up thread must never race the main thread into the engine.
 
-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".
+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.
 """
 
 from __future__ import annotations
@@ -14,10 +12,16 @@ 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)],
@@ -56,44 +60,49 @@ def test_check_duplicate_resolves_lazily() -> None:
     assert result.returncode == 0, result.stderr
 
 
-def test_failed_warm_import_purges_orphaned_submodules() -> None:
+def test_wait_for_import_warmup_lets_main_thread_import_the_agents_graph() -> None:
     result = _run(
         """
         import sys
 
-        from strix.llm.warmup import _warm
+        from strix.llm.warmup import 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
+        # 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()
 
-        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))
+        from agents.models.interface import ModelTracing  # noqa: F401
 
-        _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 "agents" in sys.modules
+        assert "agents.models" in sys.modules
+        assert "strix.core.runner" in sys.modules
         """
     )
     assert result.returncode == 0, result.stderr
 
 
-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
+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()

tests/test_telemetry_errors.py

@@ -0,0 +1,125 @@
+"""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 run_view
+from strix.interface.viewer.cli import _state_label, run_view
 from strix.interface.viewer.server import serve
 from strix.interface.viewer.transcript import (
     build_run_state,
@@ -115,6 +115,14 @@ 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_web_search.py

@@ -0,0 +1,416 @@
+"""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",
+    }