diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index ccce5dc..6dfaf35 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -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 | 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 %} diff --git a/strix/interface/tool_components/agent_message_renderer.py b/strix/interface/tool_components/agent_message_renderer.py index a51ea2a..b66f514 100644 --- a/strix/interface/tool_components/agent_message_renderer.py +++ b/strix/interface/tool_components/agent_message_renderer.py @@ -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) diff --git a/strix/interface/tool_components/reporting_renderer.py b/strix/interface/tool_components/reporting_renderer.py index 0993fcb..885aba8 100644 --- a/strix/interface/tool_components/reporting_renderer.py +++ b/strix/interface/tool_components/reporting_renderer.py @@ -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) diff --git a/strix/llm/dedupe.py b/strix/llm/dedupe.py index 9f0364a..8d2be96 100644 --- a/strix/llm/dedupe.py +++ b/strix/llm/dedupe.py @@ -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: - -true -vuln-0001 -0.95 -Both reports describe SQL injection in /api/login via the username parameter - +{ + "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: - -false - -0.90 -Different endpoints: candidate is /api/search, existing is /api/login - +{ + "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 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}>(.*?)" - 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"(.*?)", content, re.DOTALL | re.IGNORECASE - ) - - if not result_match: - logger.warning(f"No block found in response: {content[:500]}") - raise ValueError("No 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 XML block." + f"Respond with ONLY the JSON object described in the system prompt." ), }, ] diff --git a/strix/llm/utils.py b/strix/llm/utils.py deleted file mode 100644 index fd0cdf2..0000000 --- a/strix/llm/utils.py +++ /dev/null @@ -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".*?", re.DOTALL | re.IGNORECASE), - re.compile( - r".*?", - 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() diff --git a/strix/orchestration/filter.py b/strix/orchestration/filter.py index c5256c6..c7ddde4 100644 --- a/strix/orchestration/filter.py +++ b/strix/orchestration/filter.py @@ -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 ```` 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"" - f"{content}" - f"" - ), + "content": f"{header}\n{content}", } ) return ModelInputData( diff --git a/strix/orchestration/hooks.py b/strix/orchestration/hooks.py index 3c5c68b..9293247 100644 --- a/strix/orchestration/hooks.py +++ b/strix/orchestration/hooks.py @@ -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 - ```` 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": ( - "You are at 85% of your iteration " - "budget. Begin consolidating findings." + "[System warning] You are at 85% of your iteration " + "budget. Begin consolidating findings." ), } ) @@ -61,9 +61,8 @@ class StrixOrchestrationHooks(RunHooks[Any]): { "role": "user", "content": ( - "You have 3 iterations left. Your " + "[System warning] You have 3 iterations left. Your " "next tool call MUST be the finish tool." - "" ), } ) @@ -148,11 +147,9 @@ class StrixOrchestrationHooks(RunHooks[Any]): { "from": me, "content": ( - f"" - "Agent terminated without calling agent_finish. " - "Stop waiting on this child." - "" + f"[Agent crash] {bus.names.get(me, me)} ({me}) " + f"terminated without calling agent_finish. " + f"Stop waiting on this child." ), "type": "crash", }, diff --git a/strix/runtime/tool_server.py b/strix/runtime/tool_server.py index ee5fb49..ff6de01 100644 --- a/strix/runtime/tool_server.py +++ b/strix/runtime/tool_server.py @@ -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) diff --git a/strix/tools/agents_graph/tools.py b/strix/tools/agents_graph/tools.py index cf82ad8..a6d224a 100644 --- a/strix/tools/agents_graph/tools.py +++ b/strix/tools/agents_graph/tools.py @@ -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 - # 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": "", + "content": "[Inherited context from parent — read-only history]", } ) initial_input.extend(parent_history) initial_input.append( { "role": "user", - "content": "", + "content": "[End of inherited context]", } ) initial_input.append( { "role": "user", "content": ( - f"\n" - f"You are agent {name} ({child_id}). Parent is {parent_id}.\n" - f"Maintain self-identity. Use agent_finish when complete.\n" - f"" + 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 ```` 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" {f}" for f in (findings or [])) - rec_xml = "\n".join( - f" {r}" for r in (final_recommendations or []) - ) async with bus._lock: agent_name = bus.names.get(me, me) - report = ( - f"\n" - f" {result_summary}\n" - f" \n{findings_xml}\n \n" - f" \n{rec_xml}\n \n" - f"" + 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, diff --git a/strix/tools/argument_parser.py b/strix/tools/argument_parser.py deleted file mode 100644 index 0a85f00..0000000 --- a/strix/tools/argument_parser.py +++ /dev/null @@ -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 {} diff --git a/strix/tools/reporting/tool.py b/strix/tools/reporting/tool.py index 63a4a32..2c34fa8 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -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}>(.*?)", 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"(.*?)", 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}>(.*?)", 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( diff --git a/tests/orchestration/test_filter.py b/tests/orchestration/test_filter.py index 169e44e..c38dc0d 100644 --- a/tests/orchestration/test_filter.py +++ b/tests/orchestration/test_filter.py @@ -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 " 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"}) diff --git a/tests/orchestration/test_hooks.py b/tests/orchestration/test_hooks.py index 08d9b05..004becb 100644 --- a/tests/orchestration/test_hooks.py +++ b/tests/orchestration/test_hooks.py @@ -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 " 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" diff --git a/tests/tools/test_graph_tools.py b/tests/tools/test_graph_tools.py index 1b89d82..bf99e67 100644 --- a/tests/tools/test_graph_tools.py +++ b/tests/tools/test_graph_tools.py @@ -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 "" in contents - assert "" 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 "xss in /search" 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 diff --git a/tests/tools/test_remaining_local_tools.py b/tests/tools/test_remaining_local_tools.py index 9853e7d..f4958bd 100644 --- a/tests/tools/test_remaining_local_tools.py +++ b/tests/tools/test_remaining_local_tools.py @@ -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="N", + 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="", + cvss_breakdown={"attack_vector": "N"}, cve="CVE-2024-12345", ) diff --git a/tests/tools/test_sandbox_tools.py b/tests/tools/test_sandbox_tools.py index f8990d5..f6bd054 100644 --- a/tests/tools/test_sandbox_tools.py +++ b/tests/tools/test_sandbox_tools.py @@ -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).