diff --git a/pyproject.toml b/pyproject.toml index 9dbcaf9..83dcb50 100644 --- a/pyproject.toml +++ b/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. diff --git a/strix/tools/_legacy_adapter.py b/strix/tools/_legacy_adapter.py new file mode 100644 index 0000000..2218844 --- /dev/null +++ b/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)) diff --git a/strix/tools/_sandbox_dispatch.py b/strix/tools/_sandbox_dispatch.py new file mode 100644 index 0000000..d44a060 --- /dev/null +++ b/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": }``. + On any failure: ``{"error": ""}``. + + 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 diff --git a/strix/tools/notes/notes_actions.py b/strix/tools/notes/notes_actions.py index 77cb833..accea01 100644 --- a/strix/tools/notes/notes_actions.py +++ b/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") diff --git a/strix/tools/notes/notes_sdk_tools.py b/strix/tools/notes/notes_sdk_tools.py new file mode 100644 index 0000000..4969e37 --- /dev/null +++ b/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/.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) diff --git a/strix/tools/thinking/thinking_sdk_tools.py b/strix/tools/thinking/thinking_sdk_tools.py new file mode 100644 index 0000000..fb2889c --- /dev/null +++ b/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) diff --git a/strix/tools/todo/todo_sdk_tools.py b/strix/tools/todo/todo_sdk_tools.py new file mode 100644 index 0000000..42f7846 --- /dev/null +++ b/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), + ) diff --git a/tests/tools/test_notes_jsonl_concurrency.py b/tests/tools/test_notes_jsonl_concurrency.py new file mode 100644 index 0000000..6629eaf --- /dev/null +++ b/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) diff --git a/tests/tools/test_sandbox_dispatch.py b/tests/tools/test_sandbox_dispatch.py new file mode 100644 index 0000000..f9d175a --- /dev/null +++ b/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() diff --git a/tests/tools/test_sdk_local_tools.py b/tests/tools/test_sdk_local_tools.py new file mode 100644 index 0000000..15892a3 --- /dev/null +++ b/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"