feat: add swarm preset inspection command (#73)

agent/cli.py

@@ -671,6 +671,7 @@ def _print_help() -> None:
         ("/continue <run_id> <prompt>", "Continue an existing run"),
         ("/swarm", "List swarm team presets"),
         ("/swarm run <preset> {vars}", "Run a swarm team"),
+        ("/swarm inspect <preset>", "Inspect preset DAG and validation"),
         ("/swarm list", "List swarm run history"),
         ("/swarm show <run_id>", "Show swarm run details"),
         ("/swarm cancel <run_id>", "Cancel a swarm run"),
@@ -779,6 +780,11 @@ def _handle_swarm_command(arg: str) -> None:
         preset = run_parts[0]
         vars_json = run_parts[1] if len(run_parts) > 1 else None
         cmd_swarm_run_live(preset, vars_json)
+    elif sub == "inspect":
+        if sub_arg:
+            cmd_swarm_inspect(sub_arg)
+        else:
+            console.print("[red]Usage: /swarm inspect <preset>[/red]")
     elif sub == "list":
         cmd_swarm_list()
     elif sub == "show":
@@ -1409,6 +1415,78 @@ def cmd_swarm_run(preset: str, vars_json: Optional[str] = None) -> None:
     cmd_swarm_run_live(preset, vars_json)
 
 
+def cmd_swarm_inspect(preset: str) -> int:
+    """Inspect a swarm preset without starting workers."""
+    from src.swarm.presets import inspect_preset
+
+    try:
+        report = inspect_preset(preset)
+    except FileNotFoundError as exc:
+        console.print(f"[red]{exc}[/red]")
+        return EXIT_USAGE_ERROR
+    except Exception as exc:
+        console.print(f"[red]Failed to inspect preset:[/red] {exc}")
+        return EXIT_RUN_FAILED
+
+    status = "OK" if report["valid"] else "INVALID"
+    status_color = "green" if report["valid"] else "red"
+    lines = [
+        f"[bold]Preset:[/bold] {report['name']}",
+        f"[bold]Title:[/bold] {report.get('title') or '-'}",
+        f"[bold]Status:[/bold] [{status_color}]{status}[/{status_color}]",
+        f"[bold]Agents:[/bold] {len(report['agents'])}",
+        f"[bold]Tasks:[/bold] {len(report['tasks'])}",
+        f"[bold]Variables:[/bold] {', '.join(report['variables']) or '-'}",
+    ]
+    if report.get("description"):
+        lines.append(f"[bold]Description:[/bold] {report['description']}")
+    console.print(Panel("\n".join(lines), border_style=status_color, title="Swarm Preset Inspect"))
+
+    agent_table = Table(title="Agents", show_lines=False)
+    agent_table.add_column("ID", style="cyan", no_wrap=True)
+    agent_table.add_column("Role")
+    agent_table.add_column("Tools", max_width=40)
+    for agent in report["agents"]:
+        agent_table.add_row(
+            agent["id"],
+            agent.get("role", ""),
+            ", ".join(agent.get("tools", [])),
+        )
+    console.print(agent_table)
+
+    dag_table = Table(title="DAG Execution Plan", show_lines=False)
+    dag_table.add_column("Layer", justify="right", width=6)
+    dag_table.add_column("Task", style="cyan")
+    dag_table.add_column("Agent")
+    dag_table.add_column("Depends On")
+    task_details = {task["id"]: task for task in report["tasks"]}
+    for idx, layer in enumerate(report["layers"], start=1):
+        for item in layer:
+            task = task_details[item["task_id"]]
+            dag_table.add_row(
+                str(idx),
+                item["task_id"],
+                item["agent_id"],
+                ", ".join(task.get("depends_on", [])) or "-",
+            )
+    console.print(dag_table)
+
+    validation_table = Table(title="Validation", show_lines=False)
+    validation_table.add_column("Level", width=8)
+    validation_table.add_column("Message")
+    if report["errors"]:
+        for error in report["errors"]:
+            validation_table.add_row("[red]ERROR[/red]", error)
+    if report["warnings"]:
+        for warning in report["warnings"]:
+            validation_table.add_row("[yellow]WARN[/yellow]", warning)
+    if not report["errors"] and not report["warnings"]:
+        validation_table.add_row("[green]OK[/green]", "No issues found")
+    console.print(validation_table)
+
+    return EXIT_SUCCESS if report["valid"] else EXIT_RUN_FAILED
+
+
 def cmd_swarm_list() -> None:
     """List swarm run history."""
     from src.swarm.store import SwarmStore
@@ -1684,6 +1762,7 @@ def _build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--max-iter", type=int, default=50, help="Maximum agent iterations")
 
     parser.add_argument("--swarm-presets", action="store_true", help="List swarm presets")
+    parser.add_argument("--swarm-inspect", metavar="PRESET", help="Inspect a swarm preset without running it")
     parser.add_argument("--swarm-run", nargs="+", metavar=("PRESET", "VARS"), help="Run a swarm preset")
     parser.add_argument("--swarm-list", action="store_true", help="List swarm runs")
     parser.add_argument("--swarm-show", metavar="RUN_ID", help="Show a swarm run")
@@ -2070,6 +2149,8 @@ def main(argv: list[str] | None = None) -> int:
 
     if args.swarm_presets:
         return _coerce_exit_code(cmd_swarm_presets())
+    if args.swarm_inspect:
+        return _coerce_exit_code(cmd_swarm_inspect(args.swarm_inspect))
     if args.swarm_run:
         preset_name = args.swarm_run[0]
         vars_json = args.swarm_run[1] if len(args.swarm_run) > 1 else None

agent/src/swarm/__init__.py

@@ -12,7 +12,7 @@ from src.swarm.models import (
     TaskStatus,
     WorkerResult,
 )
-from src.swarm.presets import build_run_from_preset, list_presets, load_preset
+from src.swarm.presets import build_run_from_preset, inspect_preset, list_presets, load_preset
 from src.swarm.runtime import SwarmRuntime
 from src.swarm.store import SwarmStore
 from src.swarm.worker import run_worker
@@ -29,6 +29,7 @@ __all__ = [
     "TaskStatus",
     "WorkerResult",
     "build_run_from_preset",
+    "inspect_preset",
     "list_presets",
     "load_preset",
     "run_worker",

agent/src/swarm/presets.py

@@ -9,14 +9,18 @@ behavior under editable installs and built wheels.
 from __future__ import annotations
 
 import uuid
+from collections import Counter, defaultdict
 from datetime import datetime, timezone
 from pathlib import Path
+from string import Formatter
 
 import yaml
 
 from src.swarm.models import RunStatus, SwarmAgentSpec, SwarmRun, SwarmTask, TaskStatus
+from src.swarm.task_store import topological_layers, validate_dag
 
 PRESETS_DIR = Path(__file__).resolve().parent / "presets"
+_INTERNAL_TEMPLATE_VARS = {"upstream_context"}
 
 
 def load_preset(name: str) -> dict:
@@ -67,6 +71,146 @@ def list_presets() -> list[dict]:
     return results
 
 
+def _declared_variable_names(raw_variables: list) -> set[str]:
+    """Extract variable names from the YAML variables section."""
+    names: set[str] = set()
+    for item in raw_variables:
+        if isinstance(item, dict):
+            name = item.get("name")
+        else:
+            name = str(item)
+        if name:
+            names.add(str(name))
+    return names
+
+
+def _template_variables(template: str) -> set[str]:
+    """Return Python format fields referenced by a prompt template."""
+    variables: set[str] = set()
+    for _, field_name, _, _ in Formatter().parse(template or ""):
+        if not field_name:
+            continue
+        root = field_name.split(".", 1)[0].split("[", 1)[0]
+        if root and root not in _INTERNAL_TEMPLATE_VARS:
+            variables.add(root)
+    return variables
+
+
+def inspect_preset(name: str) -> dict:
+    """Validate a swarm preset and return a dry-run execution plan.
+
+    This does not start workers or call an LLM. It catches common YAML/DAG
+    mistakes early and exposes the topological task layers used by the runtime.
+    """
+    data = load_preset(name)
+    run = build_run_from_preset(name, {})
+
+    errors: list[str] = []
+    warnings: list[str] = []
+
+    agent_ids = [agent.id for agent in run.agents]
+    task_ids = [task.id for task in run.tasks]
+    agent_id_set = set(agent_ids)
+    task_id_set = set(task_ids)
+
+    for duplicate in sorted(item for item, count in Counter(agent_ids).items() if count > 1):
+        errors.append(f"Duplicate agent id: {duplicate}")
+    for duplicate in sorted(item for item, count in Counter(task_ids).items() if count > 1):
+        errors.append(f"Duplicate task id: {duplicate}")
+
+    for task in run.tasks:
+        if task.agent_id not in agent_id_set:
+            errors.append(f"Task '{task.id}' references unknown agent '{task.agent_id}'")
+        for _, upstream_task_id in task.input_from.items():
+            if upstream_task_id not in task_id_set:
+                errors.append(
+                    f"Task '{task.id}' input_from references unknown task '{upstream_task_id}'"
+                )
+
+    layers: list[list[str]] = []
+    try:
+        validate_dag(run.tasks)
+        layers = topological_layers(run.tasks)
+    except ValueError as exc:
+        errors.append(str(exc))
+
+    dependents: dict[str, list[str]] = defaultdict(list)
+    for task in run.tasks:
+        for dep in task.depends_on:
+            dependents[dep].append(task.id)
+
+    def is_upstream(candidate: str, task_id: str) -> bool:
+        """Return whether candidate can reach task_id through dependency edges."""
+        seen: set[str] = set()
+        stack = [candidate]
+        while stack:
+            current = stack.pop()
+            if current == task_id:
+                return True
+            if current in seen:
+                continue
+            seen.add(current)
+            stack.extend(dependents.get(current, []))
+        return False
+
+    for task in run.tasks:
+        for key, upstream_task_id in task.input_from.items():
+            if upstream_task_id in task_id_set and not is_upstream(upstream_task_id, task.id):
+                warnings.append(
+                    f"Task '{task.id}' input_from '{key}' references '{upstream_task_id}', "
+                    "which is not upstream in the DAG"
+                )
+
+    declared_variables = _declared_variable_names(data.get("variables", []))
+    used_variables: set[str] = set()
+    for task in data.get("tasks", []):
+        try:
+            used_variables.update(_template_variables(task.get("prompt_template", "")))
+        except ValueError as exc:
+            errors.append(f"Task '{task.get('id', '?')}' has invalid prompt template: {exc}")
+
+    missing_declarations = sorted(used_variables - declared_variables)
+    unused_declarations = sorted(declared_variables - used_variables)
+    if missing_declarations:
+        warnings.append(
+            "Prompt templates use undeclared variables: " + ", ".join(missing_declarations)
+        )
+    if unused_declarations:
+        warnings.append(
+            "Declared variables are not used by task prompt templates: "
+            + ", ".join(unused_declarations)
+        )
+
+    task_agent = {task.id: task.agent_id for task in run.tasks}
+    return {
+        "name": data.get("name", name),
+        "title": data.get("title", ""),
+        "description": data.get("description", ""),
+        "valid": not errors,
+        "errors": errors,
+        "warnings": warnings,
+        "variables": sorted(declared_variables),
+        "used_variables": sorted(used_variables),
+        "agents": [
+            {"id": agent.id, "role": agent.role, "tools": agent.tools, "skills": agent.skills}
+            for agent in run.agents
+        ],
+        "tasks": [
+            {
+                "id": task.id,
+                "agent_id": task.agent_id,
+                "depends_on": task.depends_on,
+                "input_from": task.input_from,
+            }
+            for task in run.tasks
+        ],
+        "layers": [
+            [{"task_id": task_id, "agent_id": task_agent.get(task_id, "")} for task_id in layer]
+            for layer in layers
+        ],
+    }
+
+
 def build_run_from_preset(preset_name: str, user_vars: dict[str, str]) -> SwarmRun:
     """Create a SwarmRun from a preset with user variables applied.
 

agent/tests/test_swarm_preset_inspect.py

@@ -0,0 +1,65 @@
+"""Tests for swarm preset inspection and static validation."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from src.swarm import presets
+
+
+def test_all_bundled_presets_inspect_without_errors() -> None:
+    """Bundled presets should have valid agent/task references and DAGs."""
+    for entry in presets.list_presets():
+        report = presets.inspect_preset(entry["name"])
+        assert report["valid"], f"{entry['name']} errors: {report['errors']}"
+        assert report["layers"], f"{entry['name']} has no execution layers"
+
+
+def test_inspect_preset_returns_dry_run_layers() -> None:
+    report = presets.inspect_preset("investment_committee")
+
+    assert report["valid"]
+    assert report["variables"] == ["market", "target"]
+    assert report["layers"][0] == [
+        {"task_id": "task-bull", "agent_id": "bull_advocate"},
+        {"task_id": "task-bear", "agent_id": "bear_advocate"},
+    ]
+    assert report["layers"][-1] == [
+        {"task_id": "task-decision", "agent_id": "portfolio_manager"}
+    ]
+
+
+def test_inspect_preset_reports_invalid_references(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    preset_dir = tmp_path / "presets"
+    preset_dir.mkdir()
+    (preset_dir / "broken.yaml").write_text(
+        """
+name: broken
+title: Broken Preset
+agents:
+  - id: analyst
+    role: Analyst
+    system_prompt: ""
+tasks:
+  - id: task-a
+    agent_id: missing_agent
+    prompt_template: "Analyze {target}"
+    depends_on: [missing_task]
+variables:
+  - name: market
+""",
+        encoding="utf-8",
+    )
+
+    monkeypatch.setattr(presets, "PRESETS_DIR", preset_dir)
+
+    report = presets.inspect_preset("broken")
+
+    assert not report["valid"]
+    assert "Task 'task-a' references unknown agent 'missing_agent'" in report["errors"]
+    assert "Task 'task-a' depends on unknown task 'missing_task'" in report["errors"]
+    assert "Prompt templates use undeclared variables: target" in report["warnings"]
+    assert "Declared variables are not used by task prompt templates: market" in report["warnings"]