Fall back to token-based cost estimation when observed LiteLLM cost is missing

* fix(proxy,tooling): serialize+reconnect Caido client, actionable HTTPQL errors, sandbox tool guidance

Addresses the top recurring agent tool-call failures observed in telemetry:

- proxy: the shared Caido client had no locking or reconnect, so concurrent
  agent calls raced ("Transport is already connected") and a dead transport
  poisoned the rest of the run ("Connector is closed"/"Server disconnected").
  Add an asyncio lock + bounded reconnect in caido_api.call_with_client (sandbox
  path) and a scan-wide caido_lock in the run context that host-side proxy tools
  hold around every call. Deterministic errors are not retried.
- proxy: list_requests now returns Caido's exact parser message, echoes the
  offending query, and includes a corrected-syntax hint so agents self-correct
  instead of retrying a broken HTTPQL filter.
- shell/prompt: document that write_stdin requires a process started with
  tty=true; nudge toward writing Python to a file over deeply-nested one-liners;
  note the venv pre-installs common libs.
- agent-browser: distinguish daemon/connection failures (run doctor, don't loop)
  from malformed commands; invoke directly (no sh -c wrapper).
- containers: use POSIX '.' instead of the bashism 'source' in generated rc
  files (fixes 'sh: source: not found'); add file + xxd and pre-install
  requests/httpx/beautifulsoup4/lxml/pyjwt/cryptography in the sandbox venv.
- tests: cover proxy serialization/reconnect/no-retry and HTTPQL errors.

* fix(proxy): host-side reconnect, close stale clients, don't retry mutations

Addresses Greptile review on the reconnect logic:

- Host path had no reconnect: a dead shared context client (Caido restart /
  network blip) previously disabled proxy tools for the rest of the scan. Add
  SharedCaidoClient, a serialized reconnect-safe holder stored once per scan in
  the run context and shared across agents. On a dead transport it rebuilds via
  reconnect_caido, which re-selects the SAME Caido project (preserving captured
  traffic) instead of creating a new empty one.
- Don't repeat completed mutations: call_with_client / SharedCaidoClient.call
  take idempotent=. Reads retry once on reconnect; replay + scope
  create/update/delete heal the client but re-raise instead of risking a
  double-apply.
- Don't leak replaced clients: the stale client is aclose()d (best-effort) on
  every reconnect.
- Extend tests to cover close-on-reconnect, non-idempotent re-raise, and the
  SharedCaidoClient holder.

* fix(proxy): close replacement Caido client when project.select fails

Addresses Greptile P1: in reconnect_caido (and bootstrap_caido) a successful
connect() followed by a failing project.select()/create() discarded the
connected client without closing it, so a missing/unavailable project could
leak a transport on every retry. Close the client before re-raising.

---------

Co-authored-by: Alex Schapiro <bearsyankees@gmail.com>

containers/Dockerfile

@@ -24,6 +24,7 @@ RUN apt-get update && \
     python3 python3-pip python3-dev python3-venv python3-setuptools \
     golang-go \
     net-tools dnsutils whois \
+    file xxd \
     jq parallel ripgrep grep \
     less man-db procps htop \
     iproute2 iputils-ping netcat-traditional \
@@ -192,6 +193,8 @@ RUN mkdir -p /workspace && chown -R pentester:pentester /workspace /app
 USER pentester
 RUN python3 -m venv /app/.venv && \
     /app/.venv/bin/pip install --no-cache-dir caido-sdk-client && \
+    /app/.venv/bin/pip install --no-cache-dir \
+        requests httpx beautifulsoup4 lxml pyjwt cryptography && \
     /app/.venv/bin/pip install --no-cache-dir -r /home/pentester/tools/jwt_tool/requirements.txt && \
     printf '%s\n' \
         '#!/bin/bash' \

containers/docker-entrypoint.sh

@@ -91,10 +91,13 @@ http_proxy=http://127.0.0.1:${CAIDO_PORT}
 https_proxy=http://127.0.0.1:${CAIDO_PORT}
 EOF
 
-echo "source /etc/profile.d/proxy.sh" >> ~/.bashrc
-echo "source /etc/profile.d/proxy.sh" >> ~/.zshrc
+# Use POSIX `.` (not the bashism `source`) so these lines are safe when the rc
+# files are read by a POSIX shell (e.g. `sh -lc`), which otherwise fails with
+# "source: not found". `.` is understood by bash, zsh, and dash alike.
+echo ". /etc/profile.d/proxy.sh" >> ~/.bashrc
+echo ". /etc/profile.d/proxy.sh" >> ~/.zshrc
 
-source /etc/profile.d/proxy.sh
+. /etc/profile.d/proxy.sh
 
 echo "✅ System-wide proxy configuration complete"
 

strix/agents/prompts/system_prompt.jinja

@@ -168,9 +168,24 @@ EFFICIENCY TACTICS:
 - Download additional tools as needed for specific tasks
 - Run multiple scans in parallel when possible
 - Load the most relevant skill before starting a specialized testing workflow if doing so will improve accuracy, speed, or tool usage
-- Use `exec_command` for Python code: write reusable scripts under
-  `/workspace/scratch/` and run them with `python3`. For one-off snippets,
-  `python3 -c` or a here-document is acceptable.
+- Use `exec_command` for Python code: write reusable scripts to a file and
+  run them with `python3 script.py`. For one-off snippets, `python3 -c` or a
+  here-document is acceptable, but avoid deeply nested quotes/parentheses — if
+  a snippet needs complex quoting or is more than a few lines, write it to a
+  file first to prevent syntax errors.
+- Before importing a third-party Python library, make sure it is installed. The
+  sandbox's `python3` runs inside a preconfigured virtualenv that ships
+  `requests`, `httpx`, `beautifulsoup4` (bs4), `lxml`, `pyjwt`, and
+  `cryptography`; for anything else prefer the stdlib or run `pip install <pkg>`
+  (it installs into that active venv) before importing, rather than letting the
+  script fail with `ModuleNotFoundError`.
+- `exec_command` runs each command in a fresh non-interactive shell (plain
+  pipes, no TTY). To drive an interactive or long-running process with
+  `write_stdin` — REPLs, `ssh`/`nc`/`ftp`, `msfconsole`, or to send Ctrl-C —
+  you MUST start it with `exec_command(cmd="...", tty=true)` and then
+  `write_stdin(session_id=<id>, chars="...")`. Calling `write_stdin` on a
+  default (non-TTY) command or on a process that has already exited fails with
+  "stdin is not available".
 - For Caido proxy automation inside Python, explicitly import from
   `caido_api`:
   `from caido_api import list_requests, view_request, repeat_request, list_sitemap, view_sitemap_entry, scope_rules`
@@ -186,7 +201,7 @@ EFFICIENCY TACTICS:
 VALIDATION REQUIREMENTS:
 - Full validation required - no assumptions
 - Demonstrate concrete impact with evidence
-- Consider business context for severity assessment
+- Consider business context for severity assessment — check whether the target is a demo/sandbox environment or content meant to be public, and factor that in
 - Independent verification through subagent
 - Document complete attack chain
 - Keep going until you find something that matters
@@ -240,12 +255,18 @@ AGENT ISOLATION & SANDBOXING:
 - All agents share the same /workspace directory and proxy history
 - Agents can see each other's files and proxy traffic for better collaboration
 
+DISK & SCRATCH HYGIENE:
+- /workspace is a shared, finite disk used by all agents at once — be a considerate tenant
+- Prefer bounded recon: scope crawls and scans by depth, duration, and target rather than "collect everything"
+- Redirect large tool output to a file, and once you've extracted what you need (e.g. a URL/endpoint list), remove the raw output
+- If disk gets tight or a write fails for space, check what's large under /workspace and clean up files from your own task; leave another agent's files unless you've confirmed they're no longer in use
+
 MANDATORY INITIAL PHASES:
 
 BLACK-BOX TESTING - PHASE 1 (RECON & MAPPING):
 - COMPLETE full reconnaissance: subdomain enumeration, port scanning, service detection
 - MAP entire attack surface: all endpoints, parameters, APIs, forms, inputs
-- CRAWL thoroughly: spider all pages (authenticated and unauthenticated), discover hidden paths, analyze JS files
+- CRAWL thoroughly: spider all pages (authenticated and unauthenticated), discover hidden paths, analyze JS files — keep each crawl bounded by depth/duration, and tidy up raw output once endpoints are extracted
 - ENUMERATE technologies: frameworks, libraries, versions, dependencies
 - Reconnaissance should normally happen before targeted vulnerability discovery unless the correct next move is already obvious or the user/system explicitly asks to prioritize a specific area first
 - ONLY AFTER comprehensive mapping → proceed to vulnerability testing

strix/config/settings.py

@@ -56,6 +56,8 @@ class RuntimeSettings(BaseSettings):
     # on large repos). Above this, the user must bind-mount via ``--mount``.
     # Set to 0 (or less) to disable the pre-flight check entirely.
     max_local_copy_mb: int = Field(default=1024, alias="STRIX_MAX_LOCAL_COPY_MB")
+    # Max screenshot/image tool outputs kept live per agent context (0 = none).
+    max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES")
 
 
 class TelemetrySettings(BaseSettings):

strix/core/agents.py

@@ -10,6 +10,8 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal, cast
 
+from strix.core.sessions import session_write_lock
+
 
 if TYPE_CHECKING:
     from agents.items import TResponseInputItem
@@ -137,7 +139,8 @@ class AgentCoordinator:
             )
             return False
         try:
-            await session.add_items([self._message_to_session_item(message)])
+            async with session_write_lock(session):
+                await session.add_items([self._message_to_session_item(message)])
         except Exception:
             logger.exception(
                 "agent.send failed to append to SDK session target=%s",

strix/core/execution.py

@@ -17,7 +17,11 @@ from openai import APIError
 
 from strix.core.hooks import BudgetExceededError
 from strix.core.inputs import child_initial_input
-from strix.core.sessions import open_agent_session, strip_all_images_from_session
+from strix.core.sessions import (
+    enforce_image_budget,
+    open_agent_session,
+    strip_all_images_from_session,
+)
 
 
 if TYPE_CHECKING:
@@ -349,6 +353,13 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
     while True:
         try:
             await coordinator.mark_running(agent_id)
+            if session is not None:
+                max_images = context.get("max_context_images")
+                if isinstance(max_images, int):
+                    try:
+                        await enforce_image_budget(session, max_images)
+                    except Exception:
+                        logger.exception("image-budget enforcement failed for %s", agent_id)
             stream = Runner.run_streamed(
                 agent,
                 input=input_data,

strix/core/inputs.py

@@ -13,6 +13,7 @@ from strix.config.models import (
     is_known_openai_bare_model,
     model_supports_reasoning,
 )
+from strix.core.sessions import scrub_images_from_items
 
 
 if TYPE_CHECKING:
@@ -161,7 +162,11 @@ def child_initial_input(
     """
     parts: list[str] = []
     if parent_history:
-        rendered = json.dumps(parent_history, ensure_ascii=False, default=str)
+        rendered = json.dumps(
+            scrub_images_from_items(parent_history),
+            ensure_ascii=False,
+            default=str,
+        )
         parts.append(
             "== Inherited context from parent (background only) ==\n"
             f"{rendered}\n"

strix/core/runner.py

@@ -282,11 +282,16 @@ async def run_strix_scan(
         context: dict[str, Any] = {
             "coordinator": coordinator,
             "sandbox_session": bundle["session"],
+            # One ``SharedCaidoClient`` is reused by every agent in the scan
+            # (child contexts are shallow copies via ``dict(parent_ctx)``). It
+            # serializes access to the non-concurrency-safe GraphQL transport
+            # and rebuilds it if it dies mid-scan.
             "caido_client": bundle["caido_client"],
             "agent_id": root_id,
             "parent_id": None,
             "interactive": interactive,
             "spawn_child_agent": spawn_child_agent,
+            "max_context_images": settings.runtime.max_context_images,
         }
 
         root_session = open_agent_session(root_id, agents_db)

strix/core/sessions.py

@@ -2,64 +2,149 @@
 
 from __future__ import annotations
 
-import contextlib
+import asyncio
+import logging
 from typing import TYPE_CHECKING, Any, cast
+from weakref import WeakKeyDictionary
 
 from agents.memory import SQLiteSession
 
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from pathlib import Path
 
     from agents.items import TResponseInputItem
     from agents.memory import Session
 
 
+logger = logging.getLogger(__name__)
+
+
 def open_agent_session(agent_id: str, path: Path) -> SQLiteSession:
     path.parent.mkdir(parents=True, exist_ok=True)
     return SQLiteSession(session_id=agent_id, db_path=path)
 
 
 _IMAGE_REJECTED_TEXT = "[image rejected by the model]"
+_IMAGE_ELIDED_TEXT = "[older screenshot elided to bound context memory]"
+_INHERITED_IMAGE_TEXT = "[screenshot omitted from inherited context]"
+
+
+def _output_has_image(item_dict: dict[str, Any]) -> bool:
+    return (
+        item_dict.get("type") == "function_call_output"
+        and isinstance(item_dict.get("output"), list)
+        and any(isinstance(b, dict) and b.get("type") == "input_image" for b in item_dict["output"])
+    )
+
+
+def _elided_output(item_dict: dict[str, Any], text: str) -> dict[str, Any]:
+    # Replace only image blocks; sibling text blocks are preserved.
+    output = item_dict.get("output")
+    blocks = output if isinstance(output, list) else []
+    return {
+        "type": "function_call_output",
+        "call_id": item_dict.get("call_id"),
+        "output": [
+            {"type": "input_text", "text": text}
+            if isinstance(block, dict) and block.get("type") == "input_image"
+            else block
+            for block in blocks
+        ],
+    }
+
+
+_session_write_locks: WeakKeyDictionary[Session, asyncio.Lock] = WeakKeyDictionary()
+
+
+def session_write_lock(session: Session) -> asyncio.Lock:
+    """Lock serialising all out-of-band writes to ``session``."""
+    lock = _session_write_locks.get(session)
+    if lock is None:
+        lock = asyncio.Lock()
+        _session_write_locks[session] = lock
+    return lock
+
+
+async def _rewrite_session(
+    session: Session,
+    transform: Callable[[list[Any]], tuple[list[Any], bool]],
+) -> bool:
+    """Read-modify-write a session under its write lock, restoring on failure."""
+    async with session_write_lock(session):
+        items = await session.get_items()
+        if not items:
+            return False
+        rebuilt, changed = transform(list(items))
+        if not changed:
+            return False
+        rebuilt_items = cast("list[TResponseInputItem]", rebuilt)
+        original_items = cast("list[TResponseInputItem]", list(items))
+        await session.clear_session()
+        try:
+            await session.add_items(rebuilt_items)
+        except Exception:
+            logger.exception("session rewrite failed; restoring original items")
+            await session.clear_session()
+            await session.add_items(original_items)
+            raise
+        return True
 
 
 async def strip_all_images_from_session(session: Session) -> bool:
-    items = await session.get_items()
-    if not items:
+    """Replace every image tool output with a text placeholder (rejection recovery)."""
+
+    def _transform(items: list[Any]) -> tuple[list[Any], bool]:
+        rebuilt: list[Any] = []
+        changed = False
+        for item in items:
+            item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None
+            if item_dict is not None and _output_has_image(item_dict):
+                rebuilt.append(_elided_output(item_dict, _IMAGE_REJECTED_TEXT))
+                changed = True
+            else:
+                rebuilt.append(item)
+        return rebuilt, changed
+
+    return await _rewrite_session(session, _transform)
+
+
+async def enforce_image_budget(session: Session, max_images: int) -> bool:
+    """Keep only the most recent ``max_images`` image outputs; elide older ones."""
+    if max_images < 0:
         return False
 
-    rebuilt: list[Any] = []
-    changed = False
-    for item in items:
-        item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None
-        if (
-            item_dict is not None
-            and item_dict.get("type") == "function_call_output"
-            and isinstance(item_dict.get("output"), list)
-            and any(
-                isinstance(b, dict) and b.get("type") == "input_image" for b in item_dict["output"]
-            )
-        ):
-            rebuilt.append(
-                {
-                    "type": "function_call_output",
-                    "call_id": item_dict.get("call_id"),
-                    "output": [{"type": "input_text", "text": _IMAGE_REJECTED_TEXT}],
-                },
-            )
-            changed = True
-        else:
-            rebuilt.append(item)
+    def _transform(items: list[Any]) -> tuple[list[Any], bool]:
+        image_indices = [
+            i
+            for i, item in enumerate(items)
+            if isinstance(item, dict) and _output_has_image(cast("dict[str, Any]", item))
+        ]
+        if len(image_indices) <= max_images:
+            return items, False
+        to_elide = set(image_indices[: len(image_indices) - max_images])
+        rebuilt = [
+            _elided_output(cast("dict[str, Any]", item), _IMAGE_ELIDED_TEXT)
+            if i in to_elide
+            else item
+            for i, item in enumerate(items)
+        ]
+        return rebuilt, True
 
-    if not changed:
-        return False
+    return await _rewrite_session(session, _transform)
 
-    rebuilt_items = cast("list[TResponseInputItem]", rebuilt)
-    await session.clear_session()
-    try:
-        await session.add_items(rebuilt_items)
-    except Exception:
-        with contextlib.suppress(Exception):
-            await session.add_items(rebuilt_items)
-        raise
-    return True
+
+def scrub_images_from_items(items: list[Any]) -> list[Any]:
+    """Return a copy of ``items`` with every image block replaced by text."""
+
+    def _scrub(obj: Any) -> Any:
+        if isinstance(obj, dict):
+            if obj.get("type") == "input_image":
+                return {"type": "input_text", "text": _INHERITED_IMAGE_TEXT}
+            return {k: _scrub(v) for k, v in obj.items()}
+        if isinstance(obj, list):
+            return [_scrub(v) for v in obj]
+        return obj
+
+    return [_scrub(item) for item in items]

strix/report/usage.py

@@ -19,6 +19,8 @@ class LLMUsageLedger:
         self._agent_usage: dict[str, Usage] = {}
         self._agent_metadata: dict[str, dict[str, str]] = {}
         self._total_cost = 0.0
+        self._observed_cost = 0.0
+        self._routed_estimated_cost = 0.0
 
     def record(
         self,
@@ -41,9 +43,14 @@ class LLMUsageLedger:
         if model:
             metadata["model"] = model
 
-        if not _is_litellm_routed(model):
-            estimated = _estimate_litellm_cost(usage, model)
-            if estimated:
+        estimated = _estimate_litellm_cost(usage, model)
+        if estimated:
+            if _is_litellm_routed(model):
+                # Fallback for routed models whose provider-reported cost never
+                # arrives (e.g. missing LiteLLM pricing map entry); only counted
+                # when no observed cost is recorded for the run.
+                self._routed_estimated_cost += estimated
+            else:
                 self._total_cost += estimated
 
         return True
@@ -51,14 +58,21 @@ class LLMUsageLedger:
     def record_observed_cost(self, cost: float) -> None:
         if isinstance(cost, int | float) and cost > 0:
             self._total_cost += float(cost)
+            self._observed_cost += float(cost)
+
+    def _effective_cost(self) -> float:
+        if self._observed_cost > 0:
+            return self._total_cost
+        return self._total_cost + self._routed_estimated_cost
 
     @property
     def total_cost(self) -> float:
-        return _round_cost(self._total_cost)
+        return _round_cost(self._effective_cost())
 
     def to_record(self) -> dict[str, Any]:
         record = serialize_usage(self._total_usage)
-        record["cost"] = _round_cost(self._total_cost)
+        effective_cost = self._effective_cost()
+        record["cost"] = _round_cost(effective_cost)
         record["agents"] = []
 
         agent_tokens = {aid: _resolve_total_tokens(u) for aid, u in self._agent_usage.items()}
@@ -67,7 +81,7 @@ class LLMUsageLedger:
             usage = self._agent_usage[agent_id]
             metadata = self._agent_metadata.get(agent_id, {})
             agent_cost = (
-                self._total_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0
+                effective_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0
             )
 
             agent_record = serialize_usage(usage)
@@ -88,6 +102,8 @@ class LLMUsageLedger:
         self._agent_usage.clear()
         self._agent_metadata.clear()
         self._total_cost = 0.0
+        self._observed_cost = 0.0
+        self._routed_estimated_cost = 0.0
 
         if not isinstance(raw_usage, dict):
             return

strix/runtime/caido_bootstrap.py

@@ -10,6 +10,7 @@ exposed-port URL for all subsequent SDK calls.
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
 import logging
 from typing import TYPE_CHECKING
@@ -79,23 +80,72 @@ async def _login_as_guest(
     raise RuntimeError(f"loginAsGuest failed after {attempts} attempts: {last_err}")
 
 
-async def bootstrap_caido(
+async def _aclose_quietly(client: Client) -> None:
+    """Best-effort close of a client whose setup failed; never raises."""
+    with contextlib.suppress(Exception):
+        await client.aclose()
+
+
+async def _connect_client(
     session: BaseSandboxSession,
     *,
     host_url: str,
     container_url: str,
 ) -> Client:
-    """Connect to the in-container Caido sidecar and select a fresh project."""
-    logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url)
-
     access_token = await _login_as_guest(session, container_url=container_url)
-
     client = Client(host_url, auth=TokenAuthOptions(token=access_token))
     await client.connect()
+    return client
 
-    project = await client.project.create(
-        CreateProjectOptions(name="sandbox", temporary=True),
-    )
-    await client.project.select(project.id)
+
+async def bootstrap_caido(
+    session: BaseSandboxSession,
+    *,
+    host_url: str,
+    container_url: str,
+) -> tuple[Client, str]:
+    """Connect to the in-container Caido sidecar and select a fresh project.
+
+    Returns the connected client and the id of the temporary project it
+    selected. The project id lets :func:`reconnect_caido` rebuild a dead
+    transport while staying on the *same* project (and its captured traffic)
+    instead of creating a new empty one.
+    """
+    logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url)
+
+    client = await _connect_client(session, host_url=host_url, container_url=container_url)
+    try:
+        project = await client.project.create(
+            CreateProjectOptions(name="sandbox", temporary=True),
+        )
+        await client.project.select(project.id)
+    except BaseException:
+        # Don't leak the connected transport if project setup fails.
+        await _aclose_quietly(client)
+        raise
     logger.info("Caido project selected: %s", project.id)
+    return client, str(project.id)
+
+
+async def reconnect_caido(
+    session: BaseSandboxSession,
+    *,
+    host_url: str,
+    container_url: str,
+    project_id: str,
+) -> Client:
+    """Rebuild a Caido client after its transport died, keeping the project.
+
+    Re-authenticates, reconnects, and re-selects the existing project so the
+    caller keeps access to the traffic captured before the disconnect.
+    """
+    logger.info("Reconnecting Caido client (host=%s, project=%s)", host_url, project_id)
+    client = await _connect_client(session, host_url=host_url, container_url=container_url)
+    try:
+        await client.project.select(project_id)
+    except BaseException:
+        # A missing/unavailable project must not leave the freshly-connected
+        # transport dangling — otherwise every retry leaks another one.
+        await _aclose_quietly(client)
+        raise
     return client

strix/runtime/docker_client.py

@@ -24,20 +24,25 @@ from __future__ import annotations
 
 import contextlib
 import logging
+import os
 import uuid
-from typing import Any
+from typing import Any, cast
 
+from agents.sandbox.errors import ExposedPortUnavailableError
 from agents.sandbox.manifest import Manifest
 from agents.sandbox.sandboxes.docker import (
     DockerSandboxClient,
+    DockerSandboxSession,
     _build_docker_volume_mounts,
     _docker_port_key,
     _manifest_requires_fuse,
     _manifest_requires_sys_admin,
 )
 from agents.sandbox.session.sandbox_session import SandboxSession
+from agents.sandbox.types import ExposedPortEndpoint
 from docker import errors as docker_errors  # type: ignore[import-untyped, unused-ignore]
 from docker.models.containers import Container  # type: ignore[import-untyped, unused-ignore]
+from docker.types import LogConfig  # type: ignore[import-untyped, unused-ignore]
 from docker.types import Mount as DockerSDKMount  # type: ignore[import-untyped, unused-ignore]
 from docker.utils import parse_repository_tag  # type: ignore[import-untyped, unused-ignore]
 from requests.exceptions import RequestException
@@ -46,6 +51,103 @@ from requests.exceptions import RequestException
 logger = logging.getLogger(__name__)
 
 
+_SANDBOX_NETWORK_ENV = "STRIX_DOCKER_SANDBOX_NETWORK"
+
+
+def _sandbox_network() -> str | None:
+    value = os.environ.get(_SANDBOX_NETWORK_ENV, "").strip()
+    return value or None
+
+
+def _apply_sandbox_network(create_kwargs: dict[str, Any]) -> None:
+    network = _sandbox_network()
+    if network:
+        create_kwargs["network"] = network
+        create_kwargs.pop("ports", None)
+
+
+def _apply_resource_limits(create_kwargs: dict[str, Any]) -> None:
+    """Apply optional cgroup resource caps from the environment. Unset/blank
+    values leave docker's default (unbounded), so this is opt-in per host."""
+    mem_limit = os.environ.get("STRIX_SANDBOX_MEM_LIMIT", "").strip()
+    if mem_limit:
+        create_kwargs["mem_limit"] = mem_limit
+
+    shm_size = os.environ.get("STRIX_SANDBOX_SHM_SIZE", "").strip()
+    if shm_size:
+        create_kwargs["shm_size"] = shm_size
+
+    cpus = os.environ.get("STRIX_SANDBOX_CPUS", "").strip()
+    if cpus:
+        with contextlib.suppress(ValueError, OverflowError):
+            nano_cpus = int(float(cpus) * 1_000_000_000)
+            if 0 < nano_cpus <= 2**63 - 1:
+                create_kwargs["nano_cpus"] = nano_cpus
+
+    pids_limit = os.environ.get("STRIX_SANDBOX_PIDS_LIMIT", "").strip()
+    if pids_limit:
+        with contextlib.suppress(ValueError):
+            create_kwargs["pids_limit"] = int(pids_limit)
+
+
+def _apply_log_limits(create_kwargs: dict[str, Any]) -> None:
+    """Bound the container's json-file log so a runaway process in the sandbox
+    (e.g. a tool that busy-loops writing to stdout) cannot fill the host disk
+    and take the Docker daemon down with it.
+
+    Unlike the cgroup caps above, this defaults **on** — docker's own default
+    is an unbounded json-file, which is unsafe for an autonomous agent that
+    executes arbitrary commands. ``max-file`` rotation means the on-disk cap is
+    ``max-size * max-file``. Set ``STRIX_SANDBOX_LOG_MAX_SIZE`` to ``0``/``off``
+    to opt back out to docker's default."""
+    max_size = os.environ.get("STRIX_SANDBOX_LOG_MAX_SIZE", "50m").strip()
+    if max_size.lower() in ("0", "off", "none", "unlimited"):
+        return
+    max_file = os.environ.get("STRIX_SANDBOX_LOG_MAX_FILE", "3").strip() or "3"
+    create_kwargs["log_config"] = LogConfig(
+        type=LogConfig.types.JSON,
+        config={"max-size": max_size, "max-file": max_file},
+    )
+
+
+class StrixDockerSandboxSession(DockerSandboxSession):
+    sandbox_network: str = ""
+
+    async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
+        try:
+            self._container.reload()
+        except docker_errors.APIError as e:
+            raise ExposedPortUnavailableError(
+                port=port,
+                exposed_ports=self.state.exposed_ports,
+                reason="backend_unavailable",
+                context={
+                    "backend": "docker",
+                    "detail": "container_reload_failed",
+                    "network": self.sandbox_network,
+                },
+                cause=e,
+            ) from e
+
+        attrs = getattr(self._container, "attrs", {}) or {}
+        networks = attrs.get("NetworkSettings", {}).get("Networks", {})
+        endpoint = networks.get(self.sandbox_network) or {}
+        ip = endpoint.get("IPAddress") or endpoint.get("GlobalIPv6Address")
+        if not isinstance(ip, str) or not ip:
+            raise ExposedPortUnavailableError(
+                port=port,
+                exposed_ports=self.state.exposed_ports,
+                reason="backend_unavailable",
+                context={
+                    "backend": "docker",
+                    "detail": "container_not_on_network",
+                    "network": self.sandbox_network,
+                },
+            )
+        host = f"[{ip}]" if ":" in ip else ip
+        return ExposedPortEndpoint(host=host, port=port, tls=False)
+
+
 class StrixDockerSandboxClient(DockerSandboxClient):
     # Host directories to bind-mount into the container, set by the docker
     # backend before ``create()``. Each item is ``{source, target, read_only}``.
@@ -117,6 +219,10 @@ class StrixDockerSandboxClient(DockerSandboxClient):
         extra_hosts = create_kwargs.setdefault("extra_hosts", {})
         extra_hosts["host.docker.internal"] = "host-gateway"
 
+        _apply_sandbox_network(create_kwargs)
+        _apply_resource_limits(create_kwargs)
+        _apply_log_limits(create_kwargs)
+
         # Strix injection: host bind mounts (e.g. large repos passed via --mount)
         # that bypass the SDK's file-by-file LocalDir copy.
         bind_mounts = getattr(self, "strix_bind_mounts", ())
@@ -146,6 +252,15 @@ class StrixDockerSandboxClient(DockerSandboxClient):
         )
         return container
 
+    async def create(self, **kwargs: Any) -> SandboxSession:
+        session = await super().create(**kwargs)
+        network = _sandbox_network()
+        inner = session._inner
+        if network and isinstance(inner, DockerSandboxSession):
+            inner.__class__ = StrixDockerSandboxSession
+            cast("StrixDockerSandboxSession", inner).sandbox_network = network
+        return session
+
     async def delete(self, session: SandboxSession) -> SandboxSession:
         container_id = getattr(getattr(session._inner, "state", None), "container_id", None)
         if container_id:

strix/runtime/session_manager.py

@@ -5,15 +5,20 @@ from __future__ import annotations
 import logging
 import shutil
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from agents.sandbox.entries import BaseEntry, LocalDir
 from agents.sandbox.manifest import Environment, Manifest
 
 from strix.config import load_settings
 from strix.runtime.backends import get_backend
-from strix.runtime.caido_bootstrap import bootstrap_caido
+from strix.runtime.caido_bootstrap import bootstrap_caido, reconnect_caido
 from strix.runtime.local_dir_staging import stage_symlink_safe_dir
+from strix.tools.proxy.caido_api import SharedCaidoClient
+
+
+if TYPE_CHECKING:
+    from caido_sdk_client import Client as CaidoClient
 
 
 logger = logging.getLogger(__name__)
@@ -131,16 +136,24 @@ async def create_or_reuse(
     host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
     logger.debug("Caido host endpoint resolved: %s", host_caido_url)
 
-    caido_client = await bootstrap_caido(
+    caido_client, caido_project_id = await bootstrap_caido(
         session,
         host_url=host_caido_url,
         container_url=container_caido_url,
     )
 
+    async def _reconnect_caido() -> CaidoClient:
+        return await reconnect_caido(
+            session,
+            host_url=host_caido_url,
+            container_url=container_caido_url,
+            project_id=caido_project_id,
+        )
+
     bundle = {
         "client": client,
         "session": session,
-        "caido_client": caido_client,
+        "caido_client": SharedCaidoClient(caido_client, _reconnect_caido),
     }
     _SESSION_CACHE[scan_id] = bundle
     logger.info("Sandbox session for scan %s ready and cached", scan_id)
@@ -167,11 +180,19 @@ async def cleanup(scan_id: str) -> None:
         except Exception:  # noqa: BLE001
             logger.debug("cleanup(%s): caido_client.aclose() raised", scan_id, exc_info=True)
 
+    client = bundle["client"]
     try:
-        await bundle["client"].delete(bundle["session"])
+        await client.delete(bundle["session"])
         logger.info("Cleaned up sandbox session for scan %s", scan_id)
     except Exception:
         logger.exception(
             "cleanup(%s): client.delete raised; container may need manual reaping",
             scan_id,
         )
+
+    docker_client = getattr(client, "docker_client", None)
+    if docker_client is not None:
+        try:
+            docker_client.close()
+        except Exception:  # noqa: BLE001
+            logger.debug("cleanup(%s): docker_client.close() raised", scan_id, exc_info=True)

strix/skills/tooling/agent_browser.md

@@ -365,6 +365,23 @@ agent-browser dialog accept "text"    # accept with prompt input
 agent-browser dialog dismiss          # cancel
 ```
 
+## Readiness & recovery
+
+The first `agent-browser open` in a session launches the headless-Chrome
+daemon; later commands reuse it. Distinguish the two failure modes and react
+differently — do **not** blindly re-run the same failing command in a loop:
+
+- **Daemon / connection failure** (`Failed to connect`, `connection refused`,
+  socket missing, `browser not running`): the daemon isn't up or has died. Run
+  `agent-browser doctor` (add `--fix` if it reports repairable problems), then
+  re-open the page. Retrying the original command unchanged will keep failing.
+- **Malformed command** (`Unknown command`, `Ref not found`, bad flag): fix the
+  command itself — re-snapshot for fresh refs, or correct the syntax.
+
+Invoke `agent-browser` directly through `exec_command`; there is no need to wrap
+it in an extra `sh -c "..."` / `bash -lc "..."` layer, which only adds shell
+quoting and startup-file pitfalls.
+
 ## Diagnosing install issues
 
 If a command fails unexpectedly (`Unknown command`, `Failed to connect`,

strix/skills/tooling/katana.md

@@ -24,7 +24,15 @@ High-signal flags:
 - `-p, -parallelism <n>` concurrent input targets
 - `-rl, -rate-limit <n>` request rate limit
 - `-timeout <seconds>` request timeout
+- `-ct, -crawl-duration <s|m|h|d>` maximum time to crawl the target
 - `-retry <n>` retry count
+- `-mdp, -max-domain-pages <n>` cap pages crawled per domain (default: unlimited)
+- `-fsu, -filter-similar` collapse similar URLs (e.g. /users/123 and /users/456)
+- `-fs, -field-scope <dn|rdn|fqdn|regex>` crawl scope (default `rdn` = root domain + ALL subdomains)
+- `-f, -field <url|path|...>` emit only one field (e.g. `-f url` for a plain URL list)
+- `-or, -omit-raw` omit raw request/response from JSONL output
+- `-ob, -omit-body` omit response body from JSONL output
+- `-mrs, -max-response-size <bytes>` cap per-response bytes read (default 4194304)
 - `-ef, -extension-filter <list>` extension exclusions
 - `-tlsi, -tls-impersonate` experimental JA3/TLS impersonation
 - `-hl, -headless` enable hybrid headless crawling
@@ -37,13 +45,13 @@ High-signal flags:
 - `-silent`, `-j, -jsonl`, `-o <file>` output controls
 
 Agent-safe baseline for automation:
-`mkdir -p crawl && katana -u https://target.tld -d 3 -jc -kf robotstxt -c 10 -p 10 -rl 50 -timeout 10 -retry 1 -ef png,jpg,jpeg,gif,svg,css,woff,woff2,ttf,eot,map -silent -j -o crawl/katana.jsonl`
+`mkdir -p crawl && katana -u https://target.tld -d 3 -ct 10m -mdp 2000 -fsu -jc -kf robotstxt -c 10 -p 10 -rl 50 -timeout 10 -retry 1 -ef png,jpg,jpeg,gif,svg,css,woff,woff2,ttf,eot,map -silent -j -o crawl/katana.jsonl`
 
 Common patterns:
 - Fast crawl baseline:
   `katana -u https://target.tld -d 3 -jc -silent`
-- Deeper JS-aware crawl:
-  `katana -u https://target.tld -d 5 -jc -jsl -kf all -c 10 -p 10 -rl 50 -o katana_urls.txt`
+- Deeper JS-aware crawl (narrowed target; keep it time-bounded):
+  `katana -u https://target.tld -d 5 -ct 15m -jc -jsl -kf all -c 10 -p 10 -rl 50 -o katana_urls.txt`
 - Multi-target run with JSONL output:
   `katana -list urls.txt -d 3 -jc -silent -j -o katana.jsonl`
 - Headless crawl with local Chrome:
@@ -59,6 +67,13 @@ Critical correctness rules:
 - For `-kf`, keep depth at least `-d 3` so known files are fully covered.
 - If writing to a file, ensure parent directory exists before `-o`.
 
+Keeping output small (katana has NO default page cap, so plan for volume):
+- Bound scope and volume: `-fs fqdn` (or `-cs`/`-cos` regex) so the crawl doesn't wander across every subdomain, `-mdp <n>` to cap pages per domain, `-fsu` to collapse near-identical URLs, and `-ct`/`-d` to bound time and depth.
+- Shrink each record: default JSONL is verbose. If you only need endpoints, emit a plain URL list with `-f url` instead of `-j`. If you need JSONL, drop the heavy parts with `-or` (omit raw) and `-ob` (omit body), and lower `-mrs` to cap per-response bytes.
+- Reserve `-jsl` / `-kf all` / higher `-d` for a specific narrowed target — they multiply output fast on large sites.
+- Reduce, then delete: once the crawl finishes, extract just what you need (e.g. `katana ... -f url -o urls.txt` or `sort -u` a URL list, or a short note of interesting paths) and remove the raw crawl file/dir. Don't keep large raw crawls around after you've distilled them.
+- Sanity-check size (`du -sh <out>`); if it's outsized for the scope, tighten `-fs`/`-mdp`/`-fsu`/`-d`/`-ct` and re-run rather than keeping it.
+
 Usage rules:
 - Keep `-d`, `-c`, `-p`, and `-rl` explicit for reproducible runs.
 - Use `-ef` early to reduce static-file noise before fuzzing.

strix/skills/tooling/python.md

@@ -7,9 +7,9 @@ description: Run Python through exec_command in the SDK sandbox. Use the image-b
 
 Use `exec_command` for Python. There is no separate Strix Python executor.
 
-Prefer writing reusable scripts to `/workspace/scratch/<name>.py` and
-running them with `python3 /workspace/scratch/<name>.py`. For short
-one-off transformations, `python3 -c` or a small here-document is fine.
+Prefer writing reusable scripts to a `.py` file and running them with
+`python3 <name>.py`. For short one-off transformations, `python3 -c` or a
+small here-document is fine.
 
 The `shell` parameter on `exec_command` is for swapping POSIX shells
 (`bash`/`zsh`/`sh`), not for picking interpreters. Put the interpreter
@@ -84,17 +84,26 @@ automatically, so it shows up in `list_requests` and you can use
 For iterative exploit work, put code in a file:
 
 ```text
-1. Create or edit `/workspace/scratch/exploit.py` with `apply_patch`.
-2. Run it with `exec_command`: `python3 /workspace/scratch/exploit.py`.
+1. Create or edit a task-unique script (e.g. `poc_<task-id>.py`, so it can't
+   clobber a project file or another agent's script) with `apply_patch`.
+2. Run it with `exec_command`: `python3 poc_<task-id>.py`.
 3. Edit and rerun until the proof-of-concept is reliable.
 ```
 
 ## Installing extra packages
 
-The sandbox's Python lives in `/app/.venv`. To add a one-off dependency
-for an exploit script, use `uv` (already in the image and much faster
-than pip):
+The sandbox's Python lives in `/app/.venv`, and it is the active virtualenv
+(`python3` / `pip` already resolve to it). The following common libraries are
+**pre-installed** — import them directly, no install step needed:
+`requests`, `httpx`, `beautifulsoup4` (`bs4`), `lxml`, `pyjwt` (`jwt`),
+`cryptography`.
+
+To add a one-off dependency for an exploit script, use `uv` (already in the
+image and much faster than pip):
 
 ```bash
 uv pip install --python /app/.venv/bin/python <package>
 ```
+
+Plain `pip install <package>` also works because the venv is active. Install
+before you import, so scripts don't fail with `ModuleNotFoundError`.

strix/tools/agents_graph/tools.py

@@ -229,7 +229,8 @@ async def wait_for_message(  # noqa: PLR0911
     Use when you have nothing useful to do until a child/peer responds
     — typically after spawning subagents and you want to wait for
     their completion reports. The agent automatically resumes when any
-    message arrives.
+    message arrives, so pick a ``timeout_seconds`` proportional to the
+    work you're awaiting.
 
     **Critical caveats:**
 
@@ -246,9 +247,19 @@ async def wait_for_message(  # noqa: PLR0911
         reason: One-line note shown in graph snapshots while you're
             waiting (helps a human or sibling agent debug who's stuck
             on what).
-        timeout_seconds: Hard cap (default 600s). On timeout the tool
-            returns and you decide whether to keep working or wait
-            again.
+        timeout_seconds: Max seconds to wait (default 600). This is only
+            a cap — the tool returns the INSTANT a message arrives, so a
+            larger value never makes you wait longer when the reply does
+            come. Right-size it to what you're waiting on: a short wait
+            (e.g. 10-60s) for a quick ack or a small/fast subtask, and a
+            longer one (e.g. ~100-200s) only for genuinely long-running
+            work (deep recon, exploitation, a full sub-scan). The cap only
+            bites when the expected message never arrives — so an oversized
+            timeout on a trivial wait just strands you idle until it
+            elapses. On timeout the tool returns and you decide whether to
+            keep working or wait again. (Applies to autonomous multi-agent
+            runs; in interactive/chat sessions the agent instead parks until
+            a message arrives and this cap is not enforced.)
     """
     inner = _ctx(ctx)
     coordinator = coordinator_from_context(inner)

strix/tools/proxy/caido_api.py

@@ -3,7 +3,9 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
+import logging
 import os
 import time
 import urllib.request
@@ -21,9 +23,14 @@ from caido_sdk_client.types import (
 
 
 if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
     from caido_sdk_client import Client as CaidoClient
 
 
+logger = logging.getLogger(__name__)
+
+
 RequestPart = Literal["request", "response"]
 SortBy = Literal[
     "timestamp",
@@ -42,6 +49,19 @@ _SITEMAP_PAGE_SIZE = 30
 
 _DEFAULT_CAIDO_URL = "http://127.0.0.1:48080"
 _CLIENT_CACHE: dict[str, Client] = {}
+_CLIENT_LOCK = asyncio.Lock()
+
+# Substrings that mean the shared client's transport has died or is being used
+# concurrently — recoverable by rebuilding the client and retrying once.
+_CONNECTION_ERROR_MARKERS = (
+    "transport is already connected",
+    "connector is closed",
+    "server disconnected",
+    "session is closed",
+    "cannot write to closing transport",
+    "connection reset",
+    "connection closed",
+)
 _REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = {
     "timestamp": ("req", "created_at"),
     "host": ("req", "host"),
@@ -81,19 +101,116 @@ def _login_as_guest() -> str:
     return str(payload["data"]["loginAsGuest"]["token"]["accessToken"])
 
 
-async def get_client() -> Client:
-    if client := _CLIENT_CACHE.get("default"):
-        return client
-
+async def _new_client() -> Client:
     token = await asyncio.to_thread(_login_as_guest)
     client = Client(caido_url(), auth=TokenAuthOptions(token=token))
     await client.connect()
-    _CLIENT_CACHE["default"] = client
     return client
 
 
+async def _safe_aclose(client: Client | None) -> None:
+    """Close a (possibly dead) client without letting teardown errors escape."""
+    if client is None:
+        return
+    with contextlib.suppress(Exception):
+        await client.aclose()
+
+
+def _is_connection_error(exc: BaseException) -> bool:
+    message = str(exc).lower()
+    if any(marker in message for marker in _CONNECTION_ERROR_MARKERS):
+        return True
+    cause = exc.__cause__ or exc.__context__
+    return cause is not None and cause is not exc and _is_connection_error(cause)
+
+
+async def get_client() -> Client:
+    """Return the shared Caido client, creating it under a lock if needed.
+
+    The lock prevents two concurrent callers from each building a client and
+    racing ``connect()`` on the same transport ("Transport is already
+    connected").
+    """
+    async with _CLIENT_LOCK:
+        client = _CLIENT_CACHE.get("default")
+        if client is None:
+            client = await _new_client()
+            _CLIENT_CACHE["default"] = client
+        return client
+
+
+async def call_with_client[T](
+    fn: Callable[[Client], Awaitable[T]], *, idempotent: bool = True
+) -> T:
+    """Run ``fn`` against the shared client, serialized and reconnect-safe.
+
+    The Caido GraphQL transport is not safe for concurrent use: two in-flight
+    requests race and raise "Transport is already connected". All proxy calls
+    are therefore serialized through ``_CLIENT_LOCK``. If the cached client's
+    transport has since died ("Connector is closed" / "Server disconnected"),
+    the stale client is closed and rebuilt so subsequent calls stop failing
+    against a dead client.
+
+    ``fn`` is only re-run automatically when ``idempotent`` is true. For
+    mutations (replay, scope create/update/delete) a connection error may
+    arrive *after* Caido applied the change, so we heal the client for future
+    calls but re-raise instead of risking a double-apply.
+    """
+    async with _CLIENT_LOCK:
+        client = _CLIENT_CACHE.get("default")
+        if client is None:
+            client = await _new_client()
+            _CLIENT_CACHE["default"] = client
+        try:
+            return await fn(client)
+        except Exception as exc:
+            if not _is_connection_error(exc):
+                raise
+            new_client = await _new_client()
+            _CLIENT_CACHE["default"] = new_client
+            await _safe_aclose(client)
+            if not idempotent:
+                raise
+            return await fn(new_client)
+
+
+class SharedCaidoClient:
+    """Serialized, reconnect-safe wrapper around one host-side Caido client.
+
+    Every agent in a scan shares a single instance (propagated through the
+    shallow-copied run context). ``call`` serializes access — the SDK transport
+    is not concurrency-safe — and, when the transport dies, rebuilds the client
+    via ``reconnect`` (which preserves the Caido project) and closes the dead
+    one, so a transient Caido restart no longer disables proxy tools for the
+    rest of the scan.
+    """
+
+    def __init__(self, client: Client, reconnect: Callable[[], Awaitable[Client]]) -> None:
+        self._client = client
+        self._reconnect = reconnect
+        self._lock = asyncio.Lock()
+
+    async def call[T](self, fn: Callable[[Client], Awaitable[T]], *, idempotent: bool = True) -> T:
+        async with self._lock:
+            try:
+                return await fn(self._client)
+            except Exception as exc:
+                if not _is_connection_error(exc):
+                    raise
+                dead, self._client = self._client, await self._reconnect()
+                await _safe_aclose(dead)
+                if not idempotent:
+                    raise
+                return await fn(self._client)
+
+    async def aclose(self) -> None:
+        async with self._lock:
+            await _safe_aclose(self._client)
+
+
 async def close_client() -> None:
-    client = _CLIENT_CACHE.pop("default", None)
+    async with _CLIENT_LOCK:
+        client = _CLIENT_CACHE.pop("default", None)
     if client is None:
         return
     await client.aclose()
@@ -385,19 +502,23 @@ async def list_requests(
     sort_order: SortOrder = "desc",
     scope_id: str | None = None,
 ) -> Any:
-    return await list_requests_with_client(
-        await get_client(),
-        httpql_filter=httpql_filter,
-        first=first,
-        after=after,
-        sort_by=sort_by,
-        sort_order=sort_order,
-        scope_id=scope_id,
+    return await call_with_client(
+        lambda client: list_requests_with_client(
+            client,
+            httpql_filter=httpql_filter,
+            first=first,
+            after=after,
+            sort_by=sort_by,
+            sort_order=sort_order,
+            scope_id=scope_id,
+        )
     )
 
 
 async def view_request(request_id: str, *, part: RequestPart = "request") -> Any:
-    return await get_request_with_client(await get_client(), request_id, part=part)
+    return await call_with_client(
+        lambda client: get_request_with_client(client, request_id, part=part)
+    )
 
 
 async def repeat_request(
@@ -406,22 +527,28 @@ async def repeat_request(
     modifications: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     mods = modifications or {}
-    result = await get_request_with_client(await get_client(), request_id, part="request")
-    if result is None or result.request.raw is None:
-        raise ValueError(f"Request {request_id} not found")
 
-    original = result.request
-    raw_str = result.request.raw.decode("utf-8", errors="replace")
-    components = parse_raw_request(raw_str)
-    full_url = full_url_from_components(original, components, mods)
-    modified = apply_modifications(components, mods, full_url)
-    connection, raw = build_raw_request(
-        method=modified["method"],
-        url=modified["url"],
-        headers=modified["headers"],
-        body=modified["body"],
-    )
-    return await replay_send_raw(await get_client(), raw=raw, connection=connection)
+    async def _run(client: CaidoClient) -> dict[str, Any]:
+        result = await get_request_with_client(client, request_id, part="request")
+        if result is None or result.request.raw is None:
+            raise ValueError(f"Request {request_id} not found")
+
+        original = result.request
+        raw_str = result.request.raw.decode("utf-8", errors="replace")
+        components = parse_raw_request(raw_str)
+        full_url = full_url_from_components(original, components, mods)
+        modified = apply_modifications(components, mods, full_url)
+        connection, raw = build_raw_request(
+            method=modified["method"],
+            url=modified["url"],
+            headers=modified["headers"],
+            body=modified["body"],
+        )
+        return await replay_send_raw(client, raw=raw, connection=connection)
+
+    # A replay mutates server state; don't auto-retry if the transport dies
+    # mid-send (the request may already have been sent).
+    return await call_with_client(_run, idempotent=False)
 
 
 async def scope_rules(
@@ -432,7 +559,29 @@ async def scope_rules(
     scope_id: str | None = None,
     scope_name: str | None = None,
 ) -> Any:
-    client = await get_client()
+    async def _run(client: CaidoClient) -> Any:
+        return await _scope_rules_with_client(
+            client,
+            action,
+            allowlist=allowlist,
+            denylist=denylist,
+            scope_id=scope_id,
+            scope_name=scope_name,
+        )
+
+    # get/list are read-only and safe to retry; create/update/delete mutate.
+    return await call_with_client(_run, idempotent=action in {"get", "list"})
+
+
+async def _scope_rules_with_client(
+    client: CaidoClient,
+    action: ScopeAction,
+    *,
+    allowlist: list[str] | None = None,
+    denylist: list[str] | None = None,
+    scope_id: str | None = None,
+    scope_name: str | None = None,
+) -> Any:
     if action == "list":
         result = await scope_list(client)
     elif action == "get":
@@ -651,26 +800,30 @@ async def list_sitemap(
     page: int = 1,
     page_size: int = _SITEMAP_PAGE_SIZE,
 ) -> dict[str, Any]:
-    return await list_sitemap_with_client(
-        await get_client(),
-        scope_id=scope_id,
-        parent_id=parent_id,
-        depth=depth,
-        page=page,
-        page_size=page_size,
+    return await call_with_client(
+        lambda client: list_sitemap_with_client(
+            client,
+            scope_id=scope_id,
+            parent_id=parent_id,
+            depth=depth,
+            page=page,
+            page_size=page_size,
+        )
     )
 
 
 async def view_sitemap_entry(entry_id: str) -> dict[str, Any]:
-    return await view_sitemap_entry_with_client(await get_client(), entry_id)
+    return await call_with_client(lambda client: view_sitemap_entry_with_client(client, entry_id))
 
 
 __all__ = [
     "RequestPart",
     "ScopeAction",
+    "SharedCaidoClient",
     "SitemapDepth",
     "SortBy",
     "SortOrder",
+    "call_with_client",
     "close_client",
     "get_client",
     "list_requests",

strix/tools/proxy/tools.py

@@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, Any, Literal
 from agents import RunContextWrapper, function_tool
 
 from strix.tools.proxy import caido_api
+from strix.tools.proxy.caido_api import SharedCaidoClient
 
 
 logger = logging.getLogger(__name__)
@@ -28,7 +29,7 @@ if TYPE_CHECKING:
         SortOrder,
     )
 else:
-    from strix.tools.proxy.caido_api import (  # noqa: TC001
+    from strix.tools.proxy.caido_api import (
         RequestPart,
         SitemapDepth,
         SortBy,
@@ -39,9 +40,18 @@ else:
 ScopeAction = Literal["get", "list", "create", "update", "delete"]
 
 
-def _ctx_client(ctx: RunContextWrapper) -> Client | None:
+def _ctx_proxy(ctx: RunContextWrapper) -> SharedCaidoClient | None:
+    """Return the scan-wide serialized, reconnect-safe Caido client holder.
+
+    All agents in a scan share one :class:`SharedCaidoClient` whose GraphQL
+    transport is not concurrency-safe (parallel calls raise "Transport is
+    already connected"). ``SharedCaidoClient.call`` serializes access and
+    rebuilds the transport if it dies mid-scan. Returns ``None`` when no holder
+    is present (e.g. standalone tool invocation outside a scan run).
+    """
     inner = ctx.context if isinstance(ctx.context, dict) else {}
-    return inner.get("caido_client")
+    proxy = inner.get("caido_client")
+    return proxy if isinstance(proxy, SharedCaidoClient) else None
 
 
 def _to_tool_json(value: Any) -> Any:
@@ -83,6 +93,39 @@ def _err(name: str, exc: Exception) -> str:
     )
 
 
+_HTTPQL_HINT = (
+    "HTTPQL syntax: quote string values and leave integers unquoted; combine "
+    "terms with AND / OR (there is no NOT). Numeric fields (resp.code, req.port, "
+    "id, roundtrip) use eq/ne/gt/gte/lt/lte; text/byte fields (req.host, req.path, "
+    "req.method, req.raw, resp.raw) use cont/ncont/eq/ne/like/nlike/regex/nregex. "
+    "Example: 'resp.code.gte:200 AND resp.code.lt:300 AND req.host.cont:\"api\"'."
+)
+
+
+def _is_httpql_error(exc: Exception) -> bool:
+    message = str(exc).lower()
+    return "httpql" in message or ("filter" in message and "pars" in message)
+
+
+def _httpql_error(exc: Exception, httpql_filter: str | None) -> str:
+    """Return an actionable error for a rejected HTTPQL filter.
+
+    Preserves Caido's exact parser message and echoes the offending query so
+    the agent can self-correct instead of retrying the same broken filter.
+    """
+    logger.info("list_requests rejected HTTPQL filter %r: %s", httpql_filter, exc)
+    return json.dumps(
+        {
+            "success": False,
+            "error": f"Invalid HTTPQL filter: {exc}",
+            "httpql_filter": httpql_filter,
+            "hint": _HTTPQL_HINT,
+        },
+        ensure_ascii=False,
+        default=str,
+    )
+
+
 @function_tool(timeout=120)
 async def list_requests(
     ctx: RunContextWrapper,
@@ -141,19 +184,21 @@ async def list_requests(
         sort_order: ``asc`` or ``desc``.
         scope_id: Restrict to a Caido scope (managed via ``scope_rules``).
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
 
     try:
-        connection = await caido_api.list_requests_with_client(
-            client,
-            httpql_filter=httpql_filter,
-            first=first,
-            after=after,
-            sort_by=sort_by,
-            sort_order=sort_order,
-            scope_id=scope_id,
+        connection = await proxy.call(
+            lambda client: caido_api.list_requests_with_client(
+                client,
+                httpql_filter=httpql_filter,
+                first=first,
+                after=after,
+                sort_by=sort_by,
+                sort_order=sort_order,
+                scope_id=scope_id,
+            )
         )
 
         entries = []
@@ -207,6 +252,8 @@ async def list_requests(
             default=str,
         )
     except Exception as exc:  # noqa: BLE001
+        if httpql_filter and _is_httpql_error(exc):
+            return _httpql_error(exc, httpql_filter)
         return _err("list_requests", exc)
 
 
@@ -244,12 +291,14 @@ async def view_request(
         page: 1-indexed page number (only when no ``search_pattern``).
         page_size: Lines per page.
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
 
     try:
-        result = await caido_api.get_request_with_client(client, request_id, part=part)
+        result = await proxy.call(
+            lambda client: caido_api.get_request_with_client(client, request_id, part=part)
+        )
         if result is None:
             return json.dumps(
                 {"success": False, "error": f"Request {request_id} not found"},
@@ -359,20 +408,15 @@ async def repeat_request(
             - ``body`` — replace the body string entirely.
             - ``cookies`` — dict of cookies to add/update.
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
     mods = modifications or {}
 
-    try:
+    async def _do(client: Client) -> dict[str, Any] | None:
         result = await caido_api.get_request_with_client(client, request_id, part="request")
         if result is None or result.request.raw is None:
-            return json.dumps(
-                {"success": False, "error": f"Request {request_id} not found"},
-                ensure_ascii=False,
-                default=str,
-            )
-
+            return None
         original = result.request
         raw_str = result.request.raw.decode("utf-8", errors="replace")
         components = caido_api.parse_raw_request(raw_str)
@@ -384,7 +428,18 @@ async def repeat_request(
             headers=modified["headers"],
             body=modified["body"],
         )
-        replay = await caido_api.replay_send_raw(client, raw=raw, connection=connection)
+        return await caido_api.replay_send_raw(client, raw=raw, connection=connection)
+
+    try:
+        # A replay mutates target state, so don't auto-retry on a mid-send
+        # transport failure (the request may already have been sent).
+        replay = await proxy.call(_do, idempotent=False)
+        if replay is None:
+            return json.dumps(
+                {"success": False, "error": f"Request {request_id} not found"},
+                ensure_ascii=False,
+                default=str,
+            )
         return _format_replay_tool_result(replay)
     except Exception as exc:  # noqa: BLE001
         return _err("repeat_request", exc)
@@ -437,16 +492,18 @@ async def list_sitemap(
             (recursive subtree). Only meaningful with ``parent_id``.
         page: 1-indexed page (30 entries per page).
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
     try:
-        payload = await caido_api.list_sitemap_with_client(
-            client,
-            scope_id=scope_id,
-            parent_id=parent_id,
-            depth=depth,
-            page=page,
+        payload = await proxy.call(
+            lambda client: caido_api.list_sitemap_with_client(
+                client,
+                scope_id=scope_id,
+                parent_id=parent_id,
+                depth=depth,
+                page=page,
+            )
         )
         return json.dumps(payload, ensure_ascii=False, default=str)
     except Exception as exc:  # noqa: BLE001
@@ -468,11 +525,13 @@ async def view_sitemap_entry(
     Args:
         entry_id: ID from ``list_sitemap`` (or any nested entry).
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
     try:
-        payload = await caido_api.view_sitemap_entry_with_client(client, entry_id)
+        payload = await proxy.call(
+            lambda client: caido_api.view_sitemap_entry_with_client(client, entry_id)
+        )
         return json.dumps(payload, ensure_ascii=False, default=str)
     except Exception as exc:  # noqa: BLE001
         return _err("view_sitemap_entry", exc)
@@ -524,13 +583,13 @@ async def scope_rules(
         scope_id: Required for ``get`` / ``update`` / ``delete``.
         scope_name: Required for ``create`` / ``update``.
     """
-    client = _ctx_client(ctx)
-    if client is None:
+    proxy = _ctx_proxy(ctx)
+    if proxy is None:
         return _no_client()
 
     try:
         if action == "list":
-            scopes = await caido_api.scope_list(client)
+            scopes = await proxy.call(caido_api.scope_list)
             return json.dumps(
                 {"success": True, "scopes": [_to_tool_json(s) for s in scopes]},
                 ensure_ascii=False,
@@ -543,9 +602,11 @@ async def scope_rules(
                     ensure_ascii=False,
                     default=str,
                 )
-            scope = await caido_api.scope_get(client, scope_id)
+            scope = await proxy.call(lambda client: caido_api.scope_get(client, scope_id))
             return json.dumps(
-                {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
+                {"success": True, "scope": _to_tool_json(scope)},
+                ensure_ascii=False,
+                default=str,
             )
         if action == "create":
             if not scope_name:
@@ -554,11 +615,16 @@ async def scope_rules(
                     ensure_ascii=False,
                     default=str,
                 )
-            scope = await caido_api.scope_create(
-                client, name=scope_name, allowlist=allowlist, denylist=denylist
+            scope = await proxy.call(
+                lambda client: caido_api.scope_create(
+                    client, name=scope_name, allowlist=allowlist, denylist=denylist
+                ),
+                idempotent=False,
             )
             return json.dumps(
-                {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
+                {"success": True, "scope": _to_tool_json(scope)},
+                ensure_ascii=False,
+                default=str,
             )
         if action == "update":
             if not scope_id or not scope_name:
@@ -570,11 +636,16 @@ async def scope_rules(
                     ensure_ascii=False,
                     default=str,
                 )
-            scope = await caido_api.scope_update(
-                client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist
+            scope = await proxy.call(
+                lambda client: caido_api.scope_update(
+                    client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist
+                ),
+                idempotent=False,
             )
             return json.dumps(
-                {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, default=str
+                {"success": True, "scope": _to_tool_json(scope)},
+                ensure_ascii=False,
+                default=str,
             )
         if not scope_id:
             return json.dumps(
@@ -582,7 +653,7 @@ async def scope_rules(
                 ensure_ascii=False,
                 default=str,
             )
-        await caido_api.scope_delete(client, scope_id)
+        await proxy.call(lambda client: caido_api.scope_delete(client, scope_id), idempotent=False)
         return json.dumps(
             {
                 "success": True,

strix/tools/shell/README.md

@@ -5,6 +5,23 @@ invocation the agent makes (nmap, ffuf, agent-browser, python3, …) goes
 through `exec_command`. `write_stdin` streams input to a still-running
 process started by an earlier `exec_command` (for interactive prompts).
 
+## `write_stdin` requires a TTY-backed process
+
+`exec_command` runs each command in a fresh **non-interactive** shell (plain
+pipes, no TTY) by default. `write_stdin` only works against a process that is
+still running **and** was started with a PTY. The canonical sequence is:
+
+```text
+exec_command(cmd="python3", tty=true)      # start a PTY-backed process
+write_stdin(session_id=<id>, chars="print(1)\n")
+```
+
+Calling `write_stdin` on a command started with the default `tty=false`, or on
+a process that has already exited, fails with
+`stdin is not available for this process. Start the command with 'tty=true' in
+'exec_command' before using 'write_stdin'.` Use `tty=true` for REPLs,
+`ssh`/`nc`/`ftp`, `msfconsole`, or to deliver a Ctrl-C to a long-running job.
+
 - **Implementation:** `agents.sandbox.capabilities.tools.shell_tool.ShellTool`
   (in the upstream `agents` SDK)
 - **Wired in:** `strix/agents/factory.py` — added per-run via the SDK

tests/test_proxy_client.py

@@ -0,0 +1,321 @@
+"""Tests for the shared Caido client lifecycle and proxy error handling.
+
+Covers the concurrency/reconnect guarantees of ``caido_api.call_with_client``
+(the sandbox-imported path) and ``caido_api.SharedCaidoClient`` (the host-side
+holder), plus the actionable HTTPQL errors in ``proxy.tools``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import TYPE_CHECKING, Any, cast
+
+import pytest
+
+from strix.tools.proxy import caido_api, tools
+from strix.tools.proxy.caido_api import SharedCaidoClient
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+class _FakeClient:
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.closed = False
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+@pytest.fixture(autouse=True)
+def _clear_cache() -> Iterator[None]:
+    caido_api._CLIENT_CACHE.clear()
+    yield
+    caido_api._CLIENT_CACHE.clear()
+
+
+async def test_call_with_client_reuses_cached_client(monkeypatch: pytest.MonkeyPatch) -> None:
+    cached = _FakeClient("cached")
+    caido_api._CLIENT_CACHE["default"] = cast("Any", cached)
+
+    async def _new() -> Any:
+        raise AssertionError("_new_client must not run when a client is cached")
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    seen: dict[str, Any] = {}
+
+    async def fn(client: Any) -> str:
+        seen["client"] = client
+        return "ok"
+
+    assert await caido_api.call_with_client(fn) == "ok"
+    assert seen["client"] is cached
+
+
+async def test_call_with_client_creates_and_caches_when_empty(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    created = _FakeClient("fresh")
+
+    async def _new() -> Any:
+        return created
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    seen: dict[str, Any] = {}
+
+    async def fn(client: Any) -> str:
+        seen["client"] = client
+        return "ok"
+
+    assert await caido_api.call_with_client(fn) == "ok"
+    assert seen["client"] is created
+    assert caido_api._CLIENT_CACHE["default"] is created
+
+
+async def test_failed_init_does_not_poison_cache(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def _new() -> Any:
+        raise ConnectionRefusedError("caido not up yet")
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    async def fn(_client: Any) -> str:
+        return "unreachable"
+
+    with pytest.raises(ConnectionRefusedError):
+        await caido_api.call_with_client(fn)
+    assert "default" not in caido_api._CLIENT_CACHE
+
+
+async def test_call_with_client_reconnects_and_closes_dead_transport(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    dead = _FakeClient("dead")
+    fresh = _FakeClient("fresh")
+    caido_api._CLIENT_CACHE["default"] = cast("Any", dead)
+
+    new_calls = {"n": 0}
+
+    async def _new() -> Any:
+        new_calls["n"] += 1
+        return fresh
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    attempts: list[Any] = []
+
+    async def fn(client: Any) -> str:
+        attempts.append(client)
+        if len(attempts) == 1:
+            raise RuntimeError("Transport is already connected")
+        return "ok"
+
+    assert await caido_api.call_with_client(fn) == "ok"
+    assert attempts == [dead, fresh]
+    assert new_calls["n"] == 1
+    assert caido_api._CLIENT_CACHE["default"] is fresh
+    assert dead.closed is True  # stale transport is not leaked
+
+
+async def test_call_with_client_non_idempotent_rebuilds_but_reraises(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    dead = _FakeClient("dead")
+    fresh = _FakeClient("fresh")
+    caido_api._CLIENT_CACHE["default"] = cast("Any", dead)
+
+    async def _new() -> Any:
+        return fresh
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    calls = {"n": 0}
+
+    async def fn(_client: Any) -> str:
+        calls["n"] += 1
+        raise RuntimeError("Server disconnected")
+
+    # A mutation must not be auto-retried (it may already have applied), but the
+    # dead client is still healed so later calls succeed.
+    with pytest.raises(RuntimeError, match="Server disconnected"):
+        await caido_api.call_with_client(fn, idempotent=False)
+    assert calls["n"] == 1
+    assert caido_api._CLIENT_CACHE["default"] is fresh
+    assert dead.closed is True
+
+
+async def test_call_with_client_does_not_retry_application_errors(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    cached = _FakeClient("cached")
+    caido_api._CLIENT_CACHE["default"] = cast("Any", cached)
+
+    async def _new() -> Any:
+        raise AssertionError("deterministic errors must not trigger a reconnect")
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    calls = {"n": 0}
+
+    async def fn(_client: Any) -> str:
+        calls["n"] += 1
+        raise ValueError("Invalid HTTPQL filter")
+
+    with pytest.raises(ValueError, match="Invalid HTTPQL"):
+        await caido_api.call_with_client(fn)
+    assert calls["n"] == 1
+    assert caido_api._CLIENT_CACHE["default"] is cached
+
+
+async def test_call_with_client_serializes_concurrent_calls(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    caido_api._CLIENT_CACHE["default"] = cast("Any", _FakeClient("shared"))
+
+    async def _new() -> Any:
+        raise AssertionError("no reconnect expected")
+
+    monkeypatch.setattr(caido_api, "_new_client", _new)
+
+    state = {"active": 0, "max": 0}
+
+    async def fn(_client: Any) -> str:
+        state["active"] += 1
+        state["max"] = max(state["max"], state["active"])
+        await asyncio.sleep(0.01)
+        state["active"] -= 1
+        return "ok"
+
+    await asyncio.gather(*(caido_api.call_with_client(fn) for _ in range(6)))
+    assert state["max"] == 1
+
+
+async def test_shared_client_reconnects_and_closes_dead_transport() -> None:
+    dead = _FakeClient("dead")
+    fresh = _FakeClient("fresh")
+
+    async def _reconnect() -> Any:
+        return fresh
+
+    holder = SharedCaidoClient(cast("Any", dead), _reconnect)
+
+    attempts: list[Any] = []
+
+    async def fn(client: Any) -> str:
+        attempts.append(client)
+        if len(attempts) == 1:
+            raise RuntimeError("Connector is closed")
+        return "ok"
+
+    assert await holder.call(fn) == "ok"
+    assert attempts == [dead, fresh]
+    assert dead.closed is True
+
+
+async def test_shared_client_non_idempotent_rebuilds_but_reraises() -> None:
+    dead = _FakeClient("dead")
+    fresh = _FakeClient("fresh")
+
+    async def _reconnect() -> Any:
+        return fresh
+
+    holder = SharedCaidoClient(cast("Any", dead), _reconnect)
+
+    calls = {"n": 0}
+
+    async def fn(_client: Any) -> str:
+        calls["n"] += 1
+        raise RuntimeError("Server disconnected")
+
+    with pytest.raises(RuntimeError, match="Server disconnected"):
+        await holder.call(fn, idempotent=False)
+    assert calls["n"] == 1
+    assert dead.closed is True
+    # The healthy client remains for the next call.
+    assert await holder.call(lambda _c: _ok()) == "ok"
+
+
+async def _ok() -> str:
+    return "ok"
+
+
+async def test_shared_client_serializes_concurrent_calls() -> None:
+    async def _reconnect() -> Any:
+        raise AssertionError("no reconnect expected")
+
+    holder = SharedCaidoClient(cast("Any", _FakeClient("shared")), _reconnect)
+
+    state = {"active": 0, "max": 0}
+
+    async def fn(_client: Any) -> str:
+        state["active"] += 1
+        state["max"] = max(state["max"], state["active"])
+        await asyncio.sleep(0.01)
+        state["active"] -= 1
+        return "ok"
+
+    await asyncio.gather(*(holder.call(fn) for _ in range(6)))
+    assert state["max"] == 1
+
+
+async def test_shared_client_passes_through_application_errors() -> None:
+    async def _reconnect() -> Any:
+        raise AssertionError("deterministic errors must not trigger a reconnect")
+
+    holder = SharedCaidoClient(cast("Any", _FakeClient("c")), _reconnect)
+
+    async def fn(_client: Any) -> str:
+        raise ValueError("Invalid HTTPQL filter")
+
+    with pytest.raises(ValueError, match="Invalid HTTPQL"):
+        await holder.call(fn)
+
+
+def test_is_connection_error_matches_markers_and_causes() -> None:
+    assert caido_api._is_connection_error(RuntimeError("Transport is already connected"))
+    assert caido_api._is_connection_error(RuntimeError("Connector is closed"))
+    assert caido_api._is_connection_error(RuntimeError("Server disconnected"))
+    assert not caido_api._is_connection_error(ValueError("Invalid HTTPQL filter"))
+
+    nested = RuntimeError("wrapper")
+    nested.__cause__ = RuntimeError("connection reset by peer")
+    assert caido_api._is_connection_error(nested)
+
+
+class _Ctx:
+    def __init__(self, context: Any) -> None:
+        self.context = context
+
+
+def test_ctx_proxy_returns_holder_when_present() -> None:
+    async def _reconnect() -> Any:
+        raise AssertionError("unused")
+
+    holder = SharedCaidoClient(cast("Any", _FakeClient("c")), _reconnect)
+    got = tools._ctx_proxy(cast("Any", _Ctx({"caido_client": holder})))
+    assert got is holder
+
+
+def test_ctx_proxy_returns_none_without_holder() -> None:
+    assert tools._ctx_proxy(cast("Any", _Ctx({}))) is None
+    assert tools._ctx_proxy(cast("Any", _Ctx(None))) is None
+    assert tools._ctx_proxy(cast("Any", _Ctx({"caido_client": object()}))) is None
+
+
+def test_is_httpql_error_detection() -> None:
+    assert tools._is_httpql_error(RuntimeError("HTTPQL parse error at column 4"))
+    assert tools._is_httpql_error(RuntimeError("failed to parse filter"))
+    assert not tools._is_httpql_error(RuntimeError("Transport is already connected"))
+
+
+def test_httpql_error_preserves_message_and_query() -> None:
+    exc = RuntimeError("HTTPQL parse error: unexpected token at column 12")
+    payload = json.loads(tools._httpql_error(exc, 'resp.code.eq:"200"'))
+    assert payload["success"] is False
+    assert "unexpected token at column 12" in payload["error"]
+    assert payload["httpql_filter"] == 'resp.code.eq:"200"'
+    assert "AND / OR" in payload["hint"]

tests/test_runner_rate_limit.py

@@ -37,7 +37,8 @@ async def test_persistent_rate_limit_stops_gracefully(
             model="openai/gpt-4o",
             reasoning_effort="high",
             force_required_tool_choice=False,
-        )
+        ),
+        runtime=types.SimpleNamespace(max_context_images=3),
     )
     monkeypatch.setattr(runner, "load_settings", lambda: settings)
     monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
@@ -54,8 +55,8 @@ async def test_persistent_rate_limit_stops_gracefully(
     async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
         return None
 
-    monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)
-    monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)
+    monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)  # type: ignore[attr-defined]
+    monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)  # type: ignore[attr-defined]
 
     monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
     monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: "")