aaronrene / gabriel-muse public
dag.py python
448 lines 16.0 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago
1 """``muse coord dag`` — inspect the coordination dependency DAG.
2
3 Shows which reservations are waiting for others to complete, reports blocked
4 status, detects cycles, and computes the correct execution order via
5 topological sort.
6
7 Overview
8 --------
9 When multiple agents reserve addresses in parallel, some agents may declare
10 that their work depends on another agent's reservation being released first.
11 The dependency DAG captures these ordering constraints.
12
13 ``muse coord dag`` reads all dependency records from
14 ``.muse/coordination/dependencies/``, cross-references the currently active
15 reservations, and answers three questions:
16
17 1. **Who is blocked?** — which reservations cannot start because a dependency
18 is still active.
19 2. **What order should work proceed?** — topological sort respecting all
20 declared dependencies.
21 3. **Are there any cycles?** — flag immediately; cycles mean no agent can ever
22 become unblocked.
23
24 Output examples
25 ---------------
26 Text (full DAG, default)::
27
28 Dependency DAG — 4 node(s), 3 edge(s)
29 1 blocked 3 unblocked 0 cycles
30
31 TOPO STATUS ID DEPENDS-ON
32 ────────────────────────────────────────────────────────────────────
33 1 unblocked a1b2c3d4 (no deps)
34 2 unblocked e5f6a7b8 (no deps)
35 3 BLOCKED c9d0e1f2 a1b2c3d4
36 4 unblocked 01234567 e5f6a7b8
37
38 Text (flat, no topo column, default without --topo)::
39
40 STATUS ID DEPENDS-ON
41 ────────────────────────────────────────────────────────────────────
42 unblocked a1b2c3d4 (no deps)
43 BLOCKED c9d0e1f2 a1b2c3d4
44
45 JSON::
46
47 {
48 "schema_version": "...",
49 "total_nodes": 4,
50 "total_edges": 3,
51 "blocked_count": 1,
52 "active_only": false,
53 "cycle": null,
54 "nodes": [
55 {
56 "reservation_id": "c9d0e1f2-...",
57 "depends_on": ["a1b2c3d4-..."],
58 "active": true,
59 "blocked": true,
60 "blocking": ["a1b2c3d4-..."],
61 "topo_index": 3
62 },
63 ...
64 ]
65 }
66
67 Exit codes::
68
69 0 — success (even when blocked nodes exist)
70 1 — cycle detected, bad arguments, or unexpected error
71 """
72
73 from __future__ import annotations
74
75 import argparse
76 import json
77 import sys
78
79 from muse.core.coordination import _validate_reservation_id, active_reservations
80 from muse.core.dag import (
81 DependencyRecord,
82 detect_cycle,
83 get_blocking,
84 is_blocked,
85 load_all_dependencies,
86 load_dag,
87 topological_sort,
88 )
89 from muse.core.errors import ExitCode
90 from muse.core.repo import require_repo
91 from muse.core.validation import sanitize_display
92
93
94
95 type _Graph = dict[str, set[str]]
96 def register(
97 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
98 ) -> None:
99 """Register ``dag`` on *subparsers* (under ``muse coord``).
100
101 Wires all flags with their defaults, choices, and help text so that
102 ``--help`` output is accurate. Sets ``func`` to :func:`run`.
103
104 Flags registered
105 ----------------
106 ``--reservation-id UUID``
107 Show dependency details for a single reservation. Must be a valid
108 UUID when provided.
109 ``--topo``
110 Print nodes in topological execution order with an explicit TOPO
111 column. Without this flag, a simpler flat table (no TOPO column)
112 is shown.
113 ``--active-only``
114 Restrict the graph to nodes that correspond to currently active
115 reservations. Nodes whose reservation has already expired or been
116 released are hidden.
117 ``--format`` / ``--json``
118 Emit compact JSON to stdout; default is human-readable text.
119 """
120 parser = subparsers.add_parser(
121 "dag",
122 help="Inspect the coordination dependency DAG.",
123 formatter_class=argparse.RawDescriptionHelpFormatter,
124 description=__doc__,
125 )
126 parser.add_argument(
127 "--reservation-id",
128 dest="reservation_id",
129 default=None,
130 metavar="UUID",
131 help=(
132 "Show dependency details for a single reservation ID. "
133 "Must be a valid UUID. When omitted the full DAG is shown."
134 ),
135 )
136 parser.add_argument(
137 "--topo",
138 action="store_true",
139 default=False,
140 help=(
141 "Print nodes in topological execution order with a TOPO column. "
142 "Without this flag, a simpler flat table is shown."
143 ),
144 )
145 parser.add_argument(
146 "--active-only",
147 action="store_true",
148 default=False,
149 dest="active_only",
150 help=(
151 "Restrict output to nodes whose reservation is currently active. "
152 "Expired and released reservations are hidden."
153 ),
154 )
155 parser.add_argument(
156 "--format", "-f",
157 default="text",
158 dest="fmt",
159 choices=("text", "json"),
160 help="Output format: text (default) or json.",
161 )
162 parser.add_argument(
163 "--json",
164 action="store_const",
165 const="json",
166 dest="fmt",
167 help="Shorthand for --format json.",
168 )
169 parser.set_defaults(func=run)
170
171
172 def run(args: argparse.Namespace) -> None:
173 """Inspect the coordination dependency DAG.
174
175 Loads all dependency records and active reservations, derives blocked
176 status for each node, and optionally computes topological order.
177
178 Execution order
179 ---------------
180 1. **Validate inputs** — ``--reservation-id``, when provided, must be a
181 valid UUID. Validation fires before any file I/O; failure exits
182 :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message on
183 *stderr* (or compact JSON when ``--format json``).
184 2. **Resolve repo** — calls :func:`~muse.core.repo.require_repo`.
185 3. **Load graph** — reads all dependency records and active reservations
186 from ``.muse/coordination/``.
187 4. **Detect cycle** — O(V + E) DFS; if a cycle is found it is included in
188 all output shapes and the command exits 1 after printing.
189 5. **Emit output** — compact JSON to *stdout* when ``--format json``, or
190 human-readable text otherwise.
191
192 Filtering
193 ---------
194 ``--active-only`` prunes the graph to nodes whose reservation ID appears
195 in the active-reservation set. This is useful when the dependency dir
196 has accumulated records for long-expired reservations that are no longer
197 relevant.
198
199 ``--topo`` adds a TOPO column showing dependency-first execution order.
200 Without ``--topo`` a simpler flat table is shown (STATUS / ID / DEPENDS-ON
201 only).
202
203 Security
204 --------
205 All reservation IDs and run-IDs from persisted files are passed through
206 :func:`~muse.core.validation.sanitize_display` before printing to prevent
207 ANSI injection. ``--reservation-id`` is validated as a UUID before any
208 path construction or file I/O.
209
210 Performance
211 -----------
212 O(V + E) where V is the number of dependency records and E is the total
213 number of dependency edges. Active-reservation loading is O(r) where r is
214 the number of reservation files.
215
216 Args:
217 args: Parsed ``argparse.Namespace`` with attributes ``reservation_id``,
218 ``topo``, ``active_only``, and ``fmt``.
219
220 Exit codes:
221 0 — success (blocked nodes or empty graph are still success).
222 1 — cycle detected, bad arguments, or unexpected error loading state.
223 """
224 as_json = args.fmt == "json"
225 reservation_id: str | None = args.reservation_id
226 active_only: bool = args.active_only
227
228 # ── Input validation (before any file I/O) ────────────────────────────────
229
230 if reservation_id is not None:
231 try:
232 _validate_reservation_id(reservation_id)
233 except ValueError as exc:
234 msg = str(exc)
235 if as_json:
236 print(json.dumps({"error": msg, "status": "bad_reservation_id"}))
237 else:
238 print(f"❌ --reservation-id: {msg}", file=sys.stderr)
239 raise SystemExit(ExitCode.USER_ERROR)
240
241 root = require_repo()
242
243 all_deps = load_all_dependencies(root)
244 graph = load_dag(root)
245
246 # Derive the set of currently active reservation IDs.
247 try:
248 active = active_reservations(root)
249 except Exception as exc: # noqa: BLE001
250 if as_json:
251 print(json.dumps({"error": str(exc)}))
252 else:
253 print(f"❌ {exc}", file=sys.stderr)
254 raise SystemExit(ExitCode.USER_ERROR)
255
256 active_ids: frozenset[str] = frozenset(r.reservation_id for r in active)
257
258 # ── Active-only filter ────────────────────────────────────────────────────
259 if active_only:
260 graph = {k: v for k, v in graph.items() if k in active_ids}
261
262 # ── Cycle detection ───────────────────────────────────────────────────────
263 cycle = detect_cycle(graph)
264
265 # ── Single-reservation mode ───────────────────────────────────────────────
266 if reservation_id is not None:
267 _run_single(
268 reservation_id, graph, active_ids, cycle, as_json
269 )
270 return
271
272 # ── Full DAG mode ─────────────────────────────────────────────────────────
273 _run_full(graph, active_ids, cycle, all_deps, as_json, args.topo, active_only)
274
275 if cycle is not None:
276 raise SystemExit(ExitCode.USER_ERROR)
277
278
279 # ── Output helpers ─────────────────────────────────────────────────────────────
280
281
282 def _run_single(
283 reservation_id: str,
284 graph: _Graph,
285 active_ids: frozenset[str],
286 cycle: list[str] | None,
287 as_json: bool,
288 ) -> None:
289 """Emit dependency details for a single reservation.
290
291 Args:
292 reservation_id: The UUID to look up in the graph.
293 graph: Adjacency map — node → set of nodes it depends on.
294 active_ids: Reservation IDs whose reservation file is still active.
295 cycle: Cycle path returned by :func:`~muse.core.dag.detect_cycle`,
296 or ``None`` if the graph is acyclic.
297 as_json: Emit compact JSON when ``True``; human-readable text otherwise.
298 """
299 deps = sorted(graph.get(reservation_id, set()))
300 blocked = is_blocked(reservation_id, graph, active_ids)
301 blocking = get_blocking(reservation_id, graph, active_ids)
302
303 if as_json:
304 print(json.dumps({
305 "reservation_id": reservation_id,
306 "depends_on": deps,
307 "active": reservation_id in active_ids,
308 "blocked": blocked,
309 "blocking": blocking,
310 "cycle": cycle,
311 }))
312 return
313
314 rid_short = sanitize_display(reservation_id[:8])
315 status = "BLOCKED" if blocked else "unblocked"
316 print(f"\nReservation {rid_short}… [{status}]")
317 if deps:
318 print(f" depends_on ({len(deps)}):")
319 for dep in deps:
320 active_marker = " ← ACTIVE" if dep in active_ids else ""
321 print(f" {sanitize_display(dep[:8])}…{active_marker}")
322 else:
323 print(" no declared dependencies")
324 if blocking:
325 print(f"\n Blocking ({len(blocking)}):")
326 for b in blocking:
327 print(f" {sanitize_display(b[:8])}… (still active)")
328 if cycle:
329 print(f"\n ⚠️ Cycle detected: {' → '.join(sanitize_display(c[:8]) + '…' for c in cycle)}")
330
331
332 def _run_full(
333 graph: _Graph,
334 active_ids: frozenset[str],
335 cycle: list[str] | None,
336 all_deps: list[DependencyRecord],
337 as_json: bool,
338 topo_mode: bool,
339 active_only: bool,
340 ) -> None:
341 """Emit the full DAG listing.
342
343 Args:
344 graph: Adjacency map — node → set of nodes it depends on. May be
345 pre-filtered to active-only nodes by the caller.
346 active_ids: Reservation IDs whose reservation file is still active.
347 cycle: Cycle path from :func:`~muse.core.dag.detect_cycle`, or
348 ``None`` if acyclic.
349 all_deps: Raw dependency records loaded from disk (used for metadata
350 in future extensions; not currently used directly in output).
351 as_json: Emit compact JSON when ``True``; human-readable text otherwise.
352 topo_mode: When ``True``, include a TOPO column in text output and
353 order rows by dependency-first execution order.
354 active_only: Reflected in JSON output as ``active_only`` field so
355 consumers know whether the graph was filtered.
356 """
357 from muse._version import __version__
358
359 total_nodes = len(graph)
360 total_edges = sum(len(v) for v in graph.values())
361
362 # Compute topological order (or handle cycle gracefully).
363 if cycle is None:
364 try:
365 topo_order = topological_sort(graph)
366 except ValueError:
367 topo_order = sorted(graph)
368 else:
369 topo_order = sorted(graph)
370
371 topo_index = {node: i + 1 for i, node in enumerate(topo_order)}
372
373 # Build per-node records.
374 nodes = []
375 for node in topo_order:
376 deps = sorted(graph.get(node, set()))
377 blocked = is_blocked(node, graph, active_ids)
378 blocking = get_blocking(node, graph, active_ids)
379 nodes.append({
380 "reservation_id": node,
381 "depends_on": deps,
382 "active": node in active_ids,
383 "blocked": blocked,
384 "blocking": blocking,
385 "topo_index": topo_index.get(node),
386 })
387
388 blocked_count = sum(1 for n in nodes if n["blocked"])
389
390 if as_json:
391 print(json.dumps({
392 "schema_version": __version__,
393 "total_nodes": total_nodes,
394 "total_edges": total_edges,
395 "blocked_count": blocked_count,
396 "active_only": active_only,
397 "cycle": cycle,
398 "nodes": nodes,
399 }))
400 return
401
402 # Text output.
403 print(f"\nDependency DAG — {total_nodes} node(s), {total_edges} edge(s)")
404 if cycle:
405 cycle_str = " → ".join(sanitize_display(c[:8]) + "…" for c in cycle)
406 print(f" ⚠️ CYCLE DETECTED: {cycle_str}")
407 print(" No topological order is possible until the cycle is resolved.")
408 else:
409 print(
410 f" {blocked_count} blocked "
411 f"{total_nodes - blocked_count} unblocked"
412 )
413
414 if not nodes:
415 print("\n (no dependency records)")
416 return
417
418 print()
419 if topo_mode:
420 print(f"{'TOPO':>4} {'STATUS':<12} {'ID':8} DEPENDS-ON")
421 print("─" * 72)
422 for node_rec in nodes:
423 idx = node_rec["topo_index"] or 0
424 status = "BLOCKED" if node_rec["blocked"] else "unblocked"
425 rid = sanitize_display(node_rec["reservation_id"][:8]) + "…"
426 if node_rec["depends_on"]:
427 deps_str = ", ".join(
428 sanitize_display(d[:8]) + "…" for d in node_rec["depends_on"]
429 )
430 else:
431 deps_str = "(no deps)"
432 print(f"{idx:>4} {status:<12} {rid:<9} {deps_str}")
433 else:
434 print(f"{'STATUS':<12} {'ID':8} DEPENDS-ON")
435 print("─" * 56)
436 for node_rec in nodes:
437 status = "BLOCKED" if node_rec["blocked"] else "unblocked"
438 rid = sanitize_display(node_rec["reservation_id"][:8]) + "…"
439 if node_rec["depends_on"]:
440 deps_str = ", ".join(
441 sanitize_display(d[:8]) + "…" for d in node_rec["depends_on"]
442 )
443 else:
444 deps_str = "(no deps)"
445 print(f"{status:<12} {rid:<9} {deps_str}")
446
447 if cycle:
448 print(f"\n ⚠️ Resolve the cycle before any blocked agent can proceed.")
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago