refactor: nuke gratuitous XML serialization + delete argument_parser

Argument parser:
- Delete ``strix/tools/argument_parser.py`` and its tests. The SDK
  validates and types tool arguments via Pydantic before they hit our
  wrappers, and the in-container tool server receives JSON-typed
  kwargs over the wire. The string-coercion belt-and-suspenders is no
  longer pulling its weight.

XML → JSON / typed structures:
- ``create_vulnerability_report``: ``cvss_breakdown`` is now a
  ``dict[str, str]`` of the 8 metrics; ``code_locations`` is a
  ``list[dict]``. No more XML parsing in the tool or the renderer.
- ``check_duplicate``: the dedup judge now emits a single JSON object
  instead of an ``<dedupe_result>`` block. Strict JSON parser handles
  optional code-fence wrappers.
- ``agent_finish``: completion report posted to the parent inbox is a
  JSON object (``kind``, ``from``, ``agent_id``, ``success``,
  ``summary``, ``findings``, ``recommendations``) rather than a
  hand-rolled ``<agent_completion_report>`` XML envelope.
- ``create_agent``: identity preamble + inherited-context markers are
  plain bracketed labels rather than ``<agent_delegation>`` /
  ``<inherited_context_from_parent>`` envelopes.
- ``inject_messages_filter``: peer messages get a
  ``[Message from agent <id> | type=... | priority=...]`` header line
  instead of an ``<inter_agent_message>`` envelope.
- Crash + system-warning messages: bracketed labels, no XML.
- System prompt: the inter-agent block now describes the new header
  format and drops the "never echo XML envelope" rule.
- ``strix/llm/utils.py``: deleted. ``clean_content`` collapsed into a
  one-line blank-line normalizer in the agent-message renderer (the
  XML envelope scrub had nothing left to scrub).

Tests updated to match the new shapes.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam
2026-04-25 12:31:07 -07:00
parent 369fa56148
commit f08ad2a634
17 changed files with 219 additions and 667 deletions
+2 -3
View File
@@ -16,9 +16,8 @@ CLI OUTPUT:
- NEVER use "Strix" or any identifiable names/markers in HTTP requests, payloads, user-agents, or any inputs
INTER-AGENT MESSAGES:
- NEVER echo inter_agent_message or agent_completion_report blocks that are sent to you in your output.
- Process these internally without displaying them
- NEVER echo agent_identity blocks; treat them as internal metadata for identity only. Do not include them in outputs or tool calls.
- Messages from other agents arrive prefixed with a header like `[Message from agent <name> | type=... | priority=...]`. Treat them as internal context — never repeat them verbatim in your own output.
- Treat agent identity / inherited-context preambles as internal metadata; do not echo them in outputs or tool calls.
- Minimize inter-agent messaging: only message when essential for coordination or assistance; avoid routine status updates; batch non-urgent information; prefer parent/child completion flows and shared artifacts over messaging
{% if interactive %}
@@ -1,3 +1,4 @@
import re
from functools import cache
from typing import Any, ClassVar
@@ -11,6 +12,9 @@ from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
_BLANK_LINE_RUNS = re.compile(r"\n\s*\n")
_HEADER_STYLES = [
("###### ", 7, "bold #4ade80"),
("##### ", 6, "bold #22c55e"),
@@ -180,11 +184,7 @@ class AgentMessageRenderer(BaseToolRenderer):
def render_simple(cls, content: str) -> Text:
if not content:
return Text()
from strix.llm.utils import clean_content
cleaned = clean_content(content)
cleaned = _BLANK_LINE_RUNS.sub("\n\n", content).strip()
if not cleaned:
return Text()
return _apply_markdown_styles(cleaned)
@@ -6,17 +6,22 @@ from pygments.styles import get_style_by_name
from rich.text import Text
from textual.widgets import Static
from strix.tools.reporting.tool import (
_parse_code_locations_xml as parse_code_locations_xml,
)
from strix.tools.reporting.tool import (
_parse_cvss_xml as parse_cvss_xml,
)
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
def _coerce_dict(value: Any) -> dict[str, Any]:
if isinstance(value, dict):
return value
return {}
def _coerce_list_of_dicts(value: Any) -> list[dict[str, Any]]:
if isinstance(value, list):
return [item for item in value if isinstance(item, dict)]
return []
@cache
def _get_style_colors() -> dict[Any, str]:
style = get_style_by_name("native")
@@ -94,8 +99,8 @@ class CreateVulnerabilityReportRenderer(BaseToolRenderer):
poc_script_code = args.get("poc_script_code", "")
remediation_steps = args.get("remediation_steps", "")
cvss_breakdown_xml = args.get("cvss_breakdown", "")
code_locations_xml = args.get("code_locations", "")
cvss_breakdown = _coerce_dict(args.get("cvss_breakdown"))
code_locations = _coerce_list_of_dicts(args.get("code_locations"))
endpoint = args.get("endpoint", "")
method = args.get("method", "")
@@ -154,8 +159,7 @@ class CreateVulnerabilityReportRenderer(BaseToolRenderer):
text.append("CWE: ", style=FIELD_STYLE)
text.append(cwe)
parsed_cvss = parse_cvss_xml(cvss_breakdown_xml) if cvss_breakdown_xml else None
if parsed_cvss:
if cvss_breakdown:
text.append("\n\n")
cvss_parts = []
for key, prefix in [
@@ -168,7 +172,7 @@ class CreateVulnerabilityReportRenderer(BaseToolRenderer):
("integrity", "I"),
("availability", "A"),
]:
val = parsed_cvss.get(key)
val = cvss_breakdown.get(key)
if val:
cvss_parts.append(f"{prefix}:{val}")
text.append("CVSS Vector: ", style=FIELD_STYLE)
@@ -192,13 +196,10 @@ class CreateVulnerabilityReportRenderer(BaseToolRenderer):
text.append("\n")
text.append(technical_analysis)
parsed_locations = (
parse_code_locations_xml(code_locations_xml) if code_locations_xml else None
)
if parsed_locations:
if code_locations:
text.append("\n\n")
text.append("Code Locations", style=FIELD_STYLE)
for i, loc in enumerate(parsed_locations):
for i, loc in enumerate(code_locations):
text.append("\n\n")
text.append(f" Location {i + 1}: ", style=DIM_STYLE)
text.append(loc.get("file", "unknown"), style=FILE_STYLE)
+39 -51
View File
@@ -1,6 +1,5 @@
import json
import logging
import re
from typing import Any
import litellm
@@ -49,30 +48,30 @@ FIELDS TO ANALYZE:
- poc_description: How it's exploited
- impact: What damage it can cause
YOU MUST RESPOND WITH EXACTLY THIS XML FORMAT AND NOTHING ELSE:
Respond with a single JSON object and nothing else:
<dedupe_result>
<is_duplicate>true</is_duplicate>
<duplicate_id>vuln-0001</duplicate_id>
<confidence>0.95</confidence>
<reason>Both reports describe SQL injection in /api/login via the username parameter</reason>
</dedupe_result>
{
"is_duplicate": true,
"duplicate_id": "vuln-0001",
"confidence": 0.95,
"reason": "Both reports describe SQL injection in /api/login via the username parameter"
}
OR if not a duplicate:
Or, if not a duplicate:
<dedupe_result>
<is_duplicate>false</is_duplicate>
<duplicate_id></duplicate_id>
<confidence>0.90</confidence>
<reason>Different endpoints: candidate is /api/search, existing is /api/login</reason>
</dedupe_result>
{
"is_duplicate": false,
"duplicate_id": "",
"confidence": 0.90,
"reason": "Different endpoints: candidate is /api/search, existing is /api/login"
}
RULES:
- is_duplicate MUST be exactly "true" or "false" (lowercase)
- duplicate_id MUST be the exact ID from existing reports or empty if not duplicate
- confidence MUST be a decimal (your confidence level in the decision)
- reason MUST be a specific explanation mentioning endpoint/parameter/root cause
- DO NOT include any text outside the <dedupe_result> tags"""
Rules:
- ``is_duplicate`` is a boolean.
- ``duplicate_id`` is the exact id from existing reports, or "" if not a duplicate.
- ``confidence`` is a number between 0 and 1.
- ``reason`` is a specific explanation mentioning endpoint/parameter/root cause.
- Output ONLY the JSON object — no surrounding prose, no code fences."""
def _prepare_report_for_comparison(report: dict[str, Any]) -> dict[str, Any]:
@@ -99,42 +98,31 @@ def _prepare_report_for_comparison(report: dict[str, Any]) -> dict[str, Any]:
return cleaned
def _extract_xml_field(content: str, field: str) -> str:
pattern = rf"<{field}>(.*?)</{field}>"
match = re.search(pattern, content, re.DOTALL | re.IGNORECASE)
if match:
return match.group(1).strip()
return ""
def _parse_dedupe_response(content: str) -> dict[str, Any]:
result_match = re.search(
r"<dedupe_result>(.*?)</dedupe_result>", content, re.DOTALL | re.IGNORECASE
)
if not result_match:
logger.warning(f"No <dedupe_result> block found in response: {content[:500]}")
raise ValueError("No <dedupe_result> block found in response")
result_content = result_match.group(1)
is_duplicate_str = _extract_xml_field(result_content, "is_duplicate")
duplicate_id = _extract_xml_field(result_content, "duplicate_id")
confidence_str = _extract_xml_field(result_content, "confidence")
reason = _extract_xml_field(result_content, "reason")
is_duplicate = is_duplicate_str.lower() == "true"
text = content.strip()
if text.startswith("```"):
text = text.strip("`")
if text.lower().startswith("json"):
text = text[4:]
text = text.strip()
start = text.find("{")
end = text.rfind("}")
if start == -1 or end == -1 or end <= start:
raise ValueError(f"No JSON object found in dedupe response: {content[:500]}")
parsed = json.loads(text[start : end + 1])
duplicate_id = str(parsed.get("duplicate_id") or "")[:64]
reason = str(parsed.get("reason") or "")[:500]
try:
confidence = float(confidence_str) if confidence_str else 0.0
except ValueError:
confidence = float(parsed.get("confidence", 0.0))
except (TypeError, ValueError):
confidence = 0.0
return {
"is_duplicate": is_duplicate,
"duplicate_id": duplicate_id[:64] if duplicate_id else "",
"is_duplicate": bool(parsed.get("is_duplicate", False)),
"duplicate_id": duplicate_id,
"confidence": confidence,
"reason": reason[:500] if reason else "",
"reason": reason,
}
@@ -165,7 +153,7 @@ def check_duplicate(
"content": (
f"Compare this candidate vulnerability against existing reports:\n\n"
f"{json.dumps(comparison_data, indent=2)}\n\n"
f"Respond with ONLY the <dedupe_result> XML block."
f"Respond with ONLY the JSON object described in the system prompt."
),
},
]
-26
View File
@@ -1,26 +0,0 @@
"""Helpers for the TUI message renderer.
The model occasionally echoes inter-agent XML envelopes in plain text
despite the system prompt's "don't echo" rule, so :func:`clean_content`
strips them defensively before display.
"""
import re
_HIDDEN_XML_PATTERNS = [
re.compile(r"<inter_agent_message>.*?</inter_agent_message>", re.DOTALL | re.IGNORECASE),
re.compile(
r"<agent_completion_report>.*?</agent_completion_report>",
re.DOTALL | re.IGNORECASE,
),
]
_BLANK_LINE_RUNS = re.compile(r"\n\s*\n")
def clean_content(content: str) -> str:
if not content:
return ""
for pattern in _HIDDEN_XML_PATTERNS:
content = pattern.sub("", content)
return _BLANK_LINE_RUNS.sub("\n\n", content).strip()
+7 -12
View File
@@ -23,11 +23,9 @@ logger = logging.getLogger(__name__)
async def inject_messages_filter(data: CallModelData) -> ModelInputData:
"""Drain bus inbox and append messages as user-role items before the LLM call.
Each drained message is wrapped in an ``<inter_agent_message>`` XML envelope
so the system prompt's rules around inter-agent communication apply.
Messages from the literal sender ``"user"`` (a real human via TUI)
skip the XML wrap and are added as plain user messages.
Messages from peer agents are formatted with a labeled header so the
receiving model can attribute them. Messages from the literal sender
``"user"`` (a real human via TUI) are added as plain user messages.
Any exception inside the filter — malformed message dict, bug in
``bus.drain``, etc. — is caught and the original ``data.model_data``
@@ -51,16 +49,13 @@ async def inject_messages_filter(data: CallModelData) -> ModelInputData:
if sender == "user":
new_input.append({"role": "user", "content": content})
else:
msg_type = msg.get("type", "info")
priority = msg.get("priority", "normal")
header = f"[Message from agent {sender} | type={msg_type} | priority={priority}]"
new_input.append(
{
"role": "user",
"content": (
f"<inter_agent_message from='{sender}' "
f"type='{msg.get('type', 'info')}' "
f"priority='{msg.get('priority', 'normal')}'>"
f"{content}"
f"</inter_agent_message>"
),
"content": f"{header}\n{content}",
}
)
return ModelInputData(
+9 -12
View File
@@ -26,9 +26,9 @@ class StrixOrchestrationHooks(RunHooks[Any]):
the agent doesn't fire tools before Caido and the tool server
are ready.
4. Subagent crash detection: if ``on_agent_end`` fires without
``agent_finish_called`` being set, posts a synthetic
``<agent_crash>`` message to the parent's inbox so the parent
learns on its next turn instead of waiting forever.
``agent_finish_called`` being set, posts a crash message to the
parent's inbox so the parent learns on its next turn instead of
waiting forever.
"""
async def on_llm_start(
@@ -51,8 +51,8 @@ class StrixOrchestrationHooks(RunHooks[Any]):
{
"role": "user",
"content": (
"<system_warning>You are at 85% of your iteration "
"budget. Begin consolidating findings.</system_warning>"
"[System warning] You are at 85% of your iteration "
"budget. Begin consolidating findings."
),
}
)
@@ -61,9 +61,8 @@ class StrixOrchestrationHooks(RunHooks[Any]):
{
"role": "user",
"content": (
"<system_warning>You have 3 iterations left. Your "
"[System warning] You have 3 iterations left. Your "
"next tool call MUST be the finish tool."
"</system_warning>"
),
}
)
@@ -148,11 +147,9 @@ class StrixOrchestrationHooks(RunHooks[Any]):
{
"from": me,
"content": (
f"<agent_crash agent_id='{me}' "
f"name='{bus.names.get(me, me)}'>"
"Agent terminated without calling agent_finish. "
"Stop waiting on this child."
"</agent_crash>"
f"[Agent crash] {bus.names.get(me, me)} ({me}) "
f"terminated without calling agent_finish. "
f"Stop waiting on this child."
),
"type": "crash",
},
+1 -3
View File
@@ -69,7 +69,6 @@ class ToolExecutionResponse(BaseModel):
async def _run_tool(agent_id: str, tool_name: str, kwargs: dict[str, Any]) -> Any:
from strix.tools.argument_parser import convert_arguments
from strix.tools.context import set_current_agent_id
from strix.tools.registry import get_tool_by_name
@@ -79,8 +78,7 @@ async def _run_tool(agent_id: str, tool_name: str, kwargs: dict[str, Any]) -> An
if not tool_func:
raise ValueError(f"Tool '{tool_name}' not found")
converted_kwargs = convert_arguments(tool_func, kwargs)
return await asyncio.to_thread(tool_func, **converted_kwargs)
return await asyncio.to_thread(tool_func, **kwargs)
@app.post("/execute", response_model=ToolExecutionResponse)
+18 -22
View File
@@ -371,33 +371,29 @@ async def create_agent(
await bus.register(child_id, name, parent_id)
# Build the child's input. Identity injection mirrors the legacy
# <agent_delegation> envelope so the child's system prompt's existing
# rules around self-identity still apply.
parent_history = inner.get("parent_input_items") if inherit_context else None
initial_input: list[TResponseInputItem] = []
if parent_history:
initial_input.append(
{
"role": "user",
"content": "<inherited_context_from_parent>",
"content": "[Inherited context from parent — read-only history]",
}
)
initial_input.extend(parent_history)
initial_input.append(
{
"role": "user",
"content": "</inherited_context_from_parent>",
"content": "[End of inherited context]",
}
)
initial_input.append(
{
"role": "user",
"content": (
f"<agent_delegation>\n"
f"You are agent {name} ({child_id}). Parent is {parent_id}.\n"
f"Maintain self-identity. Use agent_finish when complete.\n"
f"</agent_delegation>"
f"You are agent {name} ({child_id}); your parent is {parent_id}. "
f"Maintain your own identity. Call agent_finish when your task "
f"is complete."
),
}
)
@@ -471,8 +467,8 @@ async def agent_finish(
this tool refuses to run for root agents. Calling this:
1. Marks the subagent as ``completed``.
2. Posts a structured ``<agent_completion_report>`` to the
parent's inbox (when ``report_to_parent`` is true).
2. Posts a structured completion report to the parent's inbox
(when ``report_to_parent`` is true).
3. Stops this subagent's execution.
**Vulnerability findings must already be filed via
@@ -522,19 +518,19 @@ async def agent_finish(
parent_notified = False
if report_to_parent:
findings_xml = "\n".join(f" <finding>{f}</finding>" for f in (findings or []))
rec_xml = "\n".join(
f" <recommendation>{r}</recommendation>" for r in (final_recommendations or [])
)
async with bus._lock:
agent_name = bus.names.get(me, me)
report = (
f"<agent_completion_report from='{agent_name}' agent_id='{me}' "
f"success='{success}'>\n"
f" <summary>{result_summary}</summary>\n"
f" <findings>\n{findings_xml}\n </findings>\n"
f" <recommendations>\n{rec_xml}\n </recommendations>\n"
f"</agent_completion_report>"
report = json.dumps(
{
"kind": "agent_completion_report",
"from": agent_name,
"agent_id": me,
"success": success,
"summary": result_summary,
"findings": list(findings or []),
"recommendations": list(final_recommendations or []),
},
ensure_ascii=False,
)
await bus.send(
parent_id,
-121
View File
@@ -1,121 +0,0 @@
import contextlib
import inspect
import json
import types
from collections.abc import Callable
from typing import Any, Union, get_args, get_origin
class ArgumentConversionError(Exception):
def __init__(self, message: str, param_name: str | None = None) -> None:
self.param_name = param_name
super().__init__(message)
def convert_arguments(func: Callable[..., Any], kwargs: dict[str, Any]) -> dict[str, Any]:
try:
sig = inspect.signature(func)
converted = {}
for param_name, value in kwargs.items():
if param_name not in sig.parameters:
converted[param_name] = value
continue
param = sig.parameters[param_name]
param_type = param.annotation
if param_type == inspect.Parameter.empty or value is None:
converted[param_name] = value
continue
if not isinstance(value, str):
converted[param_name] = value
continue
try:
converted[param_name] = convert_string_to_type(value, param_type)
except (ValueError, TypeError, json.JSONDecodeError) as e:
raise ArgumentConversionError(
f"Failed to convert argument '{param_name}' to type {param_type}: {e}",
param_name=param_name,
) from e
except (ValueError, TypeError, AttributeError) as e:
raise ArgumentConversionError(f"Failed to process function arguments: {e}") from e
return converted
def convert_string_to_type(value: str, param_type: Any) -> Any:
origin = get_origin(param_type)
if origin is Union or isinstance(param_type, types.UnionType):
args = get_args(param_type)
for arg_type in args:
if arg_type is not type(None):
with contextlib.suppress(ValueError, TypeError, json.JSONDecodeError):
return convert_string_to_type(value, arg_type)
return value
if hasattr(param_type, "__args__"):
args = getattr(param_type, "__args__", ())
if len(args) == 2 and type(None) in args:
non_none_type = args[0] if args[1] is type(None) else args[1]
with contextlib.suppress(ValueError, TypeError, json.JSONDecodeError):
return convert_string_to_type(value, non_none_type)
return value
return _convert_basic_types(value, param_type, origin)
def _convert_basic_types(value: str, param_type: Any, origin: Any = None) -> Any:
basic_type_converters: dict[Any, Callable[[str], Any]] = {
int: int,
float: float,
bool: _convert_to_bool,
str: str,
}
if param_type in basic_type_converters:
return basic_type_converters[param_type](value)
if list in (origin, param_type):
return _convert_to_list(value)
if dict in (origin, param_type):
return _convert_to_dict(value)
with contextlib.suppress(json.JSONDecodeError):
return json.loads(value)
return value
def _convert_to_bool(value: str) -> bool:
if value.lower() in ("true", "1", "yes", "on"):
return True
if value.lower() in ("false", "0", "no", "off"):
return False
return bool(value)
def _convert_to_list(value: str) -> list[Any]:
try:
parsed = json.loads(value)
if isinstance(parsed, list):
return parsed
except json.JSONDecodeError:
if "," in value:
return [item.strip() for item in value.split(",")]
return [value]
else:
return [parsed]
def _convert_to_dict(value: str) -> dict[str, Any]:
try:
parsed = json.loads(value)
if isinstance(parsed, dict):
return parsed
except json.JSONDecodeError:
return {}
else:
return {}
+95 -102
View File
@@ -1,15 +1,8 @@
"""``create_vulnerability_report`` — file a vuln finding with dedup + CVSS.
Validates required fields, parses the CVSS-3.1 XML breakdown into a
score, runs LLM-based dedup against existing reports through
``strix.llm.dedupe.check_duplicate``, and persists via the global
:class:`strix.telemetry.tracer.Tracer` instance.
"""
"""``create_vulnerability_report`` — file a vuln finding with dedup + CVSS."""
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
import re
@@ -24,63 +17,29 @@ from strix.tools._decorator import strix_tool
logger = logging.getLogger(__name__)
_CVSS_FIELDS = (
"attack_vector",
"attack_complexity",
"privileges_required",
"user_interaction",
"scope",
"confidentiality",
"integrity",
"availability",
_CVSS_VALID = {
"attack_vector": ["N", "A", "L", "P"],
"attack_complexity": ["L", "H"],
"privileges_required": ["N", "L", "H"],
"user_interaction": ["N", "R"],
"scope": ["U", "C"],
"confidentiality": ["N", "L", "H"],
"integrity": ["N", "L", "H"],
"availability": ["N", "L", "H"],
}
_CODE_LOCATION_FIELDS = (
"file",
"start_line",
"end_line",
"snippet",
"label",
"fix_before",
"fix_after",
)
def _parse_cvss_xml(xml_str: str) -> dict[str, str] | None:
if not xml_str or not xml_str.strip():
return None
result: dict[str, str] = {}
for field in _CVSS_FIELDS:
match = re.search(rf"<{field}>(.*?)</{field}>", xml_str, re.DOTALL)
if match:
result[field] = match.group(1).strip()
return result if result else None
def _parse_code_locations_xml(xml_str: str) -> list[dict[str, Any]] | None:
if not xml_str or not xml_str.strip():
return None
locations: list[dict[str, Any]] = []
for loc_match in re.finditer(r"<location>(.*?)</location>", xml_str, re.DOTALL):
loc: dict[str, Any] = {}
loc_content = loc_match.group(1)
for field in (
"file",
"start_line",
"end_line",
"snippet",
"label",
"fix_before",
"fix_after",
):
field_match = re.search(rf"<{field}>(.*?)</{field}>", loc_content, re.DOTALL)
if field_match:
raw = field_match.group(1)
value = (
raw.strip("\n")
if field in ("snippet", "fix_before", "fix_after")
else raw.strip()
)
if field in ("start_line", "end_line"):
with contextlib.suppress(ValueError, TypeError):
loc[field] = int(value)
elif value:
loc[field] = value
if loc.get("file") and loc.get("start_line") is not None:
locations.append(loc)
return locations if locations else None
def _validate_file_path(path: str) -> str | None:
if not path or not path.strip():
return "file path cannot be empty"
@@ -92,6 +51,36 @@ def _validate_file_path(path: str) -> str | None:
return None
def _normalize_code_locations(
raw: list[dict[str, Any]] | None,
) -> list[dict[str, Any]] | None:
if not raw:
return None
cleaned: list[dict[str, Any]] = []
for loc in raw:
normalized: dict[str, Any] = {}
for field in _CODE_LOCATION_FIELDS:
if field not in loc or loc[field] is None:
continue
value = loc[field]
if field in ("start_line", "end_line"):
try:
normalized[field] = int(value)
except (TypeError, ValueError):
continue
else:
text = (
str(value).strip("\n")
if field in ("snippet", "fix_before", "fix_after")
else str(value).strip()
)
if text:
normalized[field] = text
if normalized.get("file") and normalized.get("start_line") is not None:
cleaned.append(normalized)
return cleaned or None
def _validate_code_locations(locations: list[dict[str, Any]]) -> list[str]:
errors: list[str] = []
for i, loc in enumerate(locations):
@@ -133,15 +122,15 @@ def _validate_cwe(cwe: str) -> str | None:
return None
def _calculate_cvss(**kwargs: str) -> tuple[float, str, str]:
def _calculate_cvss(breakdown: dict[str, str]) -> tuple[float, str, str]:
try:
from cvss import CVSS3
vector = (
f"CVSS:3.1/AV:{kwargs['attack_vector']}/AC:{kwargs['attack_complexity']}/"
f"PR:{kwargs['privileges_required']}/UI:{kwargs['user_interaction']}/"
f"S:{kwargs['scope']}/C:{kwargs['confidentiality']}/"
f"I:{kwargs['integrity']}/A:{kwargs['availability']}"
f"CVSS:3.1/AV:{breakdown['attack_vector']}/AC:{breakdown['attack_complexity']}/"
f"PR:{breakdown['privileges_required']}/UI:{breakdown['user_interaction']}/"
f"S:{breakdown['scope']}/C:{breakdown['confidentiality']}/"
f"I:{breakdown['integrity']}/A:{breakdown['availability']}"
)
c = CVSS3(vector)
score = c.scores()[0]
@@ -165,18 +154,6 @@ _REQUIRED_FIELDS = {
}
_CVSS_VALID = {
"attack_vector": ["N", "A", "L", "P"],
"attack_complexity": ["L", "H"],
"privileges_required": ["N", "L", "H"],
"user_interaction": ["N", "R"],
"scope": ["U", "C"],
"confidentiality": ["N", "L", "H"],
"integrity": ["N", "L", "H"],
"availability": ["N", "L", "H"],
}
def _do_create( # noqa: PLR0912
*,
title: str,
@@ -187,12 +164,12 @@ def _do_create( # noqa: PLR0912
poc_description: str,
poc_script_code: str,
remediation_steps: str,
cvss_breakdown: str,
cvss_breakdown: dict[str, str],
endpoint: str | None,
method: str | None,
cve: str | None,
cwe: str | None,
code_locations: str | None,
code_locations: list[dict[str, Any]] | None,
) -> dict[str, Any]:
errors: list[str] = []
fields = {
@@ -209,16 +186,16 @@ def _do_create( # noqa: PLR0912
if not str(fields.get(name) or "").strip():
errors.append(msg)
parsed_cvss = _parse_cvss_xml(cvss_breakdown)
if not parsed_cvss:
errors.append("cvss: could not parse CVSS breakdown XML")
if not isinstance(cvss_breakdown, dict) or not cvss_breakdown:
errors.append("cvss_breakdown: must be an object with the 8 CVSS metrics")
cvss_breakdown = {}
else:
for name, valid in _CVSS_VALID.items():
value = parsed_cvss.get(name)
value = cvss_breakdown.get(name)
if value not in valid:
errors.append(f"Invalid {name}: {value}. Must be one of: {valid}")
parsed_locations = _parse_code_locations_xml(code_locations) if code_locations else None
parsed_locations = _normalize_code_locations(code_locations)
if parsed_locations:
errors.extend(_validate_code_locations(parsed_locations))
if cve:
@@ -235,8 +212,7 @@ def _do_create( # noqa: PLR0912
if errors:
return {"success": False, "message": "Validation failed", "errors": errors}
assert parsed_cvss is not None
cvss_score, severity, _vector = _calculate_cvss(**parsed_cvss)
cvss_score, severity, _vector = _calculate_cvss(cvss_breakdown)
try:
from strix.telemetry.tracer import get_global_tracer
@@ -294,7 +270,7 @@ def _do_create( # noqa: PLR0912
poc_script_code=poc_script_code,
remediation_steps=remediation_steps,
cvss=cvss_score,
cvss_breakdown=parsed_cvss,
cvss_breakdown=cvss_breakdown,
endpoint=endpoint,
method=method,
cve=cve,
@@ -315,7 +291,9 @@ def _do_create( # noqa: PLR0912
# 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)
# strict_mode=False because cvss_breakdown is a dict[str, str] and
# code_locations is list[dict] — both free-form for the strict schema.
@strix_tool(timeout=180, strict_mode=False)
async def create_vulnerability_report(
ctx: RunContextWrapper,
title: str,
@@ -326,12 +304,12 @@ async def create_vulnerability_report(
poc_description: str,
poc_script_code: str,
remediation_steps: str,
cvss_breakdown: str,
cvss_breakdown: dict[str, str],
endpoint: str | None = None,
method: str | None = None,
cve: str | None = None,
cwe: str | None = None,
code_locations: str | None = None,
code_locations: list[dict[str, Any]] | None = None,
) -> str:
"""File a vulnerability report — one report per fully-verified finding.
@@ -364,13 +342,13 @@ async def create_vulnerability_report(
- Avoid hedging language; be precise and non-vague.
**White-box requirement**: when source is available, you MUST
populate ``code_locations`` with nested XML including
``fix_before`` / ``fix_after`` for proposed fixes. The fix_before
must be a verbatim copy of source at the specified line range — it's
used as a literal GitHub/GitLab PR suggestion block.
populate ``code_locations`` with one entry per affected line range.
The ``fix_before`` field must be a verbatim copy of the source at
the specified line range — it's used as a literal GitHub/GitLab
PR suggestion block.
**CVSS breakdown** is required as nested XML with all 8 metrics
(each a single uppercase letter):
**CVSS breakdown** is an object with all 8 metrics (each a single
uppercase letter):
- ``attack_vector``: ``N`` (Network), ``A`` (Adjacent), ``L``
(Local), ``P`` (Physical)
@@ -381,6 +359,19 @@ async def create_vulnerability_report(
- ``confidentiality`` / ``integrity`` / ``availability``: ``N`` /
``L`` / ``H``
Example::
{
"attack_vector": "N",
"attack_complexity": "L",
"privileges_required": "N",
"user_interaction": "N",
"scope": "U",
"confidentiality": "H",
"integrity": "H",
"availability": "H"
}
**CVE / CWE rules**: pass the bare ID only (``CVE-2024-1234``,
``CWE-89``) — no name, no parenthetical. Be 100% certain; if
unsure, omit. Always prefer the most specific child CWE over a
@@ -397,14 +388,16 @@ async def create_vulnerability_report(
poc_description: Step-by-step reproduction.
poc_script_code: Working PoC (Python preferred).
remediation_steps: Specific, actionable fix.
cvss_breakdown: 8-metric XML block per the format above.
cvss_breakdown: 8-metric object per the format above.
endpoint: API path / Git path (e.g. ``/api/login``).
method: HTTP method when relevant.
cve: ``CVE-YYYY-NNNNN`` if certain, else omit.
cwe: ``CWE-NNN`` (most specific child) if certain, else omit.
code_locations: Required for white-box findings; nested XML
list with ``file``, ``start_line``, ``end_line``,
``snippet``, ``fix_before``, ``fix_after``.
code_locations: Required for white-box findings. List of
objects, each with ``file`` (relative path), ``start_line``,
``end_line``, optional ``snippet``, ``label``,
``fix_before`` (verbatim source), ``fix_after`` (suggested
replacement).
"""
del ctx
result = await asyncio.to_thread(
+3 -3
View File
@@ -53,14 +53,14 @@ async def test_pending_messages_appended_in_order() -> None:
assert len(out.input) == 3
assert out.input[0] == {"role": "user", "content": "task"}
assert "<inter_agent_message from='b'" in out.input[1]["content"]
assert "Message from agent b" in out.input[1]["content"]
assert "hello" in out.input[1]["content"]
assert "second" in out.input[2]["content"]
assert "priority='high'" in out.input[2]["content"]
assert "priority=high" in out.input[2]["content"]
@pytest.mark.asyncio
async def test_user_sender_skips_xml_wrap() -> None:
async def test_user_sender_skips_envelope() -> None:
bus = AgentMessageBus()
await bus.register("a1", "alpha", parent_id=None)
await bus.send("a1", {"from": "user", "content": "follow-up question"})
+4 -3
View File
@@ -93,7 +93,7 @@ async def test_on_llm_end_records_usage_and_increments_turn() -> None:
@pytest.mark.asyncio
async def test_on_agent_end_detects_crash() -> None:
"""C8 (AUDIT_R2): on_agent_end without agent_finish_called posts crash to parent."""
"""on_agent_end without agent_finish_called posts crash message to parent."""
hooks = StrixOrchestrationHooks()
bus = AgentMessageBus()
await bus.register("root", "root", parent_id=None)
@@ -104,8 +104,9 @@ async def test_on_agent_end_detects_crash() -> None:
drained = await bus.drain("root")
assert len(drained) == 1
assert "<agent_crash" in drained[0]["content"]
assert "agent_id='child'" in drained[0]["content"]
assert "[Agent crash]" in drained[0]["content"]
assert "child" in drained[0]["content"]
assert drained[0]["type"] == "crash"
assert bus.statuses["child"] == "crashed"
-271
View File
@@ -1,271 +0,0 @@
from collections.abc import Callable
import pytest
from strix.tools.argument_parser import (
ArgumentConversionError,
_convert_basic_types,
_convert_to_bool,
_convert_to_dict,
_convert_to_list,
convert_arguments,
convert_string_to_type,
)
class TestConvertToBool:
"""Tests for the _convert_to_bool function."""
@pytest.mark.parametrize(
"value",
["true", "True", "TRUE", "1", "yes", "Yes", "YES", "on", "On", "ON"],
)
def test_truthy_values(self, value: str) -> None:
"""Test that truthy string values are converted to True."""
assert _convert_to_bool(value) is True
@pytest.mark.parametrize(
"value",
["false", "False", "FALSE", "0", "no", "No", "NO", "off", "Off", "OFF"],
)
def test_falsy_values(self, value: str) -> None:
"""Test that falsy string values are converted to False."""
assert _convert_to_bool(value) is False
def test_non_standard_truthy_string(self) -> None:
"""Test that non-empty non-standard strings are truthy."""
assert _convert_to_bool("anything") is True
assert _convert_to_bool("hello") is True
def test_empty_string(self) -> None:
"""Test that empty string is falsy."""
assert _convert_to_bool("") is False
class TestConvertToList:
"""Tests for the _convert_to_list function."""
def test_json_array_string(self) -> None:
"""Test parsing a JSON array string."""
result = _convert_to_list('["a", "b", "c"]')
assert result == ["a", "b", "c"]
def test_json_array_with_numbers(self) -> None:
"""Test parsing a JSON array with numbers."""
result = _convert_to_list("[1, 2, 3]")
assert result == [1, 2, 3]
def test_comma_separated_string(self) -> None:
"""Test parsing a comma-separated string."""
result = _convert_to_list("a, b, c")
assert result == ["a", "b", "c"]
def test_comma_separated_no_spaces(self) -> None:
"""Test parsing comma-separated values without spaces."""
result = _convert_to_list("x,y,z")
assert result == ["x", "y", "z"]
def test_single_value(self) -> None:
"""Test that a single value returns a list with one element."""
result = _convert_to_list("single")
assert result == ["single"]
def test_json_non_array_wraps_in_list(self) -> None:
"""Test that a valid JSON non-array value is wrapped in a list."""
result = _convert_to_list('"string"')
assert result == ["string"]
def test_json_object_wraps_in_list(self) -> None:
"""Test that a JSON object is wrapped in a list."""
result = _convert_to_list('{"key": "value"}')
assert result == [{"key": "value"}]
def test_empty_json_array(self) -> None:
"""Test parsing an empty JSON array."""
result = _convert_to_list("[]")
assert result == []
class TestConvertToDict:
"""Tests for the _convert_to_dict function."""
def test_valid_json_object(self) -> None:
"""Test parsing a valid JSON object string."""
result = _convert_to_dict('{"key": "value", "number": 42}')
assert result == {"key": "value", "number": 42}
def test_empty_json_object(self) -> None:
"""Test parsing an empty JSON object."""
result = _convert_to_dict("{}")
assert result == {}
def test_invalid_json_returns_empty_dict(self) -> None:
"""Test that invalid JSON returns an empty dictionary."""
result = _convert_to_dict("not json")
assert result == {}
def test_json_array_returns_empty_dict(self) -> None:
"""Test that a JSON array returns an empty dictionary."""
result = _convert_to_dict("[1, 2, 3]")
assert result == {}
def test_nested_json_object(self) -> None:
"""Test parsing a nested JSON object."""
result = _convert_to_dict('{"outer": {"inner": "value"}}')
assert result == {"outer": {"inner": "value"}}
class TestConvertBasicTypes:
"""Tests for the _convert_basic_types function."""
def test_convert_to_int(self) -> None:
"""Test converting string to int."""
assert _convert_basic_types("42", int) == 42
assert _convert_basic_types("-10", int) == -10
def test_convert_to_float(self) -> None:
"""Test converting string to float."""
assert _convert_basic_types("3.14", float) == 3.14
assert _convert_basic_types("-2.5", float) == -2.5
def test_convert_to_str(self) -> None:
"""Test converting string to str (passthrough)."""
assert _convert_basic_types("hello", str) == "hello"
def test_convert_to_bool(self) -> None:
"""Test converting string to bool."""
assert _convert_basic_types("true", bool) is True
assert _convert_basic_types("false", bool) is False
def test_convert_to_list_type(self) -> None:
"""Test converting to list type."""
result = _convert_basic_types("[1, 2, 3]", list)
assert result == [1, 2, 3]
def test_convert_to_dict_type(self) -> None:
"""Test converting to dict type."""
result = _convert_basic_types('{"a": 1}', dict)
assert result == {"a": 1}
def test_unknown_type_attempts_json(self) -> None:
"""Test that unknown types attempt JSON parsing."""
result = _convert_basic_types('{"key": "value"}', object)
assert result == {"key": "value"}
def test_unknown_type_returns_original(self) -> None:
"""Test that unparseable values are returned as-is."""
result = _convert_basic_types("plain text", object)
assert result == "plain text"
class TestConvertStringToType:
"""Tests for the convert_string_to_type function."""
def test_basic_type_conversion(self) -> None:
"""Test basic type conversions."""
assert convert_string_to_type("42", int) == 42
assert convert_string_to_type("3.14", float) == 3.14
assert convert_string_to_type("true", bool) is True
def test_optional_type(self) -> None:
"""Test conversion with Optional type."""
result = convert_string_to_type("42", int | None)
assert result == 42
def test_union_type(self) -> None:
"""Test conversion with Union type."""
result = convert_string_to_type("42", int | str)
assert result == 42
def test_union_type_with_none(self) -> None:
"""Test conversion with Union including None."""
result = convert_string_to_type("hello", str | None)
assert result == "hello"
def test_modern_union_syntax(self) -> None:
"""Test conversion with modern union syntax (int | None)."""
result = convert_string_to_type("100", int | None)
assert result == 100
class TestConvertArguments:
"""Tests for the convert_arguments function."""
def test_converts_typed_arguments(
self, sample_function_with_types: Callable[..., None]
) -> None:
"""Test that arguments are converted based on type annotations."""
kwargs = {
"name": "test",
"count": "5",
"enabled": "true",
"ratio": "2.5",
"items": "[1, 2, 3]",
"config": '{"key": "value"}',
}
result = convert_arguments(sample_function_with_types, kwargs)
assert result["name"] == "test"
assert result["count"] == 5
assert result["enabled"] is True
assert result["ratio"] == 2.5
assert result["items"] == [1, 2, 3]
assert result["config"] == {"key": "value"}
def test_passes_through_none_values(
self, sample_function_with_types: Callable[..., None]
) -> None:
"""Test that None values are passed through unchanged."""
kwargs = {"name": "test", "count": None}
result = convert_arguments(sample_function_with_types, kwargs)
assert result["count"] is None
def test_passes_through_non_string_values(
self, sample_function_with_types: Callable[..., None]
) -> None:
"""Test that non-string values are passed through unchanged."""
kwargs = {"name": "test", "count": 42}
result = convert_arguments(sample_function_with_types, kwargs)
assert result["count"] == 42
def test_unknown_parameter_passed_through(
self, sample_function_with_types: Callable[..., None]
) -> None:
"""Test that parameters not in signature are passed through."""
kwargs = {"name": "test", "unknown_param": "value"}
result = convert_arguments(sample_function_with_types, kwargs)
assert result["unknown_param"] == "value"
def test_function_without_annotations(
self, sample_function_no_annotations: Callable[..., None]
) -> None:
"""Test handling of functions without type annotations."""
kwargs = {"arg1": "value1", "arg2": "42"}
result = convert_arguments(sample_function_no_annotations, kwargs)
assert result["arg1"] == "value1"
assert result["arg2"] == "42"
def test_raises_error_on_conversion_failure(
self, sample_function_with_types: Callable[..., None]
) -> None:
"""Test that ArgumentConversionError is raised on conversion failure."""
kwargs = {"count": "not_a_number"}
with pytest.raises(ArgumentConversionError) as exc_info:
convert_arguments(sample_function_with_types, kwargs)
assert exc_info.value.param_name == "count"
class TestArgumentConversionError:
"""Tests for the ArgumentConversionError exception class."""
def test_error_with_param_name(self) -> None:
"""Test creating error with parameter name."""
error = ArgumentConversionError("Test error", param_name="test_param")
assert error.param_name == "test_param"
assert str(error) == "Test error"
def test_error_without_param_name(self) -> None:
"""Test creating error without parameter name."""
error = ArgumentConversionError("Test error")
assert error.param_name is None
assert str(error) == "Test error"
+9 -7
View File
@@ -315,10 +315,10 @@ async def test_create_agent_spawns_and_registers_child() -> None:
assert bus.names[new_id] == "recon-bot"
assert new_id in bus.tasks
# Initial input shape: identity block + task message at the end.
# Initial input shape: identity preamble + task message at the end.
initial_input = runner_calls[0]["kwargs"]["input"]
assert any(
isinstance(item, dict) and "agent_delegation" in item.get("content", "")
isinstance(item, dict) and "You are agent recon-bot" in item.get("content", "")
for item in initial_input
)
assert initial_input[-1]["content"] == "enumerate hosts"
@@ -375,8 +375,8 @@ async def test_create_agent_inherits_parent_history() -> None:
initial_input = runner_calls[0]["input"]
contents = [item.get("content", "") for item in initial_input]
assert "<inherited_context_from_parent>" in contents
assert "</inherited_context_from_parent>" in contents
assert any("Inherited context from parent" in c for c in contents)
assert any("End of inherited context" in c for c in contents)
# Parent's exact items should be in between.
assert any(c == "scope: example.com" for c in contents)
@@ -427,9 +427,11 @@ async def test_agent_finish_posts_report_to_parent_inbox() -> None:
msg = parent_msgs[0]
assert msg["type"] == "completion"
assert msg["from"] == "child-A"
assert "found 3 issues" in msg["content"]
assert "<finding>xss in /search</finding>" in msg["content"]
assert "sanitize search input" in msg["content"]
payload = json.loads(msg["content"])
assert payload["kind"] == "agent_completion_report"
assert payload["summary"] == "found 3 issues"
assert "xss in /search" in payload["findings"]
assert "sanitize search input" in payload["recommendations"]
@pytest.mark.asyncio
+3 -3
View File
@@ -176,7 +176,7 @@ async def test_search_files_routes_to_sandbox() -> None:
@pytest.mark.asyncio
async def test_create_vulnerability_report_validates_required_fields() -> None:
"""Empty required fields should be rejected by the legacy validator."""
"""Empty required fields should be rejected by the validator."""
out = await _invoke(
create_vulnerability_report,
_ctx_for(),
@@ -188,7 +188,7 @@ async def test_create_vulnerability_report_validates_required_fields() -> None:
poc_description="pd",
poc_script_code="curl ...",
remediation_steps="rs",
cvss_breakdown="<attack_vector>N</attack_vector>",
cvss_breakdown={"attack_vector": "N"},
)
assert out["success"] is False
assert "errors" in out
@@ -220,7 +220,7 @@ async def test_create_vulnerability_report_delegates_to_impl() -> None:
poc_description="pd",
poc_script_code="pc",
remediation_steps="rs",
cvss_breakdown="<x/>",
cvss_breakdown={"attack_vector": "N"},
cve="CVE-2024-12345",
)
+6 -6
View File
@@ -1,7 +1,7 @@
"""Phase 2.5 smoke tests for the sandbox-bound SDK tool wrappers.
"""Smoke tests for the sandbox-bound SDK tool wrappers.
Covers: browser_action, terminal_execute, python_action, and the seven
Caido proxy tools.
Covers: browser_action, terminal_execute, python_action, file_edit
helpers, and the seven Caido proxy tools.
These wrappers are pure pass-throughs to ``post_to_sandbox`` — there's
no per-tool logic to assert, so the tests focus on:
@@ -9,9 +9,9 @@ no per-tool logic to assert, so the tests focus on:
- ``FunctionTool`` registration succeeds (which proves the SDK could
derive a JSON schema from the type hints — a non-trivial check given
Literal types, ``dict[str, str]``, and strict-mode opt-outs).
- The dispatch payload to ``post_to_sandbox`` mirrors the legacy XML
schema verbatim, so the in-container tool server gets the same
``kwargs`` shape it always has.
- The dispatch payload to ``post_to_sandbox`` carries the right
``kwargs`` shape so the in-container tool server can dispatch to the
matching action.
- The ``send_request`` / ``repeat_request`` tools opt out of strict
schema mode (their ``headers`` / ``modifications`` dicts are
free-form and would otherwise fail registration).