Agents graph sweep: status taxonomy, stop_agent safety, skill validation

- view_agent_graph status summary now derives buckets from the canonical
  Status literal via get_args, so adding a new status in core.agents
  auto-flows into the summary. The previous hardcoded five-bucket list
  silently omitted "failed" — buckets stopped summing to total whenever
  an agent failed.

- stop_agent rejects targets that are already in a terminal status
  (completed / stopped / crashed / failed) with a model-readable error
  pointing at view_agent_graph and send_message_to_agent. request_stop
  unconditionally overwrites status, so without this guard calling
  stop_agent on a completed agent erased the "completed" history.

- StopAgentRenderer added — was falling back to the generic key/value
  renderer; the rest of the agents_graph tools have purpose-built ones.

- agent_finish root-rejection payload trimmed from
  {success, agent_completed, error, parent_notified} to {success, error}.
  The lifecycle gate only reads success+agent_completed and they were
  always False/False on this branch, so the extra fields were dead weight.

- wait_for_message renames its top-level outcome field from "status" to
  "wait_outcome" — "status" overloaded with the coordinator's agent
  status literal (which also has "stopped" as a value, different
  meaning). Redundant "agent_waiting" boolean dropped (true iff
  wait_outcome == "waiting"). Consumer at factory._wait_tool_parked
  updated to match.

- send_message_to_agent now refuses self-send with a pointer at think /
  agent_finish / finish_scan instead of looping a message into your
  own session.

- SendMessageToAgentRenderer read args.get("agent_id") but the tool's
  param is target_agent_id, so the TUI silently never showed the target.
  Fixed.

- Restored skill validation lost during the SDK migration: skills
  module re-exports get_all_skill_names and validate_requested_skills
  (excluding internal scan_modes/coordination categories from the
  user-selectable set). create_agent now validates skills before
  spawning instead of silently accepting unknown names.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

strix/agents/factory.py

@@ -213,8 +213,7 @@ def _wait_tool_parked(tool_name: str, output: Any) -> bool:
     return bool(
         isinstance(parsed, dict)
         and parsed.get("success")
-        and parsed.get("agent_waiting")
+        and parsed.get("wait_outcome") == "waiting"
-        and parsed.get("status") == "waiting"
     )
 
 

strix/interface/tui/renderers/agents_graph_renderer.py

@@ -61,12 +61,12 @@ class SendMessageToAgentRenderer(BaseToolRenderer):
         status = tool_data.get("status", "unknown")
 
         message = args.get("message", "")
-        agent_id = args.get("agent_id", "")
+        target_agent_id = args.get("target_agent_id", "")
 
         text = Text()
         text.append("→ ", style="#60a5fa")
-        if agent_id:
+        if target_agent_id:
-            text.append(f"to {agent_id}", style="dim")
+            text.append(f"to {target_agent_id}", style="dim")
         else:
             text.append("sending message", style="dim")
 
@@ -138,3 +138,38 @@ class WaitForMessageRenderer(BaseToolRenderer):
 
         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)

strix/skills/__init__.py

@@ -8,6 +8,52 @@ logger = logging.getLogger(__name__)
 
 _FRONTMATTER_PATTERN = re.compile(r"^---\s*\n.*?\n---\s*\n", re.DOTALL)
 
+# Categories loaded by the prompt template via explicit slash-form paths
+# (``scan_modes/<mode>``, ``coordination/<role>``). They're not
+# user-selectable through ``create_agent``'s ``skills`` argument.
+_INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordination"})
+
+
+def get_all_skill_names() -> set[str]:
+    """Return every user-selectable skill name (bare, no category prefix).
+
+    Used by :func:`validate_requested_skills` so ``create_agent`` can
+    reject typos / hallucinated names with a useful list, and by callers
+    that need to enumerate the catalog.
+    """
+    skills_dir = get_strix_resource_path("skills")
+    if not skills_dir.exists():
+        return set()
+    names: set[str] = set()
+    for category_dir in skills_dir.iterdir():
+        if not category_dir.is_dir() or category_dir.name.startswith("__"):
+            continue
+        if category_dir.name in _INTERNAL_SKILL_CATEGORIES:
+            continue
+        for file_path in category_dir.glob("*.md"):
+            names.add(file_path.stem)
+    return names
+
+
+def validate_requested_skills(skill_list: list[str], max_skills: int = 5) -> str | None:
+    """Validate a list of user-passed skill names.
+
+    Returns ``None`` on success, or a model-readable error message
+    describing what was wrong (count exceeded, unknown names).
+    """
+    if len(skill_list) > max_skills:
+        return (
+            f"Cannot specify more than {max_skills} skills per agent; "
+            f"got {len(skill_list)}. Aim for 1-3 related skills per specialist."
+        )
+    if not skill_list:
+        return None
+    available = get_all_skill_names()
+    invalid = sorted({s for s in skill_list if s not in available})
+    if invalid:
+        return f"Invalid skill name(s): {invalid}. Available skills: {sorted(available)}"
+    return None
+
 
 def load_skills(skill_names: list[str]) -> dict[str, str]:
     """Load skill markdown bodies (frontmatter stripped) by name.

strix/tools/agents_graph/tools.py

@@ -15,12 +15,20 @@ import asyncio
 import json
 import logging
 import uuid
+from collections import Counter
 from datetime import UTC, datetime
-from typing import Any, Literal
+from typing import Any, Literal, get_args
 
 from agents import RunContextWrapper, function_tool
 
-from strix.core.agents import coordinator_from_context
+from strix.core.agents import Status, coordinator_from_context
+from strix.skills import validate_requested_skills
+
+
+# An agent is "active" when stop_agent can meaningfully act on it. Anything
+# else (completed / stopped / crashed / failed) is terminal — request_stop
+# would forcibly overwrite that status, erasing the original outcome.
+_ACTIVE_STATUSES: frozenset[str] = frozenset({"running", "waiting"})
 
 
 logger = logging.getLogger(__name__)
@@ -107,14 +115,13 @@ async def view_agent_graph(ctx: RunContextWrapper) -> str:
     for root in roots:
         render(root, 0)
 
-    summary = {
+    # Derive per-status counts from the canonical ``Status`` literal so a
-        "total": len(parent_of),
+    # new status added in core.agents auto-flows into this summary without
-        "running": sum(1 for s in statuses.values() if s == "running"),
+    # the buckets silently going out of sync with the source of truth.
-        "waiting": sum(1 for s in statuses.values() if s == "waiting"),
+    counts = Counter(statuses.values())
-        "completed": sum(1 for s in statuses.values() if s == "completed"),
+    summary: dict[str, int] = {"total": len(parent_of)}
-        "crashed": sum(1 for s in statuses.values() if s == "crashed"),
+    for status_name in get_args(Status):
-        "stopped": sum(1 for s in statuses.values() if s == "stopped"),
+        summary[status_name] = counts.get(status_name, 0)
-    }
     return json.dumps(
         {
             "success": True,
@@ -169,6 +176,18 @@ async def send_message_to_agent(
             ensure_ascii=False,
             default=str,
         )
+    if target_agent_id == me:
+        return json.dumps(
+            {
+                "success": False,
+                "error": (
+                    "Cannot send a message to yourself. Use `think` to record a "
+                    "private note, or `agent_finish` / `finish_scan` to terminate."
+                ),
+            },
+            ensure_ascii=False,
+            default=str,
+        )
 
     msg_id = f"msg_{uuid.uuid4().hex[:8]}"
     delivered = await coordinator.send(
@@ -263,7 +282,7 @@ async def wait_for_message(  # noqa: PLR0911
         return json.dumps(
             {
                 "success": True,
-                "status": "stopped",
+                "wait_outcome": "stopped",
                 "reason": reason,
                 "note": "Wait ended because this agent is stopped.",
             },
@@ -277,7 +296,7 @@ async def wait_for_message(  # noqa: PLR0911
         return json.dumps(
             {
                 "success": True,
-                "status": "message_arrived",
+                "wait_outcome": "message_arrived",
                 "pending_messages": pending,
                 "messages": _session_items_payload(items),
                 "reason": reason,
@@ -291,8 +310,7 @@ async def wait_for_message(  # noqa: PLR0911
         return json.dumps(
             {
                 "success": True,
-                "status": "waiting",
+                "wait_outcome": "waiting",
-                "agent_waiting": True,
                 "reason": reason,
                 "note": "Agent parked; execution will resume when a message arrives.",
             },
@@ -308,7 +326,7 @@ async def wait_for_message(  # noqa: PLR0911
         return json.dumps(
             {
                 "success": True,
-                "status": "timeout",
+                "wait_outcome": "timeout",
                 "timeout_seconds": timeout_seconds,
                 "reason": reason,
                 "note": "No messages within timeout — continue work or call agent_finish.",
@@ -323,7 +341,7 @@ async def wait_for_message(  # noqa: PLR0911
         return json.dumps(
             {
                 "success": True,
-                "status": "stopped",
+                "wait_outcome": "stopped",
                 "reason": reason,
                 "note": "Wait ended because this agent is stopped.",
             },
@@ -337,7 +355,7 @@ async def wait_for_message(  # noqa: PLR0911
     return json.dumps(
         {
             "success": True,
-            "status": "message_arrived",
+            "wait_outcome": "message_arrived",
             "pending_messages": pending,
             "messages": _session_items_payload(items),
             "reason": reason,
@@ -415,6 +433,15 @@ async def create_agent(
             default=str,
         )
 
+    skill_list = list(skills or [])
+    skill_error = validate_requested_skills(skill_list)
+    if skill_error:
+        return json.dumps(
+            {"success": False, "error": skill_error, "agent_id": None},
+            ensure_ascii=False,
+            default=str,
+        )
+
     # ``ctx.turn_input`` carries the parent's full conversation up to and
     # including the call that's currently invoking ``create_agent``.
     parent_history = list(ctx.turn_input) if inherit_context and ctx.turn_input else []
@@ -423,7 +450,7 @@ async def create_agent(
             parent_ctx=inner,
             name=name,
             task=task,
-            skills=list(skills or []),
+            skills=skill_list,
             parent_history=parent_history,
         )
     except Exception as e:
@@ -439,7 +466,7 @@ async def create_agent(
         result.get("agent_id"),
         name,
         parent_id or "-",
-        len(skills or []),
+        len(skill_list),
         len(task or ""),
     )
 
@@ -508,11 +535,9 @@ async def agent_finish(
         return json.dumps(
             {
                 "success": False,
-                "agent_completed": False,
                 "error": (
                     "agent_finish is for subagents. Root/main agents must call finish_scan instead."
                 ),
-                "parent_notified": False,
             },
             ensure_ascii=False,
             default=str,
@@ -621,6 +646,25 @@ async def stop_agent(
             default=str,
         )
 
+    current_status = statuses[target_agent_id]
+    if current_status not in _ACTIVE_STATUSES:
+        return json.dumps(
+            {
+                "success": False,
+                "error": (
+                    f"Agent {target_agent_id} is already '{current_status}'; "
+                    "stop_agent only acts on running/waiting agents. Use "
+                    "view_agent_graph to find still-active descendants and "
+                    "stop them individually, or send_message_to_agent if you "
+                    "want to wake this one with new instructions."
+                ),
+                "target_agent_id": target_agent_id,
+                "current_status": current_status,
+            },
+            ensure_ascii=False,
+            default=str,
+        )
+
     if cascade:
         await coordinator.cancel_descendants_graceful(target_agent_id)
     else: