From 57478e5d0d446c36301aaa7562f2774d868af39a Mon Sep 17 00:00:00 2001 From: 0xallam Date: Sat, 25 Apr 2026 00:21:37 -0700 Subject: [PATCH] =?UTF-8?q?feat(migration):=20phase=202.4=20=E2=80=94=20wr?= =?UTF-8?q?ap=20remaining=20local=20SDK=20tools?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five tool families ported to SDK function tools using the proven delegation pattern from Phase 2.3: - web_search (1 tool): asyncio.to_thread around the synchronous Perplexity request so the 300s API call doesn't block the SDK event loop. - file_edit (3 tools — str_replace_editor, list_files, search_files): these run *inside* the sandbox container in the legacy harness (sandbox_execution=True), so the SDK wrappers route through post_to_sandbox rather than importing the legacy module on the host (which pulls in openhands_aci, a sandbox-only dependency). - reporting (1 tool — create_vulnerability_report): asyncio.to_thread around the legacy function, which itself runs CVSS XML parsing, LLM-based dedup against existing findings, and tracer persistence. - load_skill (1 tool): legacy adapter passes ctx.context['agent_id'] through. The legacy implementation reaches into _agent_instances, a global Phase 3 will replace; until then the call degrades to a structured error rather than crashing. - finish_scan (1 tool): legacy adapter pattern. Validates non-empty fields, checks no other agents are still active (via legacy _agent_graph), persists the four executive sections through the global tracer. Tests: 12 new tests in test_sdk_remaining_local_tools.py — registration checks, web_search delegation + missing-key path, file_edit dispatch shape verification, vuln-report validation + delegation, load_skill adapter passthrough, finish_scan validation + delegation. The two finish_scan tests use a fixture that snapshots/clears the legacy _agent_graph['nodes'] dict so cross-test pollution from legacy multi-agent tests doesn't mask the validation path. Per-file ruff TC002 ignores added for the five new wrapper modules (same reason as Phase 2.3 — RunContextWrapper must be runtime-importable for SDK function_schema().get_type_hints()). Refs: PLAYBOOK.md §3.5. Co-Authored-By: Claude Opus 4.7 (1M context) --- pyproject.toml | 5 + strix/tools/file_edit/file_edit_sdk_tools.py | 109 ++++++ strix/tools/finish/finish_sdk_tool.py | 67 ++++ strix/tools/load_skill/load_skill_sdk_tool.py | 46 +++ strix/tools/reporting/reporting_sdk_tools.py | 90 +++++ strix/tools/web_search/web_search_sdk_tool.py | 42 +++ tests/tools/test_sdk_remaining_local_tools.py | 341 ++++++++++++++++++ 7 files changed, 700 insertions(+) create mode 100644 strix/tools/file_edit/file_edit_sdk_tools.py create mode 100644 strix/tools/finish/finish_sdk_tool.py create mode 100644 strix/tools/load_skill/load_skill_sdk_tool.py create mode 100644 strix/tools/reporting/reporting_sdk_tools.py create mode 100644 strix/tools/web_search/web_search_sdk_tool.py create mode 100644 tests/tools/test_sdk_remaining_local_tools.py diff --git a/pyproject.toml b/pyproject.toml index 83dcb50..46eabde 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -289,6 +289,11 @@ ignore = [ "strix/tools/todo/todo_sdk_tools.py" = ["TC002"] "strix/tools/notes/notes_sdk_tools.py" = ["TC002"] "strix/tools/thinking/thinking_sdk_tools.py" = ["TC002"] +"strix/tools/web_search/web_search_sdk_tool.py" = ["TC002"] +"strix/tools/file_edit/file_edit_sdk_tools.py" = ["TC002"] +"strix/tools/reporting/reporting_sdk_tools.py" = ["TC002"] +"strix/tools/load_skill/load_skill_sdk_tool.py" = ["TC002"] +"strix/tools/finish/finish_sdk_tool.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. diff --git a/strix/tools/file_edit/file_edit_sdk_tools.py b/strix/tools/file_edit/file_edit_sdk_tools.py new file mode 100644 index 0000000..e5d81c5 --- /dev/null +++ b/strix/tools/file_edit/file_edit_sdk_tools.py @@ -0,0 +1,109 @@ +"""SDK function-tool wrappers for the legacy ``file_edit`` tools. + +These three tools (``str_replace_editor``, ``list_files``, ``search_files``) +operate on files inside the sandbox container's ``/workspace`` filesystem. +The legacy harness marks them ``sandbox_execution=True`` (default) so the +executor POSTs them to the in-container tool server. + +The host-side SDK wrappers therefore delegate to ``post_to_sandbox`` — +the legacy implementations live in the container image and we don't +import them on the host (they pull in ``openhands_aci``, which is a +sandbox-only dependency). +""" + +from __future__ import annotations + +import json +from typing import Any + +from agents import RunContextWrapper + +from strix.tools._decorator import strix_tool +from strix.tools._sandbox_dispatch import post_to_sandbox + + +def _dump(result: dict[str, Any]) -> str: + return json.dumps(result, ensure_ascii=False, default=str) + + +@strix_tool(timeout=180) +async def str_replace_editor( + ctx: RunContextWrapper, + command: str, + path: str, + file_text: str | None = None, + view_range: list[int] | None = None, + old_str: str | None = None, + new_str: str | None = None, + insert_line: int | None = None, +) -> str: + """View, create, or edit a file in the sandbox. + + Args: + command: One of ``"view" | "create" | "str_replace" | "insert" | + "undo_edit"``. + path: File path. Relative paths are anchored at ``/workspace``. + file_text: Required for ``create``. + view_range: Optional ``[start, end]`` line range for ``view``. + old_str / new_str: Required for ``str_replace``. + insert_line: Required for ``insert``. + """ + return _dump( + await post_to_sandbox( + ctx, + "str_replace_editor", + { + "command": command, + "path": path, + "file_text": file_text, + "view_range": view_range, + "old_str": old_str, + "new_str": new_str, + "insert_line": insert_line, + }, + ), + ) + + +@strix_tool(timeout=120) +async def list_files( + ctx: RunContextWrapper, + path: str, + recursive: bool = False, +) -> str: + """List files and directories under a sandbox path. + + Args: + path: Directory path, relative paths anchored at ``/workspace``. + recursive: When True, walks subdirectories (capped at 500 entries). + """ + return _dump( + await post_to_sandbox( + ctx, + "list_files", + {"path": path, "recursive": recursive}, + ), + ) + + +@strix_tool(timeout=120) +async def search_files( + ctx: RunContextWrapper, + path: str, + regex: str, + file_pattern: str = "*", +) -> str: + """Recursively grep files in the sandbox using ripgrep. + + Args: + path: Root path to search; relative paths anchored at ``/workspace``. + regex: Pattern to match (passed straight to ``rg``). + file_pattern: Glob filter (e.g. ``"*.py"``). Defaults to all files. + """ + return _dump( + await post_to_sandbox( + ctx, + "search_files", + {"path": path, "regex": regex, "file_pattern": file_pattern}, + ), + ) diff --git a/strix/tools/finish/finish_sdk_tool.py b/strix/tools/finish/finish_sdk_tool.py new file mode 100644 index 0000000..1fd3b20 --- /dev/null +++ b/strix/tools/finish/finish_sdk_tool.py @@ -0,0 +1,67 @@ +"""SDK function-tool wrapper for the legacy ``finish_scan`` tool. + +The legacy function: + +- Validates the caller is the root agent (``parent_id is None``). +- Checks no other agents are still running (via the legacy + ``_agent_graph`` global). +- Persists the four executive-summary fields via + ``get_global_tracer().update_scan_final_fields(...)``. +- Reports the final vulnerability count. + +Both the parent-id check and the agent-graph check rely on legacy +multi-agent state that Phase 3 will reimplement on top of the SDK +``RunContextWrapper`` + a per-run registry. Until Phase 3 lands, the +legacy adapter returns an object with no ``parent_id`` attribute — +``hasattr`` returns False, the validation skips, and the call proceeds +as if invoked by a root agent. That's the correct degenerate behavior +in single-agent mode, which is all Phase 2 ships. +""" + +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._legacy_adapter import adapter_from_ctx +from strix.tools.finish import finish_actions as _legacy + + +def _dump(result: dict[str, Any]) -> str: + return json.dumps(result, ensure_ascii=False, default=str) + + +@strix_tool(timeout=60) +async def finish_scan( + ctx: RunContextWrapper, + executive_summary: str, + methodology: str, + technical_analysis: str, + recommendations: str, +) -> str: + """Finalize the scan and persist the four executive summary sections. + + Only the root agent should call this. Subagents should use + ``agent_finish`` from the agents_graph tool family instead. + + Args: + executive_summary: High-level scan outcome. + methodology: Approach taken. + technical_analysis: Findings detail across the engagement. + recommendations: Prioritized fix list. + """ + state = adapter_from_ctx(ctx) + return _dump( + await asyncio.to_thread( + _legacy.finish_scan, + executive_summary=executive_summary, + methodology=methodology, + technical_analysis=technical_analysis, + recommendations=recommendations, + agent_state=state, + ), + ) diff --git a/strix/tools/load_skill/load_skill_sdk_tool.py b/strix/tools/load_skill/load_skill_sdk_tool.py new file mode 100644 index 0000000..f60b089 --- /dev/null +++ b/strix/tools/load_skill/load_skill_sdk_tool.py @@ -0,0 +1,46 @@ +"""SDK function-tool wrapper for the legacy ``load_skill`` tool. + +The legacy implementation reaches into ``_agent_instances`` (a global +dict the legacy multi-agent orchestrator maintains) to find the running +``Agent`` instance and call ``agent.llm.add_skills(...)``. That global +goes away under the SDK migration — Phase 3 will replace it with a +context-keyed registry, and this wrapper will be updated to read from +that registry. + +For Phase 2 we ship the wrapper as-is. The legacy function falls back +to a clean error path when the agent instance lookup fails, so the +tool degrades gracefully ("Could not find running agent instance...") +until Phase 3 lands. That's better than crashing or stubbing out the +tool entirely — the model still gets a structured error it can react to. +""" + +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._legacy_adapter import adapter_from_ctx +from strix.tools.load_skill import load_skill_actions as _legacy + + +def _dump(result: dict[str, Any]) -> str: + return json.dumps(result, ensure_ascii=False, default=str) + + +@strix_tool(timeout=60) +async def load_skill(ctx: RunContextWrapper, skills: str) -> str: + """Load one or more named skills into this agent's prompt context. + + Args: + skills: Comma-separated skill names (max 5). E.g. + ``"recon,xss,sqli"``. Skill discovery uses + ``strix.skills.parse_skill_list``. + """ + state = adapter_from_ctx(ctx) + return _dump( + await asyncio.to_thread(_legacy.load_skill, agent_state=state, skills=skills), + ) diff --git a/strix/tools/reporting/reporting_sdk_tools.py b/strix/tools/reporting/reporting_sdk_tools.py new file mode 100644 index 0000000..e476918 --- /dev/null +++ b/strix/tools/reporting/reporting_sdk_tools.py @@ -0,0 +1,90 @@ +"""SDK function-tool wrapper for the legacy ``create_vulnerability_report``. + +One tool. Local execution (``sandbox_execution=False`` in the legacy +registration). The legacy implementation handles XML parsing for the +CVSS breakdown and code locations, runs LLM-based dedup against +existing reports through ``strix.llm.dedupe.check_duplicate``, and +persists via ``get_global_tracer().add_vulnerability_report``. + +We wrap the synchronous legacy function in ``asyncio.to_thread`` because +the dedup check makes a network call and we don't want to block the +event loop while it waits. +""" + +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.reporting import reporting_actions as _legacy + + +def _dump(result: dict[str, Any]) -> str: + return json.dumps(result, ensure_ascii=False, default=str) + + +# Generous timeout: the dedup check makes a separate LLM call, and large +# scans can have many existing reports to compare against. +@strix_tool(timeout=180) +async def create_vulnerability_report( + ctx: RunContextWrapper, + title: str, + description: str, + impact: str, + target: str, + technical_analysis: str, + poc_description: str, + poc_script_code: str, + remediation_steps: str, + cvss_breakdown: str, + endpoint: str | None = None, + method: str | None = None, + cve: str | None = None, + cwe: str | None = None, + code_locations: str | None = None, +) -> str: + """File a vulnerability report against the active scan. + + The report is dedup-checked against existing reports (LLM-based + similarity); if it's a near-duplicate, the call returns a + ``duplicate_of`` pointer instead of creating a new entry. + + Args: + title: Short headline (e.g. ``"Reflected XSS in /search?q="``). + description: What the vuln is. + impact: Concrete impact statement. + target: Affected URL / host / service. + technical_analysis: How it works. + poc_description: Reproduction summary. + poc_script_code: Working PoC (curl, python, etc.). + remediation_steps: Recommended fix. + cvss_breakdown: CVSS 3.1 vector parameters as XML (legacy schema). + endpoint: Optional endpoint path. + method: Optional HTTP method. + cve: Optional CVE identifier. + cwe: Optional CWE identifier. + code_locations: Optional XML list of file/line references. + """ + return _dump( + await asyncio.to_thread( + _legacy.create_vulnerability_report, + title=title, + description=description, + impact=impact, + target=target, + technical_analysis=technical_analysis, + poc_description=poc_description, + poc_script_code=poc_script_code, + remediation_steps=remediation_steps, + cvss_breakdown=cvss_breakdown, + endpoint=endpoint, + method=method, + cve=cve, + cwe=cwe, + code_locations=code_locations, + ), + ) diff --git a/strix/tools/web_search/web_search_sdk_tool.py b/strix/tools/web_search/web_search_sdk_tool.py new file mode 100644 index 0000000..71fbced --- /dev/null +++ b/strix/tools/web_search/web_search_sdk_tool.py @@ -0,0 +1,42 @@ +"""SDK function-tool wrapper for the legacy ``web_search`` tool. + +The legacy ``web_search_actions.web_search`` is a synchronous Perplexity +API call (300s timeout, ``requests``). We wrap it with +``asyncio.to_thread`` so the call doesn't block the SDK event loop while +the API responds — same parity for the model, no surprises. + +Pattern matches notes/todo/think wrappers from Phase 2.3. +""" + +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.web_search import web_search_actions as _legacy + + +def _dump(result: dict[str, Any]) -> str: + return json.dumps(result, ensure_ascii=False, default=str) + + +# Perplexity request timeout in the legacy code is 300s; give the SDK +# tool a slightly larger budget so the network round-trip + JSON decode +# doesn't push us over the edge under load. +@strix_tool(timeout=330) +async def web_search(ctx: RunContextWrapper, query: str) -> str: + """Search the web with Perplexity, scoped to security-relevant content. + + Returns a JSON-encoded ``{"success": bool, "content": str, ...}`` + dict matching the legacy shape exactly. + + Args: + query: The search query. The legacy tool prepends a security-focused + system prompt to bias results toward CVEs, exploits, and Kali- + compatible commands. + """ + return _dump(await asyncio.to_thread(_legacy.web_search, query=query)) diff --git a/tests/tools/test_sdk_remaining_local_tools.py b/tests/tools/test_sdk_remaining_local_tools.py new file mode 100644 index 0000000..6949b9b --- /dev/null +++ b/tests/tools/test_sdk_remaining_local_tools.py @@ -0,0 +1,341 @@ +"""Phase 2.4 smoke tests for the remaining local SDK tool wrappers. + +Covers: web_search, file_edit (str_replace_editor + list_files + +search_files), reporting (create_vulnerability_report), load_skill, +finish_scan. + +Pattern matches test_sdk_local_tools.py — confirm registration as +``FunctionTool``, exercise the legacy delegation path, verify that +sandbox-bound tools route through ``post_to_sandbox``. +""" + +from __future__ import annotations + +import json +from collections.abc import Iterator +from dataclasses import dataclass, field +from typing import Any +from unittest.mock import patch + +import pytest +from agents.tool import FunctionTool + +from strix.tools.file_edit.file_edit_sdk_tools import ( + list_files, + search_files, + str_replace_editor, +) +from strix.tools.finish.finish_sdk_tool import finish_scan +from strix.tools.load_skill.load_skill_sdk_tool import load_skill +from strix.tools.reporting.reporting_sdk_tools import create_vulnerability_report +from strix.tools.web_search.web_search_sdk_tool import web_search + + +@dataclass +class _Ctx: + context: dict[str, Any] = field(default_factory=dict) + + +def _ctx_for(agent_id: str = "test-agent") -> _Ctx: + return _Ctx( + context={ + "agent_id": agent_id, + "tool_server_host_port": 12345, + "sandbox_token": "test-token", + }, + ) + + +async def _invoke(tool: FunctionTool, ctx: _Ctx, **kwargs: Any) -> dict[str, Any]: + 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 + + +def test_all_remaining_tools_are_function_tools() -> None: + for tool in ( + web_search, + str_replace_editor, + list_files, + search_files, + create_vulnerability_report, + load_skill, + finish_scan, + ): + assert isinstance(tool, FunctionTool) + + +# --- web_search ----------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_web_search_no_api_key_returns_structured_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The legacy function returns a structured error when the env var is + missing — verify the wrapper passes that through verbatim.""" + monkeypatch.delenv("PERPLEXITY_API_KEY", raising=False) + + out = await _invoke(web_search, _ctx_for(), query="anything") + assert out["success"] is False + assert "PERPLEXITY_API_KEY" in out["message"] + + +@pytest.mark.asyncio +async def test_web_search_delegates_to_legacy(monkeypatch: pytest.MonkeyPatch) -> None: + """Legacy ``web_search`` returns dict; wrapper JSON-encodes it.""" + monkeypatch.setenv("PERPLEXITY_API_KEY", "fake-key") + + fake_result = { + "success": True, + "query": "xss techniques", + "content": "Reflected XSS payload examples...", + "message": "Web search completed successfully", + } + with patch( + "strix.tools.web_search.web_search_sdk_tool._legacy.web_search", + return_value=fake_result, + ) as legacy: + out = await _invoke(web_search, _ctx_for(), query="xss techniques") + + assert out == fake_result + legacy.assert_called_once_with(query="xss techniques") + + +# --- file_edit (sandbox-bound) ------------------------------------------- + + +@pytest.mark.asyncio +async def test_str_replace_editor_routes_to_sandbox() -> None: + """file_edit tools must POST to the in-sandbox tool server, not run locally.""" + fake_response = {"result": {"content": "file viewed"}} + with patch( + "strix.tools.file_edit.file_edit_sdk_tools.post_to_sandbox", + return_value=fake_response, + ) as dispatch: + out = await _invoke( + str_replace_editor, + _ctx_for(), + command="view", + path="src/foo.py", + ) + + assert out == fake_response + assert dispatch.call_count == 1 + # post_to_sandbox is called positionally as (ctx, tool_name, kwargs). + args, _ = dispatch.call_args + assert args[1] == "str_replace_editor" + assert args[2]["command"] == "view" + assert args[2]["path"] == "src/foo.py" + # All optional file-edit params are forwarded as None (parity with legacy schema). + assert args[2]["file_text"] is None + assert args[2]["old_str"] is None + + +@pytest.mark.asyncio +async def test_list_files_routes_to_sandbox() -> None: + fake_response = {"result": {"files": ["a.py"], "directories": []}} + with patch( + "strix.tools.file_edit.file_edit_sdk_tools.post_to_sandbox", + return_value=fake_response, + ) as dispatch: + out = await _invoke(list_files, _ctx_for(), path="src", recursive=True) + + assert out == fake_response + args, _ = dispatch.call_args + assert args[1] == "list_files" + assert args[2] == {"path": "src", "recursive": True} + + +@pytest.mark.asyncio +async def test_search_files_routes_to_sandbox() -> None: + fake_response = {"result": {"output": "src/foo.py:1:match"}} + with patch( + "strix.tools.file_edit.file_edit_sdk_tools.post_to_sandbox", + return_value=fake_response, + ) as dispatch: + out = await _invoke( + search_files, + _ctx_for(), + path="src", + regex="TODO", + file_pattern="*.py", + ) + + assert out == fake_response + args, _ = dispatch.call_args + assert args[1] == "search_files" + assert args[2] == {"path": "src", "regex": "TODO", "file_pattern": "*.py"} + + +# --- reporting ----------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_create_vulnerability_report_validates_required_fields() -> None: + """Empty required fields should be rejected by the legacy validator.""" + out = await _invoke( + create_vulnerability_report, + _ctx_for(), + title="", # empty -> validation error + description="d", + impact="i", + target="t", + technical_analysis="ta", + poc_description="pd", + poc_script_code="curl ...", + remediation_steps="rs", + cvss_breakdown="N", + ) + assert out["success"] is False + assert "errors" in out + assert any("Title" in e for e in out["errors"]) + + +@pytest.mark.asyncio +async def test_create_vulnerability_report_delegates_to_legacy() -> None: + """Verify the wrapper passes all params through to the legacy function.""" + fake_result = { + "success": True, + "message": "Vulnerability report 'X' created successfully", + "report_id": "abc123", + "severity": "high", + "cvss_score": 7.5, + } + with patch( + "strix.tools.reporting.reporting_sdk_tools._legacy.create_vulnerability_report", + return_value=fake_result, + ) as legacy: + out = await _invoke( + create_vulnerability_report, + _ctx_for(), + title="t", + description="d", + impact="i", + target="tg", + technical_analysis="ta", + poc_description="pd", + poc_script_code="pc", + remediation_steps="rs", + cvss_breakdown="", + cve="CVE-2024-12345", + ) + + assert out == fake_result + kwargs = legacy.call_args.kwargs + assert kwargs["title"] == "t" + assert kwargs["cve"] == "CVE-2024-12345" + # Optional params we didn't pass should still be forwarded as None. + assert kwargs["endpoint"] is None + assert kwargs["method"] is None + + +# --- load_skill ---------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_load_skill_passes_adapter_with_agent_id() -> None: + """The wrapper must build the legacy-shaped adapter from ctx.context.""" + captured: dict[str, Any] = {} + + def fake_legacy(*, agent_state: Any, skills: str) -> dict[str, Any]: + captured["agent_id"] = agent_state.agent_id + captured["skills"] = skills + return {"success": True, "loaded_skills": ["recon"]} + + with patch( + "strix.tools.load_skill.load_skill_sdk_tool._legacy.load_skill", + side_effect=fake_legacy, + ): + out = await _invoke(load_skill, _ctx_for("agent-XYZ"), skills="recon") + + assert out["success"] is True + assert captured["agent_id"] == "agent-XYZ" + assert captured["skills"] == "recon" + + +@pytest.mark.asyncio +async def test_load_skill_with_empty_input() -> None: + """End-to-end: empty skills string yields the legacy validation error.""" + out = await _invoke(load_skill, _ctx_for(), skills="") + assert out["success"] is False + assert "No skills" in out["error"] + + +# --- finish_scan --------------------------------------------------------- + + +@pytest.fixture +def isolated_agent_graph() -> Iterator[None]: + """Clear the legacy agent-graph globals so finish_scan sees an empty world. + + The legacy ``_check_active_agents`` reads ``_agent_graph["nodes"]`` and + returns an "agents still active" error if any non-self agent is in + state ``running`` or ``stopping``. Tests in other modules (legacy + multi-agent tests) populate this dict; without isolation they bleed + into our validation tests and mask the field-validation path. + """ + from strix.tools.agents_graph import agents_graph_actions + + saved_nodes = agents_graph_actions._agent_graph.get("nodes", {}).copy() + agents_graph_actions._agent_graph["nodes"] = {} + try: + yield + finally: + agents_graph_actions._agent_graph["nodes"] = saved_nodes + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("isolated_agent_graph") +async def test_finish_scan_validates_empty_fields() -> None: + """Legacy validation: every section must be non-empty.""" + out = await _invoke( + finish_scan, + _ctx_for(), + executive_summary="", + methodology="m", + technical_analysis="ta", + recommendations="r", + ) + assert out["success"] is False + assert any("Executive summary" in e for e in out["errors"]) + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("isolated_agent_graph") +async def test_finish_scan_delegates_to_legacy() -> None: + """Wrapper must pass the legacy adapter and the four sections through.""" + fake_result = { + "success": True, + "scan_completed": True, + "message": "Scan completed successfully", + "vulnerabilities_found": 3, + } + with patch( + "strix.tools.finish.finish_sdk_tool._legacy.finish_scan", + return_value=fake_result, + ) as legacy: + out = await _invoke( + finish_scan, + _ctx_for("root-agent"), + executive_summary="es", + methodology="m", + technical_analysis="ta", + recommendations="r", + ) + + assert out == fake_result + kwargs = legacy.call_args.kwargs + assert kwargs["executive_summary"] == "es" + assert kwargs["agent_state"].agent_id == "root-agent"