fix(agent): escape XML values, CDATA content, generic corrective message

- Add html.escape() to target values in <scan_task> (URLs, paths, IPs)
- Escape sender_name/sender_id in <agent_message> attributes
- CDATA-wrap message content in <agent_message> to handle any text
- Make corrective message generic (no StrixAgent-specific tool names)

Author: 0xhis

strix/agents/StrixAgent/strix_agent.py

@@ -1,5 +1,7 @@
 from typing import Any
 
+import html
+
 from strix.agents.base_agent import BaseAgent
 from strix.llm.config import LLMConfig
 
@@ -59,24 +61,24 @@ async def execute_scan(self, scan_config: dict[str, Any]) -> dict[str, Any]:  #
 
         target_lines = []
 
-        if repositories:
+if repositories:
             for repo in repositories:
                 if repo["workspace_path"]:
-                    target_lines.append(f'  <target type="repository">{repo["url"]} (code at: {repo["workspace_path"]})</target>')
+                    target_lines.append(f'  <target type="repository">{html.escape(repo["url"])} (code at: {html.escape(repo["workspace_path"])})</target>')
                 else:
-                    target_lines.append(f'  <target type="repository">{repo["url"]}</target>')
+                    target_lines.append(f'  <target type="repository">{html.escape(repo["url"])}</target>')
 
         if local_code:
             for code in local_code:
-                target_lines.append(f'  <target type="local_code">{code["path"]} (code at: {code["workspace_path"]})</target>')
+                target_lines.append(f'  <target type="local_code">{html.escape(code["path"])} (code at: {html.escape(code["workspace_path"])})</target>')
 
         if urls:
             for url in urls:
-                target_lines.append(f'  <target type="url">{url}</target>')
+                target_lines.append(f'  <target type="url">{html.escape(url)}</target>')
 
         if ip_addresses:
             for ip in ip_addresses:
-                target_lines.append(f'  <target type="ip">{ip}</target>')
+                target_lines.append(f'  <target type="ip">{html.escape(ip)}</target>')
 
         targets_block = "\n".join(target_lines)
 

strix/agents/base_agent.py

@@ -1,5 +1,6 @@
 import asyncio
 import contextlib
+import html
 import logging
 from typing import TYPE_CHECKING, Any, Optional
 
@@ -413,11 +414,7 @@ async def _process_iteration(self, tracer: Optional["Tracer"]) -> bool | None:
         corrective_message = (
             "You responded with plain text instead of a tool call. "
             "While the agent loop is running, EVERY response MUST be a tool call. "
-            "Do NOT send plain text messages. Act via tools:\n"
-            "- Use the think tool to reason through problems\n"
-            "- Use create_agent to spawn subagents for testing\n"
-            "- Use terminal_execute to run commands\n"
-            "- Use wait_for_message ONLY when waiting for subagent results\n"
+            "Do NOT send plain text messages. Act via your available tools. "
             "Review your task and take action now."
         )
         self.state.add_message("user", corrective_message)
@@ -499,12 +496,13 @@ def _check_agent_messages(self, state: AgentState) -> None:  # noqa: PLR0912
                             if sender_id and sender_id in _agent_graph.get("nodes", {}):
                                 sender_name = _agent_graph["nodes"][sender_id]["name"]
 
+                            content = message.get("content", "")
                             message_content = f"""<agent_message
-from="{sender_name}"
-id="{sender_id}"
-type="{message.get("message_type", "information")}"
-priority="{message.get("priority", "normal")}">
-{message.get("content", "")}
+from="{html.escape(sender_name)}"
+id="{html.escape(str(sender_id))}"
+type="{html.escape(message.get("message_type", "information"))}"
+priority="{html.escape(message.get("priority", "normal"))}">
+<![CDATA[{content}]]>
 </agent_message>"""
                             state.add_message("user", message_content.strip())