dc9b9f5f9c
Cleanup pass after the migration: #1 Inline ``*_actions.py`` into wrapper ``tool[s].py`` for the non-sandbox tools (think, todo, notes, reporting, web_search, finish_scan). One file per tool family now. Helpers + public function bodies live alongside the ``@strix_tool``-decorated wrappers that call them. For notes, the sync helpers are renamed to ``_create_note_impl`` / ``_list_notes_impl`` / etc. so the public names ``create_note`` / ``list_notes`` / etc. can be the FunctionTool instances the agent factory imports. ``append_note_content`` (used by the agents-graph wiki-update hook) calls the impl helpers directly. #2 Delete ``strix/tools/_state_adapter.py``. The ``AgentStateAdapter`` shim only existed to feed legacy ``*_actions.py`` functions a ``state.agent_id`` they could read. With the actions inlined, the wrappers read ``ctx.context['agent_id']`` directly. #3 Strip ``strix/tools/registry.py`` from ~250 LOC to ~110. Deleted: XML schema loading, ``_parse_param_schema``, ``get_tools_prompt``, ``get_tool_param_schema``, ``needs_agent_state``, ``should_execute_in_sandbox``, ``validate_tool_availability`` — all for the host-side legacy dispatcher path. Kept the ``register_tool`` decorator (sandbox side), ``get_tool_by_name``, ``get_tool_names``, ``tools`` list, ``clear_registry``. The Jinja prompt template's ``{{ get_tools_prompt() }}`` injection is dropped — the SDK auto-generates tool descriptions from function signatures, so the legacy XML tool block was redundant and stale. #4 Delete every ``*_actions_schema.xml`` (12 files). They were read by the now-removed ``_load_xml_schema`` to build the legacy prompt's tool descriptions. No consumer remains. Side fixes: - ``reporting_renderer.py`` updated to import ``_parse_*_xml`` from the new location with leading underscore. - ``test_local_tools.py``, ``test_notes_jsonl_concurrency.py``, ``test_notes_wiki.py`` updated to point at the new module paths and call the ``_*_impl`` sync helpers. Tests: 279/279 passing. ~1500 LOC of action files moved into the tool wrappers; ~140 LOC of registry boilerplate removed; ~400 lines of dead XML deleted. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
103 lines
3.3 KiB
Python
103 lines
3.3 KiB
Python
"""``finish_scan`` — root-agent termination + executive report persistence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
from agents import RunContextWrapper
|
|
|
|
from strix.tools._decorator import strix_tool
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _do_finish(
|
|
*,
|
|
parent_id: str | None,
|
|
executive_summary: str,
|
|
methodology: str,
|
|
technical_analysis: str,
|
|
recommendations: str,
|
|
) -> dict[str, Any]:
|
|
if parent_id is not None:
|
|
return {
|
|
"success": False,
|
|
"error": "finish_scan_wrong_agent",
|
|
"message": "This tool can only be used by the root/main agent",
|
|
"suggestion": "If you are a subagent, use agent_finish instead",
|
|
}
|
|
|
|
errors: list[str] = []
|
|
if not executive_summary.strip():
|
|
errors.append("Executive summary cannot be empty")
|
|
if not methodology.strip():
|
|
errors.append("Methodology cannot be empty")
|
|
if not technical_analysis.strip():
|
|
errors.append("Technical analysis cannot be empty")
|
|
if not recommendations.strip():
|
|
errors.append("Recommendations cannot be empty")
|
|
if errors:
|
|
return {"success": False, "message": "Validation failed", "errors": errors}
|
|
|
|
try:
|
|
from strix.telemetry.tracer import get_global_tracer
|
|
|
|
tracer = get_global_tracer()
|
|
if tracer is None:
|
|
logger.warning("No global tracer; scan results not persisted")
|
|
return {
|
|
"success": True,
|
|
"scan_completed": True,
|
|
"message": "Scan completed (not persisted)",
|
|
"warning": "Results could not be persisted - tracer unavailable",
|
|
}
|
|
tracer.update_scan_final_fields(
|
|
executive_summary=executive_summary.strip(),
|
|
methodology=methodology.strip(),
|
|
technical_analysis=technical_analysis.strip(),
|
|
recommendations=recommendations.strip(),
|
|
)
|
|
return {
|
|
"success": True,
|
|
"scan_completed": True,
|
|
"message": "Scan completed successfully",
|
|
"vulnerabilities_found": len(tracer.vulnerability_reports),
|
|
}
|
|
except (ImportError, AttributeError) as e:
|
|
return {"success": False, "message": f"Failed to complete scan: {e!s}"}
|
|
|
|
|
|
@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 must use
|
|
``agent_finish`` (from the multi-agent graph tools) instead.
|
|
|
|
Args:
|
|
executive_summary: High-level scan outcome.
|
|
methodology: Approach taken.
|
|
technical_analysis: Findings detail across the engagement.
|
|
recommendations: Prioritized fix list.
|
|
"""
|
|
inner = ctx.context if isinstance(ctx.context, dict) else {}
|
|
result = await asyncio.to_thread(
|
|
_do_finish,
|
|
parent_id=inner.get("parent_id"),
|
|
executive_summary=executive_summary,
|
|
methodology=methodology,
|
|
technical_analysis=technical_analysis,
|
|
recommendations=recommendations,
|
|
)
|
|
return json.dumps(result, ensure_ascii=False, default=str)
|