Files
strix/strix/interface/tui/renderers/agents_graph_renderer.py
T
Ahmed Allam 1c1fa49961 refactor(tools): split wait_for_message into respond_to_user + wait_for_agents
One tool was doing three jobs (wait on the user, wait on other agents, and
- wrongly - wait for a long-running command), so the driver had to guess which
one an agent meant and used parent_id as the proxy: the root waits for a human,
everyone else waits for agents. That proxy is wrong, since the user can message
any agent from the TUI's agent tree.

Tool identity now carries the intent, and the coordinator records it as a
wait_kind that survives snapshot/restore:

  respond_to_user  -> wait_kind="user",   never auto-resumed (root or not)
  wait_for_agents  -> wait_kind="agents", auto-resumed on a 300s timer
  recovery exhaust -> wait_kind="stalled"

respond_to_user fuses the message and the yield into one call, so there is no
way to answer and then forget to stop - the two-step that gpt-4o-mini skipped
2/2 in live testing. Plain text still renders as before.

Auto-resume is also bounded now: an agent that re-parks after every timeout
burned a model turn every 300s for the rest of the scan (and, since parked
children notify their parent, spammed the parent's inbox on the same cycle).
After _MAX_IDLE_AUTO_RESUMES it stays parked until a real message arrives.
2026-08-02 02:15:51 +03:00

176 lines
5.5 KiB
Python

from typing import Any, ClassVar
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
@register_tool_renderer
class ViewAgentGraphRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "view_agent_graph"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
status = tool_data.get("status", "unknown")
text = Text()
text.append("◇ ", style="#a78bfa")
text.append("viewing agents graph", style="dim")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class CreateAgentRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "create_agent"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
status = tool_data.get("status", "unknown")
task = args.get("task", "")
name = args.get("name", "Agent")
text = Text()
text.append("◈ ", style="#a78bfa")
text.append("spawning ", style="dim")
text.append(name, style="bold #a78bfa")
if task:
text.append("\n ")
text.append(task, style="dim")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class SendMessageToAgentRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "send_message_to_agent"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
status = tool_data.get("status", "unknown")
message = args.get("message", "")
target_agent_id = args.get("target_agent_id", "")
text = Text()
text.append("→ ", style="#60a5fa")
if target_agent_id:
text.append(f"to {target_agent_id}", style="dim")
else:
text.append("sending message", style="dim")
if message:
text.append("\n ")
text.append(message, style="dim")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class AgentFinishRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "agent_finish"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
result_summary = args.get("result_summary", "")
findings = args.get("findings", [])
success = args.get("success", True)
text = Text()
if success:
text.append("◆ ", style="#22c55e")
text.append("Agent completed", style="bold #22c55e")
else:
text.append("◆ ", style="#ef4444")
text.append("Agent failed", style="bold #ef4444")
if result_summary:
text.append("\n ")
text.append(result_summary, style="bold")
if findings and isinstance(findings, list):
for finding in findings:
text.append("\n • ")
text.append(str(finding), style="dim")
else:
text.append("\n ")
text.append("Completing task...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class WaitForAgentsRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "wait_for_agents"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
status = tool_data.get("status", "unknown")
reason = args.get("reason", "")
text = Text()
text.append("○ ", style="#6b7280")
text.append("waiting", style="dim")
if reason:
text.append("\n ")
text.append(reason, style="dim")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class StopAgentRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "stop_agent"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "unknown")
target_agent_id = args.get("target_agent_id", "")
cascade = args.get("cascade", True)
reason = args.get("reason", "")
text = Text()
text.append("◼ ", style="#ef4444")
text.append("stopping", style="dim")
if target_agent_id:
text.append(f" {target_agent_id}", style="bold #ef4444")
if cascade:
text.append(" + descendants", style="dim italic")
if reason:
text.append("\n ")
text.append(reason, style="dim")
if isinstance(result, dict) and result.get("success") is False and result.get("error"):
text.append("\n ")
text.append(str(result["error"]), style="#ef4444")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)