feat(migration): phase 4 — sandbox capability + healthcheck + session manager

Three modules under strix/sandbox/ that bring the per-scan container
plumbing in line with the SDK's capability model:

- healthcheck.py: wait_for_http_ready (FastAPI tool server /health)
  and wait_for_tcp_ready (Caido proxy port — no /health endpoint).
  Connect/timeout errors continue polling; the timeout error message
  carries the last failure class so a stuck scan tells you whether the
  port refused, hung, or returned a non-2xx.

- caido_capability.py: CaidoCapability subclasses agents.sandbox.
  capabilities.Capability and wires three concerns:
  1. process_manifest injects http_proxy / https_proxy / ALL_PROXY
     env vars pointing at the in-container Caido listener.
  2. tools() returns the seven Caido SDK function tools from Phase 2.5
     so the SDK runtime auto-merges them with each agent's tool list.
  3. bind() schedules an asyncio.gather of both healthcheck probes;
     StrixOrchestrationHooks.on_agent_start awaits the resulting
     task before the first LLM call.
  Pydantic v2 PrivateAttr is used for the underscore-prefixed runtime
  fields (Pydantic forbids underscore-prefixed model fields).

- session_manager.py: per-scan_id cache. create_or_reuse builds the
  StrixDockerSandboxClient with docker.from_env() (the SDK's docker
  client now requires an explicit DockerSDKClient instance at init),
  constructs the Manifest via Environment(value=...) (a flat dict is
  silently dropped by Pydantic), resolves the host-side mapped ports
  via session._resolve_exposed_port, configures the capability with
  those ports *before* binding, and returns a bundle dict the
  per-agent context reads to populate tool_server_host_port /
  caido_host_port / bearer. cleanup is best-effort: a Docker daemon
  error during delete is logged and swallowed so a stranded
  container doesn't block the next scan.

Tests: 21 new tests in tests/sandbox/ — healthcheck happy path /
polling-through-failures / timeout for both HTTP and TCP probes (the
TCP test uses a real local listener, no mocks); CaidoCapability env
injection / tool list / bind scheduling / configure_host_ports;
session_manager full create flow, cache reuse, custom timeout, cleanup
including the Docker-daemon-failure swallow path.

mypy override added for docker.* (no upstream stubs); per-file ruff
TC002 ignore added for caido_capability.py — agents.tool.Tool is used
at runtime for the cached _CAIDO_TOOLS tuple.

Refs: PLAYBOOK.md §3.1-3.3, AUDIT.md §2.5 (C5).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/sandbox/test_caido_capability.py

@@ -0,0 +1,156 @@
+"""Phase 4 tests for CaidoCapability.
+
+The capability has three observable behaviors that need parity with the
+PLAYBOOK contract:
+
+1. ``process_manifest`` injects http_proxy / https_proxy / ALL_PROXY env
+   vars into the manifest's ``Environment.value`` dict.
+2. ``tools()`` returns the seven Caido SDK function tools we wrapped in
+   Phase 2.5 — same instances, in the documented order.
+3. ``bind`` schedules an aggregated healthcheck task; the orchestration
+   hook later awaits it on first agent start.
+
+The healthcheck task itself is exercised by the healthcheck unit tests;
+here we only verify the wiring (task created, name set, points at the
+right ports).
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import MagicMock, patch
+
+import pytest
+from agents.sandbox.entries import LocalDir
+from agents.sandbox.manifest import Environment, Manifest
+
+from strix.sandbox.caido_capability import CaidoCapability
+
+
+def test_capability_type_and_default_state() -> None:
+    cap = CaidoCapability()
+    assert cap.type == "caido"
+    assert cap._healthcheck_task is None
+    assert cap._tool_server_host_port is None
+    assert cap._caido_host_port is None
+
+
+def test_process_manifest_injects_proxy_env_vars(tmp_path: object) -> None:
+    """Existing env vars must be preserved; proxy keys are added."""
+    cap = CaidoCapability()
+    manifest = Manifest(
+        environment=Environment(
+            value={"PYTHONUNBUFFERED": "1", "TOOL_SERVER_TOKEN": "abc"},
+        ),
+    )
+    out = cap.process_manifest(manifest)
+    env = out.environment.value
+    # Pre-existing entries preserved.
+    assert env["PYTHONUNBUFFERED"] == "1"
+    assert env["TOOL_SERVER_TOKEN"] == "abc"
+    # Proxy entries injected, all pointing at the in-container Caido port.
+    assert env["http_proxy"] == "http://127.0.0.1:48080"
+    assert env["https_proxy"] == "http://127.0.0.1:48080"
+    assert env["ALL_PROXY"] == "http://127.0.0.1:48080"
+
+
+def test_process_manifest_handles_missing_environment() -> None:
+    """A manifest without env entries should still get the proxy block."""
+    cap = CaidoCapability()
+    # ``LocalDir`` requires a real path on disk; use a temp one to satisfy
+    # the validator without actually mounting anything.
+    manifest = Manifest(entries={"src": LocalDir(src="/tmp")})
+    out = cap.process_manifest(manifest)
+    env = out.environment.value
+    assert env["http_proxy"] == "http://127.0.0.1:48080"
+
+
+def test_tools_returns_seven_caido_tools_in_order() -> None:
+    cap = CaidoCapability()
+    names = [t.name for t in cap.tools()]
+    assert names == [
+        "list_requests",
+        "view_request",
+        "send_request",
+        "repeat_request",
+        "scope_rules",
+        "list_sitemap",
+        "view_sitemap_entry",
+    ]
+
+
+def test_tools_returns_a_fresh_list_per_call() -> None:
+    """SDK convention — caller may mutate the returned list."""
+    cap = CaidoCapability()
+    a = cap.tools()
+    b = cap.tools()
+    assert a == b
+    assert a is not b
+
+
+@pytest.mark.asyncio
+async def test_instructions_mentions_caido_and_tools() -> None:
+    cap = CaidoCapability()
+    out = await cap.instructions(Manifest())
+    assert out is not None
+    assert "<caido_proxy>" in out
+    # Every tool name appears verbatim so the model knows what's available.
+    for name in (
+        "list_requests",
+        "view_request",
+        "send_request",
+        "repeat_request",
+        "scope_rules",
+        "list_sitemap",
+        "view_sitemap_entry",
+    ):
+        assert name in out
+
+
+def test_configure_host_ports_stores_both() -> None:
+    cap = CaidoCapability()
+    cap.configure_host_ports(tool_server_host_port=12345, caido_host_port=12346)
+    assert cap._tool_server_host_port == 12345
+    assert cap._caido_host_port == 12346
+
+
+@pytest.mark.asyncio
+async def test_bind_without_configured_ports_skips_healthcheck() -> None:
+    """If the session manager forgets to configure ports, bind shouldn't
+    schedule a probe against ``None`` — it should warn and no-op.
+    """
+    cap = CaidoCapability()
+    fake_session = MagicMock()
+    cap.bind(fake_session)
+    assert cap._healthcheck_task is None
+    assert cap.session is fake_session
+
+
+@pytest.mark.asyncio
+async def test_bind_schedules_healthcheck_task_when_ports_configured() -> None:
+    """The hook chain (StrixOrchestrationHooks.on_agent_start) awaits this
+    task — it must exist as an asyncio.Task with a useful name.
+    """
+    cap = CaidoCapability()
+    cap.configure_host_ports(tool_server_host_port=54321, caido_host_port=54322)
+    fake_session = MagicMock()
+
+    # Patch the actual probes so we don't try to connect for real.
+    async def _fake_probe(*args: object, **kwargs: object) -> None:
+        return None
+
+    with (
+        patch(
+            "strix.sandbox.caido_capability.wait_for_http_ready",
+            side_effect=_fake_probe,
+        ),
+        patch(
+            "strix.sandbox.caido_capability.wait_for_tcp_ready",
+            side_effect=_fake_probe,
+        ),
+    ):
+        cap.bind(fake_session)
+        assert cap._healthcheck_task is not None
+        assert isinstance(cap._healthcheck_task, asyncio.Task)
+        assert "caido-healthcheck-54321" in cap._healthcheck_task.get_name()
+        await cap._healthcheck_task  # must complete without error

tests/sandbox/test_healthcheck.py

@@ -0,0 +1,171 @@
+"""Phase 4 tests for the sandbox port readiness probes.
+
+The two helpers (``wait_for_http_ready`` and ``wait_for_tcp_ready``)
+gate session bring-up, so a regression here would mean every fresh
+scan hits a connection-refused on its first tool call. Tests cover:
+
+- Happy path returns when the probe succeeds.
+- Polling continues across transient failures.
+- Timeout raises ``SandboxNotReadyError`` with a useful last-error.
+- Real ``asyncio.open_connection`` against a local listener verifies
+  the TCP probe end-to-end (no mocking — the helper is small enough
+  that a real socket is the cheaper test).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import httpx
+import pytest
+
+from strix.sandbox.healthcheck import (
+    SandboxNotReadyError,
+    wait_for_http_ready,
+    wait_for_tcp_ready,
+)
+
+
+# --- HTTP probe ----------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_wait_for_http_ready_returns_immediately_on_2xx() -> None:
+    response = MagicMock(spec=httpx.Response)
+    response.status_code = 200
+    client = AsyncMock()
+    client.get = AsyncMock(return_value=response)
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=None)
+
+    with patch("strix.sandbox.healthcheck.httpx.AsyncClient", return_value=client):
+        await wait_for_http_ready("http://localhost:9999/health", timeout=1)
+
+    assert client.get.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_wait_for_http_ready_polls_through_connect_errors() -> None:
+    """Two connect errors followed by a 200 — the helper should keep going."""
+    response_ok = MagicMock(spec=httpx.Response)
+    response_ok.status_code = 200
+
+    side_effects: list[Any] = [
+        httpx.ConnectError("conn refused"),
+        httpx.ConnectError("conn refused"),
+        response_ok,
+    ]
+
+    client = AsyncMock()
+    client.get = AsyncMock(side_effect=side_effects)
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=None)
+
+    with patch("strix.sandbox.healthcheck.httpx.AsyncClient", return_value=client):
+        await wait_for_http_ready(
+            "http://localhost:9999/health",
+            timeout=5,
+            poll_interval=0.01,
+        )
+    assert client.get.await_count == 3
+
+
+@pytest.mark.asyncio
+async def test_wait_for_http_ready_raises_after_timeout() -> None:
+    client = AsyncMock()
+    client.get = AsyncMock(side_effect=httpx.ConnectError("nope"))
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=None)
+
+    with (
+        patch(
+            "strix.sandbox.healthcheck.httpx.AsyncClient",
+            return_value=client,
+        ),
+        pytest.raises(SandboxNotReadyError) as exc_info,
+    ):
+        await wait_for_http_ready(
+            "http://localhost:9999/health",
+            timeout=0.3,
+            poll_interval=0.05,
+        )
+
+    err = str(exc_info.value)
+    assert "http://localhost:9999/health" in err
+    assert "ConnectError" in err
+
+
+@pytest.mark.asyncio
+async def test_wait_for_http_ready_treats_5xx_as_not_ready() -> None:
+    response_500 = MagicMock(spec=httpx.Response)
+    response_500.status_code = 500
+    response_ok = MagicMock(spec=httpx.Response)
+    response_ok.status_code = 200
+
+    client = AsyncMock()
+    client.get = AsyncMock(side_effect=[response_500, response_ok])
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=None)
+
+    with patch("strix.sandbox.healthcheck.httpx.AsyncClient", return_value=client):
+        await wait_for_http_ready(
+            "http://localhost:9999/health",
+            timeout=2,
+            poll_interval=0.01,
+        )
+    assert client.get.await_count == 2
+
+
+# --- TCP probe -----------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_wait_for_tcp_ready_against_real_listener() -> None:
+    """Spin up a local TCP echo server and verify the probe connects."""
+
+    async def _server_handler(
+        reader: asyncio.StreamReader,
+        writer: asyncio.StreamWriter,
+    ) -> None:
+        # Drain any bytes the test sends, then close.
+        await reader.read(0)
+        writer.close()
+        with contextlib.suppress(OSError):
+            await writer.wait_closed()
+
+    server = await asyncio.start_server(_server_handler, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    try:
+        await wait_for_tcp_ready("127.0.0.1", port, timeout=2, poll_interval=0.05)
+    finally:
+        server.close()
+        await server.wait_closed()
+
+
+@pytest.mark.asyncio
+async def test_wait_for_tcp_ready_raises_when_port_closed() -> None:
+    async def _no_handler(
+        _reader: asyncio.StreamReader,
+        _writer: asyncio.StreamWriter,
+    ) -> None:
+        return
+
+    # Bind and immediately close to claim a definitely-unused port number.
+    server = await asyncio.start_server(_no_handler, "127.0.0.1", 0)
+    closed_port = server.sockets[0].getsockname()[1]
+    server.close()
+    await server.wait_closed()
+
+    with pytest.raises(SandboxNotReadyError) as exc_info:
+        await wait_for_tcp_ready(
+            "127.0.0.1",
+            closed_port,
+            timeout=0.3,
+            poll_interval=0.05,
+        )
+
+    err = str(exc_info.value)
+    assert f"127.0.0.1:{closed_port}" in err

tests/sandbox/test_session_manager.py

@@ -0,0 +1,206 @@
+"""Phase 4 tests for the per-scan sandbox session manager.
+
+We don't spin up real Docker here — the ``StrixDockerSandboxClient`` is
+patched and we assert on the manifest / options / bundle shape. Goals:
+
+- Cache hit: a second ``create_or_reuse(scan_id, ...)`` returns the same
+  bundle without calling client.create twice.
+- Manifest carries the right env vars (TOOL_SERVER_TOKEN, container ports,
+  STRIX_SANDBOX_EXECUTION_TIMEOUT, PYTHONUNBUFFERED).
+- The Docker client options request both container ports be exposed.
+- Capability is configured with the resolved host ports *before* bind,
+  so its healthcheck task probes the right ones.
+- Bundle is cached and surfaces in ``cached_scan_ids``.
+- ``cleanup`` cancels the healthcheck task and calls ``client.delete``;
+  errors during delete are swallowed.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from strix.sandbox import session_manager
+from strix.sandbox.caido_capability import CaidoCapability
+
+
+@pytest.fixture(autouse=True)
+def _isolate_cache() -> Iterator[None]:
+    session_manager._reset_cache_for_tests()
+    yield
+    session_manager._reset_cache_for_tests()
+
+
+def _noop_bind(_self: Any, _session: Any) -> None:
+    """Stand-in for CaidoCapability.bind that skips the healthcheck task."""
+
+
+def _make_endpoint(port: int) -> Any:
+    ep = MagicMock()
+    ep.port = port
+    ep.host = "127.0.0.1"
+    ep.tls = False
+    return ep
+
+
+def _make_client_and_session(
+    *,
+    tool_port: int = 12001,
+    caido_port: int = 12002,
+) -> tuple[Any, Any]:
+    """Build a fake DockerSandboxClient and session pair."""
+    session = MagicMock()
+    session._resolve_exposed_port = AsyncMock(
+        side_effect=lambda port: _make_endpoint(
+            tool_port if port == 48081 else caido_port,
+        ),
+    )
+    client = MagicMock()
+    client.create = AsyncMock(return_value=session)
+    client.delete = AsyncMock()
+    return client, session
+
+
+@pytest.mark.asyncio
+async def test_create_or_reuse_creates_new_session(tmp_path: Any) -> None:
+    client, session = _make_client_and_session()
+    # Patch the capability's bind to a no-op so we don't spin up the
+    # healthcheck task in unit tests.
+    with (
+        patch(
+            "strix.sandbox.session_manager.StrixDockerSandboxClient",
+            return_value=client,
+        ),
+        patch.object(CaidoCapability, "bind", _noop_bind),
+    ):
+        bundle = await session_manager.create_or_reuse(
+            "scan-1",
+            image="strix-sandbox:test",
+            sources_path=tmp_path,
+        )
+
+    # Bundle shape.
+    assert bundle["client"] is client
+    assert bundle["session"] is session
+    assert bundle["tool_server_host_port"] == 12001
+    assert bundle["caido_host_port"] == 12002
+    assert isinstance(bundle["bearer"], str) and len(bundle["bearer"]) >= 32
+    assert isinstance(bundle["capability"], CaidoCapability)
+    # Capability got the resolved host ports BEFORE bind would have run.
+    assert bundle["capability"]._tool_server_host_port == 12001
+    assert bundle["capability"]._caido_host_port == 12002
+
+    # client.create called exactly once with manifest + exposed ports.
+    assert client.create.await_count == 1
+    options = client.create.await_args.kwargs["options"]
+    assert options.image == "strix-sandbox:test"
+    assert set(options.exposed_ports) == {48080, 48081}
+
+    manifest = client.create.await_args.kwargs["manifest"]
+    env = manifest.environment.value
+    assert env["TOOL_SERVER_TOKEN"] == bundle["bearer"]
+    assert env["TOOL_SERVER_PORT"] == "48081"
+    assert env["CAIDO_PORT"] == "48080"
+    assert env["STRIX_SANDBOX_EXECUTION_TIMEOUT"] == "120"
+    assert env["PYTHONUNBUFFERED"] == "1"
+    assert env["HOST_GATEWAY"] == "host.docker.internal"
+
+
+@pytest.mark.asyncio
+async def test_create_or_reuse_returns_cached_bundle(tmp_path: Any) -> None:
+    client, _ = _make_client_and_session()
+    with (
+        patch(
+            "strix.sandbox.session_manager.StrixDockerSandboxClient",
+            return_value=client,
+        ),
+        patch.object(CaidoCapability, "bind", _noop_bind),
+    ):
+        first = await session_manager.create_or_reuse(
+            "scan-X",
+            image="i",
+            sources_path=tmp_path,
+        )
+        second = await session_manager.create_or_reuse(
+            "scan-X",
+            image="i",
+            sources_path=tmp_path,
+        )
+
+    assert first is second
+    assert client.create.await_count == 1
+    assert "scan-X" in session_manager.cached_scan_ids()
+
+
+@pytest.mark.asyncio
+async def test_create_or_reuse_passes_custom_execution_timeout(tmp_path: Any) -> None:
+    client, _ = _make_client_and_session()
+    with (
+        patch(
+            "strix.sandbox.session_manager.StrixDockerSandboxClient",
+            return_value=client,
+        ),
+        patch.object(CaidoCapability, "bind", _noop_bind),
+    ):
+        await session_manager.create_or_reuse(
+            "scan-2",
+            image="i",
+            sources_path=tmp_path,
+            execution_timeout=300,
+        )
+
+    manifest = client.create.await_args.kwargs["manifest"]
+    assert manifest.environment.value["STRIX_SANDBOX_EXECUTION_TIMEOUT"] == "300"
+
+
+@pytest.mark.asyncio
+async def test_cleanup_calls_delete_and_drops_cache(tmp_path: Any) -> None:
+    client, session = _make_client_and_session()
+    with (
+        patch(
+            "strix.sandbox.session_manager.StrixDockerSandboxClient",
+            return_value=client,
+        ),
+        patch.object(CaidoCapability, "bind", _noop_bind),
+    ):
+        await session_manager.create_or_reuse(
+            "scan-3",
+            image="i",
+            sources_path=tmp_path,
+        )
+        assert "scan-3" in session_manager.cached_scan_ids()
+        await session_manager.cleanup("scan-3")
+
+    client.delete.assert_awaited_once_with(session)
+    assert "scan-3" not in session_manager.cached_scan_ids()
+
+
+@pytest.mark.asyncio
+async def test_cleanup_swallows_delete_errors(tmp_path: Any) -> None:
+    """A flaky Docker daemon shouldn't prevent cache eviction."""
+    client, _ = _make_client_and_session()
+    client.delete = AsyncMock(side_effect=RuntimeError("docker daemon went away"))
+    with (
+        patch(
+            "strix.sandbox.session_manager.StrixDockerSandboxClient",
+            return_value=client,
+        ),
+        patch.object(CaidoCapability, "bind", _noop_bind),
+    ):
+        await session_manager.create_or_reuse(
+            "scan-4",
+            image="i",
+            sources_path=tmp_path,
+        )
+        await session_manager.cleanup("scan-4")  # must not raise
+
+    assert "scan-4" not in session_manager.cached_scan_ids()
+
+
+@pytest.mark.asyncio
+async def test_cleanup_unknown_scan_is_noop() -> None:
+    """No cached entry → cleanup is a quiet no-op."""
+    await session_manager.cleanup("never-existed")  # must not raise