"""``muse coord dag`` — inspect the coordination dependency DAG. Shows which reservations are waiting for others to complete, reports blocked status, detects cycles, and computes the correct execution order via topological sort. Overview -------- When multiple agents reserve addresses in parallel, some agents may declare that their work depends on another agent's reservation being released first. The dependency DAG captures these ordering constraints. ``muse coord dag`` reads all dependency records from ``.muse/coordination/dependencies/``, cross-references the currently active reservations, and answers three questions: 1. **Who is blocked?** — which reservations cannot start because a dependency is still active. 2. **What order should work proceed?** — topological sort respecting all declared dependencies. 3. **Are there any cycles?** — flag immediately; cycles mean no agent can ever become unblocked. Output examples --------------- Text (full DAG, default):: Dependency DAG — 4 node(s), 3 edge(s) 1 blocked 3 unblocked 0 cycles TOPO STATUS ID DEPENDS-ON ──────────────────────────────────────────────────────────────────── 1 unblocked a1b2c3d4 (no deps) 2 unblocked e5f6a7b8 (no deps) 3 BLOCKED c9d0e1f2 a1b2c3d4 4 unblocked 01234567 e5f6a7b8 Text (flat, no topo column, default without --topo):: STATUS ID DEPENDS-ON ──────────────────────────────────────────────────────────────────── unblocked a1b2c3d4 (no deps) BLOCKED c9d0e1f2 a1b2c3d4 JSON:: { "schema_version": "...", "total_nodes": 4, "total_edges": 3, "blocked_count": 1, "active_only": false, "cycle": null, "nodes": [ { "reservation_id": "c9d0e1f2-...", "depends_on": ["a1b2c3d4-..."], "active": true, "blocked": true, "blocking": ["a1b2c3d4-..."], "topo_index": 3 }, ... ] } Exit codes:: 0 — success (even when blocked nodes exist) 1 — cycle detected, bad arguments, or unexpected error """ from __future__ import annotations import argparse import json import sys from muse.core.coordination import _validate_reservation_id, active_reservations from muse.core.dag import ( DependencyRecord, detect_cycle, get_blocking, is_blocked, load_all_dependencies, load_dag, topological_sort, ) from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.validation import sanitize_display type _Graph = dict[str, set[str]] def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register ``dag`` on *subparsers* (under ``muse coord``). Wires all flags with their defaults, choices, and help text so that ``--help`` output is accurate. Sets ``func`` to :func:`run`. Flags registered ---------------- ``--reservation-id UUID`` Show dependency details for a single reservation. Must be a valid UUID when provided. ``--topo`` Print nodes in topological execution order with an explicit TOPO column. Without this flag, a simpler flat table (no TOPO column) is shown. ``--active-only`` Restrict the graph to nodes that correspond to currently active reservations. Nodes whose reservation has already expired or been released are hidden. ``--format`` / ``--json`` Emit compact JSON to stdout; default is human-readable text. """ parser = subparsers.add_parser( "dag", help="Inspect the coordination dependency DAG.", formatter_class=argparse.RawDescriptionHelpFormatter, description=__doc__, ) parser.add_argument( "--reservation-id", dest="reservation_id", default=None, metavar="UUID", help=( "Show dependency details for a single reservation ID. " "Must be a valid UUID. When omitted the full DAG is shown." ), ) parser.add_argument( "--topo", action="store_true", default=False, help=( "Print nodes in topological execution order with a TOPO column. " "Without this flag, a simpler flat table is shown." ), ) parser.add_argument( "--active-only", action="store_true", default=False, dest="active_only", help=( "Restrict output to nodes whose reservation is currently active. " "Expired and released reservations are hidden." ), ) parser.add_argument( "--format", "-f", default="text", dest="fmt", choices=("text", "json"), help="Output format: text (default) or json.", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Inspect the coordination dependency DAG. Loads all dependency records and active reservations, derives blocked status for each node, and optionally computes topological order. Execution order --------------- 1. **Validate inputs** — ``--reservation-id``, when provided, must be a valid UUID. Validation fires before any file I/O; failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message on *stderr* (or compact JSON when ``--format json``). 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`. 3. **Load graph** — reads all dependency records and active reservations from ``.muse/coordination/``. 4. **Detect cycle** — O(V + E) DFS; if a cycle is found it is included in all output shapes and the command exits 1 after printing. 5. **Emit output** — compact JSON to *stdout* when ``--format json``, or human-readable text otherwise. Filtering --------- ``--active-only`` prunes the graph to nodes whose reservation ID appears in the active-reservation set. This is useful when the dependency dir has accumulated records for long-expired reservations that are no longer relevant. ``--topo`` adds a TOPO column showing dependency-first execution order. Without ``--topo`` a simpler flat table is shown (STATUS / ID / DEPENDS-ON only). Security -------- All reservation IDs and run-IDs from persisted files are passed through :func:`~muse.core.validation.sanitize_display` before printing to prevent ANSI injection. ``--reservation-id`` is validated as a UUID before any path construction or file I/O. Performance ----------- O(V + E) where V is the number of dependency records and E is the total number of dependency edges. Active-reservation loading is O(r) where r is the number of reservation files. Args: args: Parsed ``argparse.Namespace`` with attributes ``reservation_id``, ``topo``, ``active_only``, and ``fmt``. Exit codes: 0 — success (blocked nodes or empty graph are still success). 1 — cycle detected, bad arguments, or unexpected error loading state. """ as_json = args.fmt == "json" reservation_id: str | None = args.reservation_id active_only: bool = args.active_only # ── Input validation (before any file I/O) ──────────────────────────────── if reservation_id is not None: try: _validate_reservation_id(reservation_id) except ValueError as exc: msg = str(exc) if as_json: print(json.dumps({"error": msg, "status": "bad_reservation_id"})) else: print(f"❌ --reservation-id: {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() all_deps = load_all_dependencies(root) graph = load_dag(root) # Derive the set of currently active reservation IDs. try: active = active_reservations(root) except Exception as exc: # noqa: BLE001 if as_json: print(json.dumps({"error": str(exc)})) else: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) active_ids: frozenset[str] = frozenset(r.reservation_id for r in active) # ── Active-only filter ──────────────────────────────────────────────────── if active_only: graph = {k: v for k, v in graph.items() if k in active_ids} # ── Cycle detection ─────────────────────────────────────────────────────── cycle = detect_cycle(graph) # ── Single-reservation mode ─────────────────────────────────────────────── if reservation_id is not None: _run_single( reservation_id, graph, active_ids, cycle, as_json ) return # ── Full DAG mode ───────────────────────────────────────────────────────── _run_full(graph, active_ids, cycle, all_deps, as_json, args.topo, active_only) if cycle is not None: raise SystemExit(ExitCode.USER_ERROR) # ── Output helpers ───────────────────────────────────────────────────────────── def _run_single( reservation_id: str, graph: _Graph, active_ids: frozenset[str], cycle: list[str] | None, as_json: bool, ) -> None: """Emit dependency details for a single reservation. Args: reservation_id: The UUID to look up in the graph. graph: Adjacency map — node → set of nodes it depends on. active_ids: Reservation IDs whose reservation file is still active. cycle: Cycle path returned by :func:`~muse.core.dag.detect_cycle`, or ``None`` if the graph is acyclic. as_json: Emit compact JSON when ``True``; human-readable text otherwise. """ deps = sorted(graph.get(reservation_id, set())) blocked = is_blocked(reservation_id, graph, active_ids) blocking = get_blocking(reservation_id, graph, active_ids) if as_json: print(json.dumps({ "reservation_id": reservation_id, "depends_on": deps, "active": reservation_id in active_ids, "blocked": blocked, "blocking": blocking, "cycle": cycle, })) return rid_short = sanitize_display(reservation_id[:8]) status = "BLOCKED" if blocked else "unblocked" print(f"\nReservation {rid_short}… [{status}]") if deps: print(f" depends_on ({len(deps)}):") for dep in deps: active_marker = " ← ACTIVE" if dep in active_ids else "" print(f" {sanitize_display(dep[:8])}…{active_marker}") else: print(" no declared dependencies") if blocking: print(f"\n Blocking ({len(blocking)}):") for b in blocking: print(f" {sanitize_display(b[:8])}… (still active)") if cycle: print(f"\n ⚠️ Cycle detected: {' → '.join(sanitize_display(c[:8]) + '…' for c in cycle)}") def _run_full( graph: _Graph, active_ids: frozenset[str], cycle: list[str] | None, all_deps: list[DependencyRecord], as_json: bool, topo_mode: bool, active_only: bool, ) -> None: """Emit the full DAG listing. Args: graph: Adjacency map — node → set of nodes it depends on. May be pre-filtered to active-only nodes by the caller. active_ids: Reservation IDs whose reservation file is still active. cycle: Cycle path from :func:`~muse.core.dag.detect_cycle`, or ``None`` if acyclic. all_deps: Raw dependency records loaded from disk (used for metadata in future extensions; not currently used directly in output). as_json: Emit compact JSON when ``True``; human-readable text otherwise. topo_mode: When ``True``, include a TOPO column in text output and order rows by dependency-first execution order. active_only: Reflected in JSON output as ``active_only`` field so consumers know whether the graph was filtered. """ from muse._version import __version__ total_nodes = len(graph) total_edges = sum(len(v) for v in graph.values()) # Compute topological order (or handle cycle gracefully). if cycle is None: try: topo_order = topological_sort(graph) except ValueError: topo_order = sorted(graph) else: topo_order = sorted(graph) topo_index = {node: i + 1 for i, node in enumerate(topo_order)} # Build per-node records. nodes = [] for node in topo_order: deps = sorted(graph.get(node, set())) blocked = is_blocked(node, graph, active_ids) blocking = get_blocking(node, graph, active_ids) nodes.append({ "reservation_id": node, "depends_on": deps, "active": node in active_ids, "blocked": blocked, "blocking": blocking, "topo_index": topo_index.get(node), }) blocked_count = sum(1 for n in nodes if n["blocked"]) if as_json: print(json.dumps({ "schema_version": __version__, "total_nodes": total_nodes, "total_edges": total_edges, "blocked_count": blocked_count, "active_only": active_only, "cycle": cycle, "nodes": nodes, })) return # Text output. print(f"\nDependency DAG — {total_nodes} node(s), {total_edges} edge(s)") if cycle: cycle_str = " → ".join(sanitize_display(c[:8]) + "…" for c in cycle) print(f" ⚠️ CYCLE DETECTED: {cycle_str}") print(" No topological order is possible until the cycle is resolved.") else: print( f" {blocked_count} blocked " f"{total_nodes - blocked_count} unblocked" ) if not nodes: print("\n (no dependency records)") return print() if topo_mode: print(f"{'TOPO':>4} {'STATUS':<12} {'ID':8} DEPENDS-ON") print("─" * 72) for node_rec in nodes: idx = node_rec["topo_index"] or 0 status = "BLOCKED" if node_rec["blocked"] else "unblocked" rid = sanitize_display(node_rec["reservation_id"][:8]) + "…" if node_rec["depends_on"]: deps_str = ", ".join( sanitize_display(d[:8]) + "…" for d in node_rec["depends_on"] ) else: deps_str = "(no deps)" print(f"{idx:>4} {status:<12} {rid:<9} {deps_str}") else: print(f"{'STATUS':<12} {'ID':8} DEPENDS-ON") print("─" * 56) for node_rec in nodes: status = "BLOCKED" if node_rec["blocked"] else "unblocked" rid = sanitize_display(node_rec["reservation_id"][:8]) + "…" if node_rec["depends_on"]: deps_str = ", ".join( sanitize_display(d[:8]) + "…" for d in node_rec["depends_on"] ) else: deps_str = "(no deps)" print(f"{status:<12} {rid:<9} {deps_str}") if cycle: print(f"\n ⚠️ Resolve the cycle before any blocked agent can proceed.")