feat(migration): phase 2.1-2.3 — sandbox dispatch + thin slice tool wrappers

Phase 2.1 — sandbox dispatch helper:
- strix/tools/_sandbox_dispatch.py: post_to_sandbox() centralizes the
  host->container HTTP wire format. Connect=10s, read=150s timeouts mirror
  legacy executor.py. 50 MB response cap (C18) prevents OOM from a runaway
  tool. All errors surface as {"error": str} so the model can recover
  instead of the run dying.

Phase 2.2 — C6 lock-protected JSONL writes:
- strix/tools/notes/notes_actions.py: notes.jsonl appends are now wrapped
  in _notes_lock so concurrent agents can't interleave half-written lines.
  Regression test in test_notes_jsonl_concurrency.py verifies 1000 parallel
  writes produce exactly 1000 valid JSON lines.

Phase 2.3 — thin-slice SDK wrappers (think + todo + notes):
- strix/tools/_legacy_adapter.py: LegacyAgentStateAdapter shim — exposes
  just enough surface (.agent_id) for legacy tools that close over
  agent_state, sourced from ctx.context['agent_id'].
- strix/tools/thinking/thinking_sdk_tools.py: 1 tool (think).
- strix/tools/todo/todo_sdk_tools.py: 6 tools (create/list/update/done/
  pending/delete) with bulk-form preserved.
- strix/tools/notes/notes_sdk_tools.py: 5 tools (create/list/get/update/
  delete) with asyncio.to_thread around the lock-protected file I/O.

Tests: 22 new tests pass (10 sandbox dispatch + 2 concurrency + 10 SDK
local). Full suite still green.

Per-file ruff ignores added for SDK wrapper files: TC002 (RunContextWrapper
must be runtime-importable because the SDK calls get_type_hints() to
derive the JSON schema) and PLR0911 (sandbox dispatch's 10 short-circuit
returns are intentional, each a distinct documented failure mode).

Refs: PLAYBOOK.md §3.4, AUDIT_R3.md C6/C18.

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

pyproject.toml

@@ -283,6 +283,16 @@ ignore = [
 "strix/tools/_decorator.py" = [
     "TC002",  # FunctionTool imported for annotation
 ]
+# SDK function-tool wrappers: the SDK calls get_type_hints() at registration
+# time to derive the JSON schema, which evaluates annotations at runtime —
+# so RunContextWrapper must be imported eagerly, not under TYPE_CHECKING.
+"strix/tools/todo/todo_sdk_tools.py" = ["TC002"]
+"strix/tools/notes/notes_sdk_tools.py" = ["TC002"]
+"strix/tools/thinking/thinking_sdk_tools.py" = ["TC002"]
+# Sandbox dispatch helper has many short-circuit error returns (auth fail,
+# size cap, decode fail, etc). Each is a distinct, documented failure mode
+# the model needs to see verbatim — collapsing them harms readability.
+"strix/tools/_sandbox_dispatch.py" = ["PLR0911"]
 # StrixSession + StrixTracingProcessor catch broad Exception intentionally:
 # the whole point is that compressor / sanitizer / disk failures must not
 # tear down the agent run (C10, C16). Calls already log at exception level.

strix/tools/_legacy_adapter.py

@@ -0,0 +1,57 @@
+"""Shim that lets SDK function tools call legacy ``agent_state``-style functions.
+
+The legacy harness's tools (notes, todos, reporting, …) take an
+``agent_state`` argument with shape ``state.agent_id`` for per-agent silo
+keying. Under the SDK migration the equivalent identity lives in
+``RunContextWrapper.context["agent_id"]``.
+
+Rather than rewrite every tool body, SDK function-tool wrappers build a
+tiny adapter from the context dict and pass it to the legacy function.
+The legacy code path remains untouched (the legacy executor still calls
+its tools with the real ``AgentState``).
+
+Used by:
+    - ``tools/todo/todo_sdk_tools.py``
+    - ``tools/notes/notes_sdk_tools.py``
+    - ``tools/reporting/reporting_sdk_tools.py``
+    - any other local tool that closes over ``agent_state.agent_id``
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from agents import RunContextWrapper
+
+
+@dataclass
+class LegacyAgentStateAdapter:
+    """Just enough surface for legacy tools that read ``state.agent_id``.
+
+    Don't rely on this for new code — it's only here to avoid touching
+    the legacy ``*_actions.py`` modules during the migration. New SDK
+    tools should read ``ctx.context["agent_id"]`` directly.
+    """
+
+    agent_id: str
+
+
+def adapter_from_ctx(
+    ctx: RunContextWrapper,
+    default_agent_id: str = "sdk-default",
+) -> LegacyAgentStateAdapter:
+    """Build a ``LegacyAgentStateAdapter`` from an SDK run context.
+
+    Falls back to ``default_agent_id`` when context is missing or its
+    ``agent_id`` is unset — keeps tests and CLI dry-runs working without
+    a fully-populated context.
+    """
+    inner = getattr(ctx, "context", None)
+    if isinstance(inner, dict):
+        agent_id = inner.get("agent_id") or default_agent_id
+    else:
+        agent_id = default_agent_id
+    return LegacyAgentStateAdapter(agent_id=str(agent_id))

strix/tools/_sandbox_dispatch.py

@@ -0,0 +1,132 @@
+"""post_to_sandbox — host-to-container HTTP transport for sandbox tools.
+
+Every Strix tool that runs inside the Kali container (browser, terminal,
+python, the seven Caido tools) has the same wire shape: POST a JSON body
+to ``http://localhost:{tool_server_host_port}/execute`` with a Bearer
+token header and ``{"agent_id", "tool_name", "kwargs"}`` as the body.
+
+This helper centralizes that transport so:
+
+- Every sandbox tool gets the same timeout policy
+  (``connect=10s`` / ``read=150s``).
+- Every sandbox tool inherits the same response-size cap (50 MB) so a
+  runaway tool body cannot OOM the host (C18).
+- Auth/transport errors surface as predictable error strings instead of
+  exceptions, so the model can retry / pick a different tool without the
+  run dying.
+
+References:
+    - PLAYBOOK.md §3.4
+    - AUDIT_R3.md C18 (sandbox response size cap)
+    - HARNESS_WIKI.md §7.2 (legacy executor.py wire format we mirror)
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+import httpx
+
+
+if TYPE_CHECKING:
+    from agents import RunContextWrapper
+
+
+logger = logging.getLogger(__name__)
+
+
+# Connect: how long to wait for the TCP handshake to complete.
+# Read: how long the tool may spend executing before we abandon the call.
+# Mirrors the legacy executor.py (``SANDBOX_EXECUTION_TIMEOUT = 120 + 30``).
+_SANDBOX_TIMEOUT = httpx.Timeout(connect=10.0, read=150.0, write=150.0, pool=150.0)
+
+#: Cap on response body size from the tool server. Anything bigger is
+#: replaced by an error string so the model sees something coherent and
+#: the host doesn't OOM trying to allocate the buffer (C18).
+_MAX_RESPONSE_BYTES = 50 * 1024 * 1024  # 50 MB
+
+
+def _ctx_dict(ctx: RunContextWrapper) -> dict[str, Any] | None:
+    """Return ``ctx.context`` if it's a dict, else ``None``.
+
+    Strix's runtime always passes a dict (``make_agent_context``); other
+    callers might not. Be defensive so a sandbox tool never raises just
+    because the context shape is wrong.
+    """
+    inner = getattr(ctx, "context", None)
+    return inner if isinstance(inner, dict) else None
+
+
+async def post_to_sandbox(
+    ctx: RunContextWrapper,
+    tool_name: str,
+    kwargs: dict[str, Any],
+) -> dict[str, Any]:
+    """POST a tool invocation to the in-container FastAPI tool server.
+
+    Returns:
+        On success: ``{"result": <whatever the tool returned>}``.
+        On any failure: ``{"error": "<human-readable error string>"}``.
+
+    Never raises. Tool authors call this and pass the return value
+    straight to the model (or extract ``result`` for further shaping).
+    """
+    inner = _ctx_dict(ctx)
+    if inner is None:
+        return {"error": "Sandbox not initialized: context is missing or not a dict."}
+
+    port = inner.get("tool_server_host_port")
+    token = inner.get("sandbox_token")
+    agent_id = inner.get("agent_id", "unknown")
+
+    if not port or not token:
+        return {"error": "Sandbox not initialized: tool server port or token missing."}
+
+    url = f"http://127.0.0.1:{port}/execute"
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+    }
+    body = {"agent_id": agent_id, "tool_name": tool_name, "kwargs": kwargs}
+
+    try:
+        async with httpx.AsyncClient(timeout=_SANDBOX_TIMEOUT) as client:
+            response = await client.post(url, json=body, headers=headers)
+    except httpx.TimeoutException:
+        return {
+            "error": (f"Sandbox tool '{tool_name}' timed out after {_SANDBOX_TIMEOUT.read}s."),
+        }
+    except httpx.RequestError as e:
+        # ConnectError, ReadError, NetworkError, etc.
+        return {"error": f"Sandbox connection failed: {e!s}"[:300]}
+
+    if response.status_code == 401:
+        return {"error": "Sandbox authorization failed (Bearer token invalid)."}
+    if response.status_code >= 400:
+        return {
+            "error": (
+                f"Sandbox tool '{tool_name}' failed with HTTP "
+                f"{response.status_code}: {response.text[:300]}"
+            ),
+        }
+
+    # Cap response size before parsing so a 1 GB rogue payload never lands
+    # in our heap. Most legitimate tool responses are well under 100 KB.
+    raw = response.content
+    if len(raw) > _MAX_RESPONSE_BYTES:
+        return {
+            "error": (f"Sandbox response too large ({len(raw)} bytes; max {_MAX_RESPONSE_BYTES})."),
+        }
+
+    try:
+        data: Any = response.json()
+    except ValueError:
+        return {
+            "error": (f"Sandbox tool '{tool_name}' returned non-JSON: {response.text[:200]}"),
+        }
+
+    if not isinstance(data, dict):
+        return {"error": f"Sandbox tool '{tool_name}' returned non-object JSON."}
+
+    return data

strix/tools/notes/notes_actions.py

@@ -38,6 +38,12 @@ def _get_notes_jsonl_path() -> Path | None:
 
 
 def _append_note_event(op: str, note_id: str, note: dict[str, Any] | None = None) -> None:
+    """Append one note operation to the run's ``notes/notes.jsonl``.
+
+    C6 (AUDIT_R2.md §1.1): hold ``_notes_lock`` across the file open + write
+    so two concurrent agents (or two parallel SDK tool calls in Phase 6)
+    cannot interleave bytes mid-line and corrupt the JSONL.
+    """
     notes_path = _get_notes_jsonl_path()
     if not notes_path:
         return
@@ -50,7 +56,7 @@ def _append_note_event(op: str, note_id: str, note: dict[str, Any] | None = None
     if note is not None:
         event["note"] = note
 
-    with notes_path.open("a", encoding="utf-8") as f:
+    with _notes_lock, notes_path.open("a", encoding="utf-8") as f:
         f.write(f"{json.dumps(event, ensure_ascii=True)}\n")
 
 

strix/tools/notes/notes_sdk_tools.py

@@ -0,0 +1,117 @@
+"""SDK function-tool wrappers for the legacy notes tools.
+
+Five tools, all module-global (no per-agent silo). The legacy
+``notes_actions.py`` module already implements JSONL persistence and
+wiki Markdown rendering; these wrappers are pure delegation.
+
+The C6 fix (lock-protected JSONL writes) was applied directly to the
+legacy module, so both code paths benefit.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Any
+
+from agents import RunContextWrapper
+
+from strix.tools._decorator import strix_tool
+from strix.tools.notes import notes_actions as _legacy
+
+
+def _dump(result: dict[str, Any]) -> str:
+    return json.dumps(result, ensure_ascii=False, default=str)
+
+
+@strix_tool(timeout=30)
+async def create_note(
+    ctx: RunContextWrapper,
+    title: str,
+    content: str,
+    category: str = "general",
+    tags: list[str] | None = None,
+) -> str:
+    """Create a note in the current run's notes store.
+
+    Notes are persisted to ``run_dir/notes/notes.jsonl`` and (for the
+    ``wiki`` category) rendered as Markdown to ``run_dir/wiki/<slug>.md``.
+
+    Args:
+        title: Required, non-empty title.
+        content: Note body. Markdown is preserved.
+        category: One of ``"general" | "findings" | "methodology" |
+            "questions" | "plan" | "wiki"``.
+        tags: Optional list of free-form tags.
+    """
+    # The legacy function does file I/O under a threading.RLock.
+    # Wrap in to_thread so we don't block the event loop while waiting
+    # on the lock or fsync.
+    result = await asyncio.to_thread(
+        _legacy.create_note,
+        title=title,
+        content=content,
+        category=category,
+        tags=tags,
+    )
+    return _dump(result)
+
+
+@strix_tool(timeout=30)
+async def list_notes(
+    ctx: RunContextWrapper,
+    category: str | None = None,
+    tags: list[str] | None = None,
+    search: str | None = None,
+    include_content: bool = False,
+) -> str:
+    """List notes, optionally filtered.
+
+    Args:
+        category: Filter by category.
+        tags: Filter to notes that have any of these tags.
+        search: Substring match against title and content.
+        include_content: When False (default), entries get a ``content_preview``;
+            when True, full content is included.
+    """
+    result = await asyncio.to_thread(
+        _legacy.list_notes,
+        category=category,
+        tags=tags,
+        search=search,
+        include_content=include_content,
+    )
+    return _dump(result)
+
+
+@strix_tool(timeout=30)
+async def get_note(ctx: RunContextWrapper, note_id: str) -> str:
+    """Fetch one note by its 5-char ID. Returns full content."""
+    result = await asyncio.to_thread(_legacy.get_note, note_id=note_id)
+    return _dump(result)
+
+
+@strix_tool(timeout=30)
+async def update_note(
+    ctx: RunContextWrapper,
+    note_id: str,
+    title: str | None = None,
+    content: str | None = None,
+    tags: list[str] | None = None,
+) -> str:
+    """Update a note's title, content, or tags. Pass ``None`` to leave a field unchanged."""
+    result = await asyncio.to_thread(
+        _legacy.update_note,
+        note_id=note_id,
+        title=title,
+        content=content,
+        tags=tags,
+    )
+    return _dump(result)
+
+
+@strix_tool(timeout=30)
+async def delete_note(ctx: RunContextWrapper, note_id: str) -> str:
+    """Delete a note. For wiki notes, also removes the rendered Markdown file."""
+    result = await asyncio.to_thread(_legacy.delete_note, note_id=note_id)
+    return _dump(result)

strix/tools/thinking/thinking_sdk_tools.py

@@ -0,0 +1,32 @@
+"""SDK function-tool wrapper for the legacy ``think`` tool.
+
+Pattern: thin async wrapper that delegates to the legacy implementation
+in :mod:`strix.tools.thinking.thinking_actions`. The legacy function is
+sync and pure (no I/O), so we don't even need ``asyncio.to_thread``.
+
+Validates the simplest tool-port pattern: legacy function in, JSON string
+out, no sandbox involvement.
+"""
+
+from __future__ import annotations
+
+import json
+
+from strix.tools._decorator import strix_tool
+from strix.tools.thinking.thinking_actions import think as _legacy_think
+
+
+@strix_tool(timeout=10)
+async def think(thought: str) -> str:
+    """Record a private chain-of-thought note without taking any action.
+
+    The "think" tool is the planning escape hatch for situations where a
+    message-without-tool-call would otherwise halt the run (per the
+    interactive-mode tool-call requirement). The thought itself is
+    recorded but produces no side effects.
+
+    Args:
+        thought: The agent's reasoning to record. Must be non-empty.
+    """
+    result = _legacy_think(thought)
+    return json.dumps(result, ensure_ascii=False)

strix/tools/todo/todo_sdk_tools.py

@@ -0,0 +1,146 @@
+"""SDK function-tool wrappers for the legacy todo tools.
+
+Six tools, all in-memory, all per-agent (keyed by ``ctx.context["agent_id"]``
+through :class:`LegacyAgentStateAdapter`). Bulk forms are preserved —
+``todos`` / ``updates`` / ``todo_ids`` accept JSON strings or comma-separated
+strings the same way the legacy XML schema documented.
+
+Pattern: thin async wrappers that delegate to the legacy implementations
+in :mod:`strix.tools.todo.todo_actions`. Legacy code is untouched.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from agents import RunContextWrapper
+
+from strix.tools._decorator import strix_tool
+from strix.tools._legacy_adapter import adapter_from_ctx
+from strix.tools.todo import todo_actions as _legacy
+
+
+def _dump(result: dict[str, Any]) -> str:
+    """JSON-dump a legacy result dict for the model. ``ensure_ascii=False``
+    so unicode flows through; ``default=str`` to handle stray datetimes."""
+    return json.dumps(result, ensure_ascii=False, default=str)
+
+
+@strix_tool(timeout=30)
+async def create_todo(
+    ctx: RunContextWrapper,
+    title: str | None = None,
+    description: str | None = None,
+    priority: str = "normal",
+    todos: str | None = None,
+) -> str:
+    """Create one or many todos for the current agent.
+
+    Args:
+        title: Title of a single todo (alternative to bulk ``todos``).
+        description: Optional details for the single todo.
+        priority: ``"low" | "normal" | "high" | "critical"``.
+        todos: Optional JSON string or comma-separated list for bulk create.
+    """
+    state = adapter_from_ctx(ctx)
+    return _dump(
+        _legacy.create_todo(
+            agent_state=state,
+            title=title,
+            description=description,
+            priority=priority,
+            todos=todos,
+        ),
+    )
+
+
+@strix_tool(timeout=30)
+async def list_todos(
+    ctx: RunContextWrapper,
+    status: str | None = None,
+    priority: str | None = None,
+) -> str:
+    """List the current agent's todos, sorted by status then priority.
+
+    Args:
+        status: Optional ``"pending" | "in_progress" | "done"`` filter.
+        priority: Optional ``"low" | "normal" | "high" | "critical"`` filter.
+    """
+    state = adapter_from_ctx(ctx)
+    return _dump(_legacy.list_todos(agent_state=state, status=status, priority=priority))
+
+
+@strix_tool(timeout=30)
+async def update_todo(
+    ctx: RunContextWrapper,
+    todo_id: str | None = None,
+    title: str | None = None,
+    description: str | None = None,
+    priority: str | None = None,
+    status: str | None = None,
+    updates: str | None = None,
+) -> str:
+    """Update one or many todos.
+
+    Args:
+        todo_id: Single-todo target (alternative to bulk ``updates``).
+        title / description / priority / status: New values for the single
+            todo. Omit to leave unchanged.
+        updates: Bulk form — JSON list of update dicts.
+    """
+    state = adapter_from_ctx(ctx)
+    return _dump(
+        _legacy.update_todo(
+            agent_state=state,
+            todo_id=todo_id,
+            title=title,
+            description=description,
+            priority=priority,
+            status=status,
+            updates=updates,
+        ),
+    )
+
+
+@strix_tool(timeout=30)
+async def mark_todo_done(
+    ctx: RunContextWrapper,
+    todo_id: str | None = None,
+    todo_ids: str | None = None,
+) -> str:
+    """Mark one (``todo_id``) or many (``todo_ids``) todos as done."""
+    state = adapter_from_ctx(ctx)
+    return _dump(
+        _legacy.mark_todo_done(agent_state=state, todo_id=todo_id, todo_ids=todo_ids),
+    )
+
+
+@strix_tool(timeout=30)
+async def mark_todo_pending(
+    ctx: RunContextWrapper,
+    todo_id: str | None = None,
+    todo_ids: str | None = None,
+) -> str:
+    """Mark one (``todo_id``) or many (``todo_ids``) todos as pending."""
+    state = adapter_from_ctx(ctx)
+    return _dump(
+        _legacy.mark_todo_pending(
+            agent_state=state,
+            todo_id=todo_id,
+            todo_ids=todo_ids,
+        ),
+    )
+
+
+@strix_tool(timeout=30)
+async def delete_todo(
+    ctx: RunContextWrapper,
+    todo_id: str | None = None,
+    todo_ids: str | None = None,
+) -> str:
+    """Delete one (``todo_id``) or many (``todo_ids``) todos."""
+    state = adapter_from_ctx(ctx)
+    return _dump(
+        _legacy.delete_todo(agent_state=state, todo_id=todo_id, todo_ids=todo_ids),
+    )

tests/tools/test_notes_jsonl_concurrency.py

@@ -0,0 +1,79 @@
+"""C6 regression test — concurrent notes JSONL writes must produce valid JSONL.
+
+This test would fail before the C6 fix (AUDIT_R2 §1.1, applied in Phase 2.2):
+the legacy ``_append_note_event`` opened the file and called ``f.write``
+without holding ``_notes_lock``, so two threads writing simultaneously
+could interleave bytes mid-line and corrupt the JSONL.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+from collections.abc import Iterator
+from pathlib import Path
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+from strix.tools.notes.notes_actions import _append_note_event
+
+
+@pytest.fixture
+def notes_path(tmp_path: Path) -> Iterator[Path]:
+    """Point ``_get_notes_jsonl_path`` at a tmp file for the test."""
+    target = tmp_path / "notes" / "notes.jsonl"
+    target.parent.mkdir(parents=True, exist_ok=True)
+
+    with patch(
+        "strix.tools.notes.notes_actions._get_notes_jsonl_path",
+        return_value=target,
+    ):
+        yield target
+
+
+def test_concurrent_note_writes_yield_valid_jsonl(notes_path: Path) -> None:
+    """C6 fix: 50 threads x 20 events = 1000 lines, all valid JSON.
+
+    Without the lock, byte-level interleaving on the file produces
+    fragments like ``{"timesta{"timestamp"...`` that fail json.loads.
+    """
+
+    def writer(thread_idx: int) -> None:
+        for i in range(20):
+            note: dict[str, Any] = {
+                "title": f"thread-{thread_idx}-note-{i}",
+                "content": "x" * 200,  # non-trivial body to widen the race
+                "category": "general",
+            }
+            _append_note_event(
+                op="create",
+                note_id=f"t{thread_idx}-i{i}",
+                note=note,
+            )
+
+    threads = [threading.Thread(target=writer, args=(t,)) for t in range(50)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    lines = notes_path.read_text(encoding="utf-8").splitlines()
+    assert len(lines) == 1000, f"expected 1000 lines, got {len(lines)}"
+    for line in lines:
+        # raises if the line is malformed JSON
+        event = json.loads(line)
+        assert event["op"] == "create"
+        assert "note_id" in event
+
+
+def test_single_writer_still_works(notes_path: Path) -> None:
+    """Sanity: serial writes still produce a valid JSONL log."""
+    _append_note_event("create", "n1", {"title": "first"})
+    _append_note_event("update", "n1", {"title": "first updated"})
+    _append_note_event("delete", "n1")
+
+    events = [json.loads(line) for line in notes_path.read_text().splitlines()]
+    assert [e["op"] for e in events] == ["create", "update", "delete"]
+    assert all(e["note_id"] == "n1" for e in events)

tests/tools/test_sandbox_dispatch.py

@@ -0,0 +1,206 @@
+"""Phase 2.1 smoke tests for the sandbox dispatch helper."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+import httpx
+import pytest
+
+from strix.tools._sandbox_dispatch import post_to_sandbox
+
+
+@dataclass
+class _Ctx:
+    """Stand-in for ``RunContextWrapper``. Only ``.context`` is touched."""
+
+    context: dict[str, Any] = field(default_factory=dict)
+
+
+def _ok_ctx(**overrides: Any) -> _Ctx:
+    base: dict[str, Any] = {
+        "tool_server_host_port": 48081,
+        "sandbox_token": "test-bearer",
+        "agent_id": "agent-1",
+    }
+    base.update(overrides)
+    return _Ctx(context=base)
+
+
+@pytest.mark.asyncio
+async def test_missing_context_returns_error() -> None:
+    """If ``ctx.context`` isn't a dict, return error — never raise."""
+    ctx = _Ctx(context="not a dict")
+    result = await post_to_sandbox(ctx, "browser_action", {"action": "launch"})
+    assert "error" in result
+    assert "context" in result["error"].lower()
+
+
+@pytest.mark.asyncio
+async def test_missing_port_or_token_returns_error() -> None:
+    ctx = _Ctx(context={"sandbox_token": "tok"})  # no port
+    result = await post_to_sandbox(ctx, "x", {})
+    assert "error" in result
+    assert "tool server port" in result["error"].lower()
+
+
+@pytest.mark.asyncio
+async def test_successful_response_returned_as_dict(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured: dict[str, Any] = {}
+
+    async def fake_post(
+        self: httpx.AsyncClient,
+        url: str,
+        *,
+        json: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> httpx.Response:
+        captured["url"] = url
+        captured["json"] = json
+        captured["headers"] = headers
+        return httpx.Response(
+            status_code=200,
+            json={"result": "ok"},
+            request=httpx.Request("POST", url),
+        )
+
+    monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
+
+    result = await post_to_sandbox(_ok_ctx(), "terminal_execute", {"command": "ls"})
+    assert result == {"result": "ok"}
+    assert captured["url"] == "http://127.0.0.1:48081/execute"
+    assert captured["json"] == {
+        "agent_id": "agent-1",
+        "tool_name": "terminal_execute",
+        "kwargs": {"command": "ls"},
+    }
+    assert captured["headers"]["Authorization"] == "Bearer test-bearer"
+
+
+@pytest.mark.asyncio
+async def test_401_returns_auth_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_post(
+        self: httpx.AsyncClient,
+        url: str,
+        **_: Any,
+    ) -> httpx.Response:
+        return httpx.Response(
+            status_code=401,
+            text="forbidden",
+            request=httpx.Request("POST", url),
+        )
+
+    monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
+    result = await post_to_sandbox(_ok_ctx(), "x", {})
+    assert "error" in result
+    assert "authorization" in result["error"].lower()
+
+
+@pytest.mark.asyncio
+async def test_5xx_returns_http_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_post(
+        self: httpx.AsyncClient,
+        url: str,
+        **_: Any,
+    ) -> httpx.Response:
+        return httpx.Response(
+            status_code=503,
+            text="server fell over",
+            request=httpx.Request("POST", url),
+        )
+
+    monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
+    result = await post_to_sandbox(_ok_ctx(), "browser_action", {})
+    assert "error" in result
+    assert "503" in result["error"]
+
+
+@pytest.mark.asyncio
+async def test_timeout_returns_timeout_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_post(*_args: Any, **_kwargs: Any) -> httpx.Response:
+        raise httpx.ReadTimeout("read timeout")
+
+    monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
+    result = await post_to_sandbox(_ok_ctx(), "python_action", {})
+    assert "error" in result
+    assert "timed out" in result["error"].lower()
+
+
+@pytest.mark.asyncio
+async def test_connection_error_returns_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_post(*_args: Any, **_kwargs: Any) -> httpx.Response:
+        raise httpx.ConnectError("refused")
+
+    monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
+    result = await post_to_sandbox(_ok_ctx(), "x", {})
+    assert "error" in result
+    assert "connection" in result["error"].lower()
+
+
+@pytest.mark.asyncio
+async def test_response_too_large_capped(monkeypatch: pytest.MonkeyPatch) -> None:
+    """C18 (AUDIT_R3): response > 50MB returns error, doesn't OOM."""
+
+    async def fake_post(
+        self: httpx.AsyncClient,
+        url: str,
+        **_: Any,
+    ) -> httpx.Response:
+        # Construct a response with a >50MB body. Use a string trick —
+        # httpx.Response stores .content directly; we make it look huge.
+        big = b"x" * (51 * 1024 * 1024)
+        return httpx.Response(
+            status_code=200,
+            content=big,
+            request=httpx.Request("POST", url),
+        )
+
+    monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
+    result = await post_to_sandbox(_ok_ctx(), "browser_action", {})
+    assert "error" in result
+    assert "too large" in result["error"].lower()
+
+
+@pytest.mark.asyncio
+async def test_non_json_response_returns_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def fake_post(
+        self: httpx.AsyncClient,
+        url: str,
+        **_: Any,
+    ) -> httpx.Response:
+        return httpx.Response(
+            status_code=200,
+            text="hello not json",
+            request=httpx.Request("POST", url),
+        )
+
+    monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
+    result = await post_to_sandbox(_ok_ctx(), "x", {})
+    assert "error" in result
+    assert "non-json" in result["error"].lower()
+
+
+@pytest.mark.asyncio
+async def test_non_object_json_returns_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def fake_post(
+        self: httpx.AsyncClient,
+        url: str,
+        **_: Any,
+    ) -> httpx.Response:
+        return httpx.Response(
+            status_code=200,
+            json=["a", "list", "not", "an", "object"],
+            request=httpx.Request("POST", url),
+        )
+
+    monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
+    result = await post_to_sandbox(_ok_ctx(), "x", {})
+    assert "error" in result
+    assert "non-object" in result["error"].lower()

tests/tools/test_sdk_local_tools.py

@@ -0,0 +1,250 @@
+"""Phase 2.3 smoke tests for the simplest SDK-wrapped local tools.
+
+Validates the wrapping pattern (legacy implementation in, JSON string out)
+on three tool families: think (trivial), todo (in-memory + agent_state
+adapter), notes (in-memory + JSONL persistence).
+
+If this slice works end-to-end the same pattern carries the rest of the
+local tools (reporting, web_search, file_edit, finish_scan, load_skill).
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Iterator
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+from agents.tool import FunctionTool
+
+from strix.tools.notes import notes_actions as _notes_legacy
+from strix.tools.notes.notes_sdk_tools import (
+    create_note,
+    delete_note,
+    get_note,
+    list_notes,
+    update_note,
+)
+from strix.tools.thinking.thinking_sdk_tools import think
+from strix.tools.todo.todo_sdk_tools import (
+    create_todo,
+    delete_todo,
+    list_todos,
+    mark_todo_done,
+    mark_todo_pending,
+    update_todo,
+)
+
+
+@dataclass
+class _Ctx:
+    """Stand-in for ``RunContextWrapper``."""
+
+    context: dict[str, Any] = field(default_factory=dict)
+
+
+def _ctx_for(agent_id: str = "test-agent") -> _Ctx:
+    return _Ctx(context={"agent_id": agent_id})
+
+
+async def _invoke(tool: FunctionTool, ctx: _Ctx, **kwargs: Any) -> dict[str, Any]:
+    """Invoke a function tool the way the SDK would and JSON-decode the result."""
+    from agents.tool_context import ToolContext
+
+    tool_ctx = ToolContext(
+        context=ctx.context,
+        usage=None,
+        tool_name=tool.name,
+        tool_call_id="test-call-id",
+        tool_arguments=json.dumps(kwargs),
+    )
+    result = await tool.on_invoke_tool(tool_ctx, json.dumps(kwargs))
+    assert isinstance(result, str)
+    decoded = json.loads(result)
+    assert isinstance(decoded, dict)
+    return decoded
+
+
+# --- think ----------------------------------------------------------------
+
+
+def test_think_is_a_function_tool() -> None:
+    assert isinstance(think, FunctionTool)
+    assert think.name == "think"
+
+
+@pytest.mark.asyncio
+async def test_think_records_thought() -> None:
+    ctx = _ctx_for()
+    thought = "planning my next move"
+    out = await _invoke(think, ctx, thought=thought)
+    assert out["success"] is True
+    assert f"{len(thought)} characters" in out["message"]
+
+
+@pytest.mark.asyncio
+async def test_think_rejects_empty() -> None:
+    ctx = _ctx_for()
+    out = await _invoke(think, ctx, thought="   ")
+    assert out["success"] is False
+
+
+# --- todo -----------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def _isolate_todo_storage() -> None:
+    """Each test starts with an empty todo store so tests don't bleed."""
+    from strix.tools.todo import todo_actions
+
+    todo_actions._todos_storage.clear()
+
+
+def test_todo_tools_are_function_tools() -> None:
+    for tool in (
+        create_todo,
+        list_todos,
+        update_todo,
+        mark_todo_done,
+        mark_todo_pending,
+        delete_todo,
+    ):
+        assert isinstance(tool, FunctionTool)
+
+
+@pytest.mark.asyncio
+async def test_todo_lifecycle() -> None:
+    ctx = _ctx_for("agent-A")
+
+    # Create
+    created = await _invoke(create_todo, ctx, title="audit endpoint", priority="high")
+    assert created["success"] is True
+    assert created["count"] == 1
+    todo_id = created["created"][0]["todo_id"]
+
+    # List
+    listed = await _invoke(list_todos, ctx)
+    assert listed["success"] is True
+    assert any(t["todo_id"] == todo_id for t in listed["todos"])
+
+    # Update
+    updated = await _invoke(update_todo, ctx, todo_id=todo_id, status="in_progress")
+    assert updated["success"] is True
+
+    # Mark done
+    done = await _invoke(mark_todo_done, ctx, todo_id=todo_id)
+    assert done["success"] is True
+
+    # Reset to pending
+    pending = await _invoke(mark_todo_pending, ctx, todo_id=todo_id)
+    assert pending["success"] is True
+
+    # Delete
+    deleted = await _invoke(delete_todo, ctx, todo_id=todo_id)
+    assert deleted["success"] is True
+
+
+@pytest.mark.asyncio
+async def test_todos_are_per_agent_isolated() -> None:
+    """Two agents should have independent todo stores."""
+    ctx_a = _ctx_for("agent-A")
+    ctx_b = _ctx_for("agent-B")
+
+    await _invoke(create_todo, ctx_a, title="A's task")
+    await _invoke(create_todo, ctx_b, title="B's task")
+
+    list_a = await _invoke(list_todos, ctx_a)
+    list_b = await _invoke(list_todos, ctx_b)
+
+    titles_a = [t["title"] for t in list_a["todos"]]
+    titles_b = [t["title"] for t in list_b["todos"]]
+    assert titles_a == ["A's task"]
+    assert titles_b == ["B's task"]
+
+
+@pytest.mark.asyncio
+async def test_create_todo_bulk_via_json_string() -> None:
+    ctx = _ctx_for()
+    out = await _invoke(
+        create_todo,
+        ctx,
+        todos=json.dumps(
+            [
+                {"title": "t1", "priority": "high"},
+                {"title": "t2", "priority": "low"},
+            ],
+        ),
+    )
+    assert out["success"] is True
+    assert out["count"] == 2
+
+
+# --- notes ----------------------------------------------------------------
+
+
+@pytest.fixture
+def notes_run_dir(tmp_path: Path) -> Iterator[Path]:
+    """Point the legacy notes module at a fresh run dir per test."""
+    run_dir = tmp_path / "strix_runs" / "test"
+    run_dir.mkdir(parents=True)
+    _notes_legacy._notes_storage.clear()
+    _notes_legacy._loaded_notes_run_dir = None
+
+    with patch.object(_notes_legacy, "_get_run_dir", return_value=run_dir):
+        yield run_dir
+
+
+def test_notes_tools_are_function_tools() -> None:
+    for tool in (create_note, list_notes, get_note, update_note, delete_note):
+        assert isinstance(tool, FunctionTool)
+
+
+@pytest.mark.asyncio
+async def test_note_lifecycle(notes_run_dir: Path) -> None:
+    ctx = _ctx_for()
+
+    created = await _invoke(
+        create_note,
+        ctx,
+        title="SQLi at /login",
+        content="Form param `email` reflects into the WHERE clause.",
+        category="findings",
+        tags=["sqli", "auth"],
+    )
+    assert created["success"] is True
+    note_id = created["note_id"]
+
+    listed = await _invoke(list_notes, ctx, category="findings")
+    assert listed["success"] is True
+    assert listed["total_count"] == 1
+
+    fetched = await _invoke(get_note, ctx, note_id=note_id)
+    assert fetched["success"] is True
+    assert "WHERE clause" in fetched["note"]["content"]
+
+    updated = await _invoke(
+        update_note,
+        ctx,
+        note_id=note_id,
+        content="Confirmed boolean-blind SQLi.",
+    )
+    assert updated["success"] is True
+
+    deleted = await _invoke(delete_note, ctx, note_id=note_id)
+    assert deleted["success"] is True
+
+
+@pytest.mark.asyncio
+async def test_notes_jsonl_appended(notes_run_dir: Path) -> None:
+    """Verify side effect: notes.jsonl receives one event per op."""
+    ctx = _ctx_for()
+    await _invoke(create_note, ctx, title="t", content="c", category="general")
+
+    jsonl = notes_run_dir / "notes" / "notes.jsonl"
+    assert jsonl.exists()
+    events = [json.loads(line) for line in jsonl.read_text().splitlines() if line]
+    assert events[0]["op"] == "create"
+    assert events[0]["note"]["title"] == "t"