"""``muse coord list`` — display the live coordination state of the swarm. Shows every reservation and intent currently stored in ``.muse/coordination/``, with optional filtering, sorting, and structured JSON output for agents. This is the primary observability primitive for a running swarm: without it, the coordination layer is a write-only black box. Without ``--all`` only *active* (non-expired) reservations are shown; intents are always shown because they carry no TTL. Usage:: muse coord list # active reservations + intents muse coord list --all # include expired reservations muse coord list --kind reservations # reservations only muse coord list --kind intents # intents only muse coord list --run-id agent-42 # filter by agent muse coord list --branch feat/billing # filter by branch (exact) muse coord list --address "billing.py::*"# fnmatch glob on addresses muse coord list --op rename # filter by operation type muse coord list --limit 20 # cap output to 20 of each kind muse coord list --summary # one-line count summary muse coord list --format json # machine-readable JSON muse coord list --json # shorthand for --format json JSON schema:: { "active_reservations": int, "expired_reservations": int, "released_reservations": int, "total_reservations_shown": int, "total_intents_shown": int, "has_conflicts": bool, "filters": { "run_id": str | null, "branch": str | null, "address_glob": str | null, "operation": str | null, "include_expired": bool, "kind": "all" | "reservations" | "intents", "limit": int | null }, "reservations": [ { "reservation_id": str, "run_id": str, "branch": str, "addresses": [str, ...], "created_at": str, // ISO 8601 "expires_at": str, // ISO 8601 (original TTL) "effective_expires_at": str, // ISO 8601 (max of expires_at and heartbeat) "ttl_remaining_seconds": float, // negative when expired "operation": str | null, "is_active": bool, "released": bool, // true when a release tombstone exists "conflict_count": int // other active reservations sharing ≥1 address }, ... ], "intents": [ { "intent_id": str, "reservation_id": str, "run_id": str, "branch": str, "addresses": [str, ...], "operation": str, "created_at": str, "detail": str }, ... ], "elapsed_seconds": float } Exit codes:: 0 — success (zero records is still success) 1 — bad arguments """ from __future__ import annotations import argparse import datetime import json import sys import time from typing import TypedDict from muse.core.coordination import ( Reservation, filter_intents, filter_reservations, load_all_intents, load_all_reservations, load_heartbeat_map, load_released_ids, ) from muse.core.repo import require_repo from muse.core.validation import sanitize_display type _ResIdMap = dict[str, set[str]] # ── Typed JSON schema ───────────────────────────────────────────────────────── class _ReservationEntry(TypedDict): reservation_id: str run_id: str branch: str addresses: list[str] created_at: str expires_at: str effective_expires_at: str ttl_remaining_seconds: float operation: str | None is_active: bool released: bool conflict_count: int class _IntentEntry(TypedDict): intent_id: str reservation_id: str run_id: str branch: str addresses: list[str] operation: str created_at: str detail: str class _FiltersEntry(TypedDict): run_id: str | None branch: str | None address_glob: str | None operation: str | None include_expired: bool kind: str limit: int | None class _ListJson(TypedDict): active_reservations: int expired_reservations: int released_reservations: int total_reservations_shown: int total_intents_shown: int has_conflicts: bool filters: _FiltersEntry reservations: list[_ReservationEntry] intents: list[_IntentEntry] elapsed_seconds: float # ── Helpers ─────────────────────────────────────────────────────────────────── def _format_ttl(seconds: float) -> str: """Format a TTL in seconds as a human-readable string. Examples -------- ``3661.0`` → ``"1h 1m 1s"`` ``90.0`` → ``"1m 30s"`` ``45.0`` → ``"45s"`` ``0.0`` → ``"EXPIRED"`` ``-10.0`` → ``"EXPIRED"`` """ if seconds <= 0: return "EXPIRED" total = int(seconds) hours, remainder = divmod(total, 3600) minutes, secs = divmod(remainder, 60) if hours: return f"{hours}h {minutes}m {secs}s" if minutes: return f"{minutes}m {secs}s" return f"{secs}s" # ── CLI registration ────────────────────────────────────────────────────────── def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``list`` subcommand on *subparsers* (under ``muse coord``).""" parser = subparsers.add_parser( "list", help="Display the current coordination state of the swarm.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--all", "-a", action="store_true", dest="include_expired", help="Include expired reservations (default: active only).", ) parser.add_argument( "--kind", default="all", choices=("all", "reservations", "intents"), metavar="KIND", help="What to show: all (default), reservations, or intents.", ) parser.add_argument( "--run-id", default=None, dest="run_id", metavar="ID", help="Show only records whose run-id exactly matches ID.", ) parser.add_argument( "--branch", "-b", default=None, dest="branch", metavar="BRANCH", help="Show only records on this branch (exact match).", ) parser.add_argument( "--address", default=None, dest="address_glob", metavar="GLOB", help=( "Show only records where at least one address matches this " "fnmatch glob (e.g. \"billing.py::*\")." ), ) parser.add_argument( "--op", default=None, dest="operation", metavar="OPERATION", help=( "Show only records with this operation type " "(e.g. rename, modify, delete, extract, move, inline, split, merge)." ), ) parser.add_argument( "--limit", "-n", type=int, default=None, dest="limit", metavar="N", help="Cap output to N reservations and N intents (useful for large swarms).", ) parser.add_argument( "--summary", action="store_true", help="Print a single-line count summary and exit.", ) 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) # ── Command implementation ──────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Display the live coordination state of the swarm. Loads all reservation and intent records from ``.muse/coordination/``, applies the requested filters in memory (no additional I/O per record after the initial directory scan), and formats the result for human or agent consumption. Filtering --------- ``--run-id``, ``--branch``, and ``--op`` are exact-match filters. ``--address`` uses :mod:`fnmatch` glob matching against each address in the record — a record is included when *any* of its addresses match the glob. All filters compose with AND semantics. Conflict detection ------------------ Each reservation entry in JSON output includes ``conflict_count``: the number of *other* active reservations that share at least one address. The top-level ``has_conflicts`` flag is ``True`` when any active reservation has ``conflict_count > 0``. This gives agents a lightweight conflict signal without running ``muse forecast``. Security -------- All strings sourced from persisted records (``run_id``, ``branch``, ``detail``, addresses) are passed through :func:`~muse.core.validation.sanitize_display` before printing to strip ANSI escape sequences and control characters that could corrupt a terminal. ``--address`` glob matching is a pure string operation — no filesystem access occurs. Performance ----------- A single ``datetime.now()`` is captured at the start of the run and reused for all TTL calculations, ensuring consistent timestamps across all entries in a single invocation. Records are loaded with one directory glob per category, then filtered in one pass. JSON output ----------- Includes ``elapsed_seconds`` and a ``filters`` object so agents can verify which filters were active. ``ttl_remaining_seconds`` is a float (negative when the reservation is expired) so agents can compute urgency without parsing ISO 8601 strings. ``conflict_count`` per entry and ``has_conflicts`` at the top level provide a quick conflict signal. """ t0 = time.monotonic() include_expired: bool = args.include_expired kind: str = args.kind run_id: str | None = args.run_id branch: str | None = args.branch address_glob: str | None = args.address_glob operation: str | None = args.operation limit: int | None = args.limit summary_only: bool = args.summary fmt: str = args.fmt as_json = fmt == "json" root = require_repo() # ── Load ────────────────────────────────────────────────────────────────── all_reservations = load_all_reservations(root) all_intents = load_all_intents(root) released_ids = load_released_ids(root) hb_map = load_heartbeat_map(root) # Pre-compute heartbeat effective-expiry mapping for filter_reservations. heartbeat_expires = {rid: hb.extended_expires_at for rid, hb in hb_map.items()} # Capture now once — reused for all TTL calculations so timestamps are # consistent across every entry in this invocation. now_utc = datetime.datetime.now(datetime.timezone.utc) def _effective_expires(r: Reservation) -> datetime.datetime: hb = hb_map.get(r.reservation_id) return max(r.expires_at, hb.extended_expires_at) if hb is not None else r.expires_at def _effective_ttl(r: Reservation) -> float: return (_effective_expires(r) - now_utc).total_seconds() # Count totals before filtering for summary bookkeeping. total_active = sum( 1 for r in all_reservations if r.reservation_id not in released_ids and now_utc < _effective_expires(r) ) total_expired = len(all_reservations) - total_active # ── Filter ──────────────────────────────────────────────────────────────── reservations = filter_reservations( all_reservations, run_id=run_id, branch=branch, address_glob=address_glob, operation=operation, include_expired=include_expired, released_ids=released_ids, heartbeat_expires=heartbeat_expires, ) intents = filter_intents( all_intents, run_id=run_id, branch=branch, address_glob=address_glob, operation=operation, ) # Sort both lists by created_at ascending (oldest first). reservations.sort(key=lambda r: r.created_at) intents.sort(key=lambda i: i.created_at) # Apply limit after sorting so the N oldest are returned. if limit is not None and limit > 0: reservations = reservations[:limit] intents = intents[:limit] # ── Conflict detection ──────────────────────────────────────────────────── # Build address → {reservation_id} map for the active (post-filter) set. # This lets each reservation report how many peers share ≥1 address. active_res_ids = { r.reservation_id for r in reservations if _effective_ttl(r) > 0 and r.reservation_id not in released_ids } addr_to_res_ids: _ResIdMap = {} for r in reservations: if r.reservation_id in active_res_ids: for addr in r.addresses: addr_to_res_ids.setdefault(addr, set()).add(r.reservation_id) def _conflict_count(r: Reservation) -> int: """Count distinct active reservations sharing ≥1 address with *r*.""" conflicting: set[str] = set() for addr in r.addresses: for rid in addr_to_res_ids.get(addr, set()): if rid != r.reservation_id: conflicting.add(rid) return len(conflicting) elapsed = round(time.monotonic() - t0, 4) # ── Summary mode ────────────────────────────────────────────────────────── if summary_only: show_res = kind in ("all", "reservations") show_int = kind in ("all", "intents") parts: list[str] = [] if show_res: parts.append(f"{len(reservations)} reservation(s)") if show_int: parts.append(f"{len(intents)} intent(s)") suffix = "" if not include_expired and total_expired: suffix = f" [{total_expired} expired hidden — use --all to show]" if not parts or (len(reservations) == 0 and len(intents) == 0 and not (show_res or show_int)): print("No coordination records.") else: print(", ".join(parts) + suffix) return # ── JSON output ─────────────────────────────────────────────────────────── if as_json: res_entries: list[_ReservationEntry] = [] for r in reservations: if kind in ("all", "reservations"): eff_exp = _effective_expires(r) eff_ttl = _effective_ttl(r) res_entries.append(_ReservationEntry( reservation_id=r.reservation_id, run_id=r.run_id, branch=r.branch, addresses=r.addresses, created_at=r.created_at.isoformat(), expires_at=r.expires_at.isoformat(), effective_expires_at=eff_exp.isoformat(), ttl_remaining_seconds=round(eff_ttl, 2), operation=r.operation, is_active=eff_ttl > 0 and r.reservation_id not in released_ids, released=r.reservation_id in released_ids, conflict_count=_conflict_count(r), )) int_entries: list[_IntentEntry] = [] for i in intents: if kind in ("all", "intents"): int_entries.append(_IntentEntry( intent_id=i.intent_id, reservation_id=i.reservation_id, run_id=i.run_id, branch=i.branch, addresses=i.addresses, operation=i.operation, created_at=i.created_at.isoformat(), detail=i.detail, )) total_released = len(released_ids) has_conflicts = any(e["conflict_count"] > 0 for e in res_entries) payload = _ListJson( active_reservations=total_active, expired_reservations=total_expired, released_reservations=total_released, total_reservations_shown=len(res_entries), total_intents_shown=len(int_entries), has_conflicts=has_conflicts, filters=_FiltersEntry( run_id=run_id, branch=branch, address_glob=address_glob, operation=operation, include_expired=include_expired, kind=kind, limit=limit, ), reservations=res_entries, intents=int_entries, elapsed_seconds=elapsed, ) print(json.dumps(payload, indent=2)) return # ── Text output ─────────────────────────────────────────────────────────── show_res = kind in ("all", "reservations") show_int = kind in ("all", "intents") n_res = len(reservations) if show_res else 0 n_int = len(intents) if show_int else 0 header = f"\nCoordination state" parts = [] if show_res: parts.append(f"{n_res} reservation(s)") if show_int: parts.append(f"{n_int} intent(s)") if parts: header += " — " + ", ".join(parts) if not include_expired and total_expired and show_res: header += f" [{total_expired} expired hidden]" print(header) print("─" * 62) if n_res == 0 and n_int == 0: print("\n (no coordination records") if not include_expired and total_expired: print(f" {total_expired} expired — run with --all to show them)") else: print(" run 'muse coord reserve' to create one)") return # Reservations block. if show_res and reservations: print("\nReservations") for res in reservations: is_released = res.reservation_id in released_ids eff_ttl = _effective_ttl(res) ttl_str = _format_ttl(eff_ttl) if is_released: status_label = "RELEASED" elif eff_ttl > 0: status_label = "expires in" else: status_label = "" op_str = f" op={sanitize_display(res.operation)}" if res.operation else "" run_label = sanitize_display(res.run_id) branch_label = sanitize_display(res.branch) res_id_short = res.reservation_id[:8] conflicts = _conflict_count(res) conflict_str = f" ⚠ {conflicts} conflict(s)" if conflicts > 0 else "" print( f" [{res_id_short}] {run_label:<20} {branch_label:<22}" f"{op_str} {status_label} {ttl_str}{conflict_str}" ) for addr in res.addresses: print(f" {sanitize_display(addr)}") # Intents block. if show_int and intents: print("\nIntents") for intent in intents: run_label = sanitize_display(intent.run_id) branch_label = sanitize_display(intent.branch) op_label = sanitize_display(intent.operation) intent_id_short = intent.intent_id[:8] print(f" [{intent_id_short}] {run_label:<20} {branch_label:<22} {op_label}") for addr in intent.addresses: print(f" {sanitize_display(addr)}") if intent.detail: print(f" → {sanitize_display(intent.detail)}") print(f"\n ({elapsed:.3f}s)")