watch_coord.py
python
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8
docs(KD-STAGING): sync governance after KD-6b DONE
Human
12 days ago
| 1 | """``muse coord watch`` — stream coordination events in real time. |
| 2 | |
| 3 | Watches ``.muse/coordination/`` for file-system changes and emits a structured |
| 4 | event for every coordination record that is added, modified, removed, or |
| 5 | expired. On macOS/BSD the kqueue kernel interface is used, so the process |
| 6 | sleeps in the kernel and wakes **only** when a file actually changes — zero |
| 7 | CPU overhead between events. On Linux and other platforms a configurable |
| 8 | polling fallback is used instead. |
| 9 | |
| 10 | This replaces the polling anti-pattern where agents repeatedly call |
| 11 | ``muse coord list`` to detect swarm state changes. Piping |
| 12 | ``muse coord watch --json`` gives an agent an event-driven coordination bus. |
| 13 | |
| 14 | Event types |
| 15 | ----------- |
| 16 | ``snapshot`` |
| 17 | Emitted once per existing record immediately after startup, before any |
| 18 | change events. Agents should consume all ``snapshot`` events first to |
| 19 | build an accurate initial picture of the swarm, then react to deltas. |
| 20 | |
| 21 | ``added`` |
| 22 | A new coordination file appeared on disk (e.g. another agent just called |
| 23 | ``muse coord reserve``). |
| 24 | |
| 25 | ``modified`` |
| 26 | An existing file changed (e.g. a heartbeat that extended a reservation's |
| 27 | TTL, or an intent that was amended). |
| 28 | |
| 29 | ``removed`` |
| 30 | A file was deleted from disk (e.g. GC cleaned up an expired reservation |
| 31 | with ``muse coord gc --execute``). The event carries the last-known data |
| 32 | for the record so filters still work correctly. |
| 33 | |
| 34 | ``expired`` |
| 35 | A reservation that was previously active has crossed its effective |
| 36 | expiry timestamp. The file may still exist on disk (not GC'd yet) — |
| 37 | this event fires when ``active_reservations()`` stops returning it. |
| 38 | |
| 39 | Event kinds |
| 40 | ----------- |
| 41 | ``reservation`` — ``.muse/coordination/reservations/<uuid>.json`` |
| 42 | ``intent`` — ``.muse/coordination/intents/<uuid>.json`` |
| 43 | ``release`` — ``.muse/coordination/releases/<uuid>.json`` |
| 44 | ``heartbeat`` — ``.muse/coordination/heartbeats/<uuid>.json`` |
| 45 | |
| 46 | Usage:: |
| 47 | |
| 48 | muse coord watch # stream all events, run forever |
| 49 | muse coord watch --once # emit current state as snapshots, then exit |
| 50 | muse coord watch --timeout 60 # run for 60 seconds then exit |
| 51 | muse coord watch --max-events 10 # exit after receiving 10 events |
| 52 | muse coord watch --kind reservation # only reservation events |
| 53 | muse coord watch --run-id agent-42 # only events from agent-42 |
| 54 | muse coord watch --branch feature/x # only events on feature/x |
| 55 | muse coord watch --json # NDJSON output (one event per line) |
| 56 | muse coord watch --poll-interval 0.5 # faster fallback poll cadence |
| 57 | |
| 58 | JSON output schema (NDJSON — one JSON object per line):: |
| 59 | |
| 60 | { |
| 61 | "schema_version": str, |
| 62 | "event_type": "snapshot" | "added" | "modified" | "removed" | "expired", |
| 63 | "kind": "reservation" | "intent" | "release" | "heartbeat", |
| 64 | "id": str, // UUID of the changed record |
| 65 | "timestamp": str, // ISO 8601 UTC — when the event was generated |
| 66 | "data": dict | {} // parsed record content; empty for "removed" |
| 67 | // if the cache had no prior data |
| 68 | } |
| 69 | |
| 70 | Text output example:: |
| 71 | |
| 72 | muse coord watch — watching .muse/coordination/ (kqueue) |
| 73 | ────────────────────────────────────────────────────────────── |
| 74 | · reservation agent-42@feature/billing a1b2c3d4 [billing.py::compute_total] |
| 75 | + reservation agent-99@main e5f6a7b8 [auth.py::verify_token] |
| 76 | ~ heartbeat agent-42 a1b2c3d4 extends to 2026-03-30T15:00Z |
| 77 | ✓ release agent-42 a1b2c3d4 completed (res: a1b2c3d4) |
| 78 | ! expired agent-99@main e5f6a7b8 [auth.py::verify_token] |
| 79 | |
| 80 | Flags:: |
| 81 | |
| 82 | --once Emit current state as snapshot events, then exit. |
| 83 | --timeout SECONDS Stop watching after N seconds (>= 0; 0 = --once). |
| 84 | --max-events N Exit after emitting N events (>= 1). Useful for |
| 85 | batch processing without a time-based timeout. |
| 86 | --poll-interval SECS Polling/kqueue-timeout interval in seconds |
| 87 | (default 1.0; range [0.01, 3600]). |
| 88 | --run-id RUNID Only emit events where data.run_id == RUNID. |
| 89 | Maximum 256 characters. |
| 90 | --branch BRANCH Only emit events where data.branch == BRANCH. |
| 91 | --kind KIND Only emit events for this kind |
| 92 | (reservation | intent | release | heartbeat). |
| 93 | --format / -f Output format: text (default) or json. |
| 94 | --json Shorthand for --format json. |
| 95 | |
| 96 | Security:: |
| 97 | |
| 98 | All agent-supplied strings are passed through |
| 99 | :func:`~muse.core.validation.sanitize_display` before printing to strip |
| 100 | ANSI escape sequences and terminal control characters. |
| 101 | |
| 102 | The ``--kind`` argument is validated against an allowlist before use. |
| 103 | The coordination directories are checked for symlink attacks before any |
| 104 | kqueue fd is opened. |
| 105 | |
| 106 | Performance:: |
| 107 | |
| 108 | kqueue backend O(0) CPU between events — woken by the kernel only when |
| 109 | a directory changes. Handles ``write_text_atomic``'s |
| 110 | mkstemp+rename pattern correctly (confirmed via testing). |
| 111 | |
| 112 | polling backend O(n) scan every ``--poll-interval`` seconds where n is |
| 113 | the total number of JSON files across all 4 coord dirs. |
| 114 | |
| 115 | Record cache The last-known data for every live record is kept in |
| 116 | memory so ``removed`` events can carry the record's data |
| 117 | even after the file is gone. The cache is bounded by the |
| 118 | number of live coordination records (typically ≪ 10 000). |
| 119 | |
| 120 | Exit codes:: |
| 121 | |
| 122 | 0 — normal exit (--once, --timeout, --max-events, SIGINT, or SIGTERM) |
| 123 | 1 — bad arguments or backend initialisation failure |
| 124 | """ |
| 125 | |
| 126 | from __future__ import annotations |
| 127 | |
| 128 | import argparse |
| 129 | import dataclasses |
| 130 | import datetime |
| 131 | import json |
| 132 | import logging |
| 133 | import os |
| 134 | import pathlib |
| 135 | import select |
| 136 | import signal |
| 137 | import sys |
| 138 | import time |
| 139 | import types |
| 140 | |
| 141 | from muse._version import __version__ |
| 142 | from muse.core.coordination import active_reservations |
| 143 | from muse.core.errors import ExitCode |
| 144 | from muse.core.repo import require_repo |
| 145 | from muse.core.validation import sanitize_display |
| 146 | |
| 147 | |
| 148 | type _IconMap = dict[str, str] |
| 149 | logger = logging.getLogger(__name__) |
| 150 | |
| 151 | # JSON-compatible value type for parsed coordination record payloads. |
| 152 | JsonValue = str | int | float | bool | None | list["JsonValue"] | "JsonDict" |
| 153 | JsonDict = dict[str, JsonValue] |
| 154 | |
| 155 | # ── Input constraints ───────────────────────────────────────────────────────── |
| 156 | |
| 157 | #: Maximum byte-length of the ``--run-id`` filter value. Matches the cap |
| 158 | #: applied to run-id in all other coordination commands. |
| 159 | _MAX_RUN_ID_LEN: int = 256 |
| 160 | |
| 161 | #: Allowed range for ``--poll-interval`` (seconds). |
| 162 | _MIN_POLL_INTERVAL: float = 0.01 |
| 163 | _MAX_POLL_INTERVAL: float = 3600.0 |
| 164 | |
| 165 | # ── Directory layout ────────────────────────────────────────────────────────── |
| 166 | |
| 167 | _SUBDIRS: tuple[str, ...] = ("reservations", "intents", "releases", "heartbeats", "tasks", "claims", "dependencies") |
| 168 | _VALID_KINDS: frozenset[str] = frozenset(_SUBDIRS) |
| 169 | |
| 170 | |
| 171 | def _coord_dir(root: pathlib.Path) -> pathlib.Path: |
| 172 | """Return ``.muse/coordination/`` for *root*.""" |
| 173 | return root / ".muse" / "coordination" |
| 174 | |
| 175 | |
| 176 | def _ensure_coord_dirs(root: pathlib.Path) -> None: |
| 177 | """Create all four coordination subdirectories if they do not exist.""" |
| 178 | for sub in _SUBDIRS: |
| 179 | (_coord_dir(root) / sub).mkdir(parents=True, exist_ok=True) |
| 180 | |
| 181 | |
| 182 | # ── Data model ──────────────────────────────────────────────────────────────── |
| 183 | |
| 184 | # Internal type alias: kind → {uuid → (mtime_ns, size)} |
| 185 | _Snapshot = dict[str, dict[str, tuple[int, int]]] |
| 186 | |
| 187 | |
| 188 | @dataclasses.dataclass |
| 189 | class WatchEvent: |
| 190 | """A single coordination event emitted by the watch loop. |
| 191 | |
| 192 | Attributes |
| 193 | ---------- |
| 194 | event_type: |
| 195 | One of ``"snapshot"``, ``"added"``, ``"modified"``, ``"removed"``, |
| 196 | or ``"expired"``. |
| 197 | kind: |
| 198 | One of ``"reservation"``, ``"intent"``, ``"release"``, |
| 199 | or ``"heartbeat"``. |
| 200 | id: |
| 201 | UUID of the coordination record that changed. |
| 202 | timestamp: |
| 203 | ISO 8601 UTC string — when this event was generated (not when the |
| 204 | underlying file change occurred). |
| 205 | data: |
| 206 | Parsed JSON content of the record. Empty dict for ``"removed"`` |
| 207 | events when no cached data exists. |
| 208 | """ |
| 209 | |
| 210 | event_type: str |
| 211 | kind: str |
| 212 | id: str |
| 213 | timestamp: str |
| 214 | data: JsonDict |
| 215 | |
| 216 | def to_dict(self) -> JsonDict: |
| 217 | """Serialise to the public JSON schema.""" |
| 218 | return { |
| 219 | "schema_version": __version__, |
| 220 | "event_type": self.event_type, |
| 221 | "kind": self.kind, |
| 222 | "id": self.id, |
| 223 | "timestamp": self.timestamp, |
| 224 | "data": self.data, |
| 225 | } |
| 226 | |
| 227 | |
| 228 | # ── Backend ABC ─────────────────────────────────────────────────────────────── |
| 229 | |
| 230 | |
| 231 | class _Backend: |
| 232 | """Abstract base for kqueue and polling backends. |
| 233 | |
| 234 | Subclasses implement :meth:`wait` and :meth:`close`. The ``name`` |
| 235 | attribute identifies the backend for diagnostic output. |
| 236 | """ |
| 237 | |
| 238 | name: str = "unknown" |
| 239 | |
| 240 | def wait(self, timeout: float) -> bool: |
| 241 | """Block up to *timeout* seconds. |
| 242 | |
| 243 | Returns ``True`` if at least one FS change was detected (kqueue), or |
| 244 | ``True`` unconditionally after sleeping (polling). Returns ``False`` |
| 245 | only if the backend times out with certainty that nothing changed |
| 246 | (never happens in the polling backend). |
| 247 | """ |
| 248 | raise NotImplementedError |
| 249 | |
| 250 | def close(self) -> None: |
| 251 | """Release all OS resources held by this backend.""" |
| 252 | |
| 253 | |
| 254 | # ── kqueue backend (macOS / BSD) ────────────────────────────────────────────── |
| 255 | |
| 256 | |
| 257 | class _KqueueBackend(_Backend): |
| 258 | """File-system event watcher using BSD kqueue. |
| 259 | |
| 260 | Opens one ``O_RDONLY`` file descriptor per coordination subdirectory and |
| 261 | registers a ``KQ_FILTER_VNODE / KQ_NOTE_WRITE`` interest. The ``wait`` |
| 262 | call blocks in the kernel until a directory entry is added, removed, or |
| 263 | replaced (which ``write_text_atomic``'s mkstemp+rename pattern triggers). |
| 264 | |
| 265 | Parameters |
| 266 | ---------- |
| 267 | dirs: |
| 268 | List of directory paths to watch. Each directory is created if it |
| 269 | does not exist. A symlink-swap attack (symlinked directory pointing |
| 270 | outside the repo) raises :exc:`ValueError` before any fd is opened. |
| 271 | |
| 272 | Raises |
| 273 | ------ |
| 274 | ValueError |
| 275 | If any watched directory is a symlink. |
| 276 | OSError |
| 277 | If a directory cannot be opened for watching. |
| 278 | """ |
| 279 | |
| 280 | name = "kqueue" |
| 281 | |
| 282 | def __init__(self, dirs: list[pathlib.Path]) -> None: |
| 283 | self._kq = select.kqueue() |
| 284 | self._fds: list[int] = [] |
| 285 | kevents = [] |
| 286 | for d in dirs: |
| 287 | d.mkdir(parents=True, exist_ok=True) |
| 288 | # Symlink-swap guard: reject symlinked directories. |
| 289 | if d.is_symlink(): |
| 290 | self._kq.close() |
| 291 | raise ValueError( |
| 292 | f"Coordination directory is a symlink — refusing to watch: {d}" |
| 293 | ) |
| 294 | fd = os.open(str(d), os.O_RDONLY) |
| 295 | self._fds.append(fd) |
| 296 | kevents.append(select.kevent( |
| 297 | fd, |
| 298 | filter=select.KQ_FILTER_VNODE, |
| 299 | flags=( |
| 300 | select.KQ_EV_ADD |
| 301 | | select.KQ_EV_ENABLE |
| 302 | | select.KQ_EV_CLEAR |
| 303 | ), |
| 304 | fflags=( |
| 305 | select.KQ_NOTE_WRITE |
| 306 | | select.KQ_NOTE_DELETE |
| 307 | | select.KQ_NOTE_EXTEND |
| 308 | | select.KQ_NOTE_RENAME |
| 309 | ), |
| 310 | )) |
| 311 | # Register all events in one syscall. |
| 312 | self._kq.control(kevents, 0) |
| 313 | |
| 314 | def wait(self, timeout: float) -> bool: |
| 315 | """Block up to *timeout* seconds; return True if any FS change fired.""" |
| 316 | events = self._kq.control(None, 64, max(0.0, timeout)) |
| 317 | return len(events) > 0 |
| 318 | |
| 319 | def close(self) -> None: |
| 320 | """Close the kqueue and all watched directory fds.""" |
| 321 | try: |
| 322 | self._kq.close() |
| 323 | except OSError: |
| 324 | pass |
| 325 | for fd in self._fds: |
| 326 | try: |
| 327 | os.close(fd) |
| 328 | except OSError: |
| 329 | pass |
| 330 | self._fds.clear() |
| 331 | |
| 332 | |
| 333 | # ── Polling backend (Linux / other) ────────────────────────────────────────── |
| 334 | |
| 335 | |
| 336 | class _PollingBackend(_Backend): |
| 337 | """mtime-scan polling backend for platforms without kqueue. |
| 338 | |
| 339 | Sleeps for *interval* seconds between directory scans. Always returns |
| 340 | ``True`` from :meth:`wait` so the caller always runs a full diff. |
| 341 | |
| 342 | Parameters |
| 343 | ---------- |
| 344 | interval: |
| 345 | Polling interval in seconds (clamped to [0.01, 3600]). |
| 346 | """ |
| 347 | |
| 348 | name = "polling" |
| 349 | |
| 350 | def __init__(self, interval: float) -> None: |
| 351 | self._interval = max(0.01, min(3600.0, interval)) |
| 352 | |
| 353 | def wait(self, timeout: float) -> bool: |
| 354 | """Sleep for ``min(interval, timeout)`` seconds; always returns True.""" |
| 355 | time.sleep(min(self._interval, max(0.0, timeout))) |
| 356 | return True |
| 357 | |
| 358 | def close(self) -> None: |
| 359 | pass |
| 360 | |
| 361 | |
| 362 | # ── Core scanning and diffing ───────────────────────────────────────────────── |
| 363 | |
| 364 | |
| 365 | def _scan_dirs(root: pathlib.Path) -> _Snapshot: |
| 366 | """Snapshot the mtime_ns and size of every ``.json`` file in all coord dirs. |
| 367 | |
| 368 | Returns a mapping ``{kind: {uuid: (mtime_ns, size_bytes)}}``. Directories |
| 369 | that do not exist yet are returned as empty dicts. Files that disappear |
| 370 | between the glob and the stat call are silently skipped. |
| 371 | |
| 372 | This function is O(n) where n is the total number of JSON files across the |
| 373 | four coordination subdirectories. On a typical swarm with hundreds of |
| 374 | agents this is a single directory scan per kind, which is very fast. |
| 375 | """ |
| 376 | snap: _Snapshot = {sub: {} for sub in _SUBDIRS} |
| 377 | coord = _coord_dir(root) |
| 378 | for sub in _SUBDIRS: |
| 379 | d = coord / sub |
| 380 | if not d.exists(): |
| 381 | continue |
| 382 | for path in d.glob("*.json"): |
| 383 | try: |
| 384 | st = path.stat() |
| 385 | snap[sub][path.stem] = (st.st_mtime_ns, st.st_size) |
| 386 | except OSError: |
| 387 | pass # Deleted between glob and stat — skip gracefully. |
| 388 | return snap |
| 389 | |
| 390 | |
| 391 | def _diff_snapshots( |
| 392 | old: _Snapshot, |
| 393 | new: _Snapshot, |
| 394 | ) -> list[tuple[str, str, str]]: |
| 395 | """Compute the delta between two snapshots. |
| 396 | |
| 397 | Returns a list of ``(event_type, kind, uuid)`` tuples: |
| 398 | |
| 399 | * ``"added"`` — uuid exists in *new* but not *old* |
| 400 | * ``"removed"`` — uuid exists in *old* but not *new* |
| 401 | * ``"modified"`` — uuid exists in both but ``(mtime_ns, size)`` changed |
| 402 | |
| 403 | Results are sorted by kind (canonical ``_SUBDIRS`` order) and then by uuid |
| 404 | within each kind for deterministic test output. |
| 405 | """ |
| 406 | events: list[tuple[str, str, str]] = [] |
| 407 | for sub in _SUBDIRS: |
| 408 | old_sub = old.get(sub, {}) |
| 409 | new_sub = new.get(sub, {}) |
| 410 | old_ids = set(old_sub) |
| 411 | new_ids = set(new_sub) |
| 412 | for uid in sorted(new_ids - old_ids): |
| 413 | events.append(("added", sub, uid)) |
| 414 | for uid in sorted(old_ids - new_ids): |
| 415 | events.append(("removed", sub, uid)) |
| 416 | for uid in sorted(old_ids & new_ids): |
| 417 | if old_sub[uid] != new_sub[uid]: |
| 418 | events.append(("modified", sub, uid)) |
| 419 | return events |
| 420 | |
| 421 | |
| 422 | def _load_record(root: pathlib.Path, kind: str, uid: str) -> JsonDict | None: |
| 423 | """Load and parse the JSON record for *kind*/*uid*. |
| 424 | |
| 425 | Returns ``None`` if the file does not exist or cannot be parsed. This |
| 426 | handles the TOCTOU window between snapshot and load gracefully. |
| 427 | """ |
| 428 | path = _coord_dir(root) / kind / f"{uid}.json" |
| 429 | try: |
| 430 | return json.loads(path.read_text()) |
| 431 | except (OSError, json.JSONDecodeError): |
| 432 | return None |
| 433 | |
| 434 | |
| 435 | # ── Filtering ───────────────────────────────────────────────────────────────── |
| 436 | |
| 437 | |
| 438 | def _passes_filters( |
| 439 | kind: str, |
| 440 | data: JsonDict, |
| 441 | kind_filter: str | None, |
| 442 | run_id_filter: str | None, |
| 443 | branch_filter: str | None, |
| 444 | ) -> bool: |
| 445 | """Return True if an event with the given *kind* and *data* passes all filters. |
| 446 | |
| 447 | Filters compose with AND semantics: |
| 448 | |
| 449 | * ``kind_filter`` — exact match on the event kind string. |
| 450 | * ``run_id_filter`` — exact match on ``data["run_id"]``. |
| 451 | * ``branch_filter`` — exact match on ``data["branch"]``. |
| 452 | |
| 453 | An event with no ``run_id`` or ``branch`` field in *data* (e.g. a |
| 454 | ``removed`` event with empty data) will NOT pass run_id/branch filters. |
| 455 | This is intentional — the record cache ensures removed events carry cached |
| 456 | data so filters still work for known records. |
| 457 | """ |
| 458 | if kind_filter and kind != kind_filter: |
| 459 | return False |
| 460 | if run_id_filter and data.get("run_id") != run_id_filter: |
| 461 | return False |
| 462 | if branch_filter and data.get("branch") != branch_filter: |
| 463 | return False |
| 464 | return True |
| 465 | |
| 466 | |
| 467 | # ── Event emission ──────────────────────────────────────────────────────────── |
| 468 | |
| 469 | |
| 470 | def _make_event( |
| 471 | event_type: str, |
| 472 | kind: str, |
| 473 | uid: str, |
| 474 | data: JsonDict, |
| 475 | ) -> WatchEvent: |
| 476 | """Construct a :class:`WatchEvent` stamped with the current UTC time.""" |
| 477 | return WatchEvent( |
| 478 | event_type=event_type, |
| 479 | kind=kind, |
| 480 | id=uid, |
| 481 | timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(), |
| 482 | data=data, |
| 483 | ) |
| 484 | |
| 485 | |
| 486 | # Icon prefixes for text output. |
| 487 | _TEXT_ICONS: _IconMap = { |
| 488 | "snapshot": "·", |
| 489 | "added": "+", |
| 490 | "modified": "~", |
| 491 | "removed": "-", |
| 492 | "expired": "!", |
| 493 | } |
| 494 | |
| 495 | |
| 496 | def _emit_event(ev: WatchEvent, as_json: bool) -> None: |
| 497 | """Print *ev* to stdout in JSON (NDJSON) or human-readable text format. |
| 498 | |
| 499 | **JSON mode**: one JSON object per line, flushed immediately so agents |
| 500 | reading from a pipe receive events without buffering. |
| 501 | |
| 502 | **Text mode**: each line is ``<icon> <kind> <summary>`` — concise enough |
| 503 | to scan in a terminal but includes the UUID prefix so events can be |
| 504 | correlated with ``muse coord list`` output. |
| 505 | |
| 506 | All agent-supplied strings (run_id, branch, addresses, reason) are passed |
| 507 | through :func:`~muse.core.validation.sanitize_display` before printing. |
| 508 | """ |
| 509 | if as_json: |
| 510 | print(json.dumps(ev.to_dict()), flush=True) |
| 511 | return |
| 512 | |
| 513 | icon = _TEXT_ICONS.get(ev.event_type, "?") |
| 514 | uid_short = sanitize_display(ev.id[:8]) |
| 515 | kind_col = f"{ev.kind:<12}" |
| 516 | |
| 517 | data = ev.data |
| 518 | if not data: |
| 519 | # Removed event with empty cache or expiration with no data. |
| 520 | print(f"{icon} {kind_col} {uid_short}", flush=True) |
| 521 | return |
| 522 | |
| 523 | run_id = sanitize_display(data.get("run_id", "?")) |
| 524 | branch = sanitize_display(data.get("branch") or "") |
| 525 | agent_str = f"{run_id}@{branch}" if branch else run_id |
| 526 | |
| 527 | if ev.kind == "reservation": |
| 528 | addrs: list[str] = data.get("addresses", []) |
| 529 | addr_parts = [sanitize_display(a) for a in addrs[:3]] |
| 530 | suffix = f" +{len(addrs) - 3} more" if len(addrs) > 3 else "" |
| 531 | addr_str = ", ".join(addr_parts) + suffix |
| 532 | print( |
| 533 | f"{icon} {kind_col} {agent_str:<32} {uid_short} [{addr_str}]", |
| 534 | flush=True, |
| 535 | ) |
| 536 | |
| 537 | elif ev.kind == "intent": |
| 538 | op = sanitize_display(data.get("operation", "?")) |
| 539 | addrs = data.get("addresses", []) |
| 540 | first_addr = sanitize_display(addrs[0]) if addrs else "?" |
| 541 | print( |
| 542 | f"{icon} {kind_col} {agent_str:<32} {uid_short} {op} {first_addr}", |
| 543 | flush=True, |
| 544 | ) |
| 545 | |
| 546 | elif ev.kind == "release": |
| 547 | reason = sanitize_display(data.get("reason", "?")) |
| 548 | res_id = sanitize_display((data.get("reservation_id") or "")[:8]) |
| 549 | print( |
| 550 | f"{icon} {kind_col} {run_id:<32} {uid_short} {reason} (res: {res_id})", |
| 551 | flush=True, |
| 552 | ) |
| 553 | |
| 554 | elif ev.kind == "heartbeat": |
| 555 | ext = sanitize_display(data.get("extended_expires_at", "?")) |
| 556 | res_id = sanitize_display((data.get("reservation_id") or "")[:8]) |
| 557 | print( |
| 558 | f"{icon} {kind_col} {run_id:<32} {uid_short} extends to {ext} (res: {res_id})", |
| 559 | flush=True, |
| 560 | ) |
| 561 | |
| 562 | else: |
| 563 | print(f"{icon} {kind_col} {agent_str:<32} {uid_short}", flush=True) |
| 564 | |
| 565 | |
| 566 | # ── Expiration detection ────────────────────────────────────────────────────── |
| 567 | |
| 568 | |
| 569 | def _check_expirations( |
| 570 | root: pathlib.Path, |
| 571 | prev_active_ids: set[str], |
| 572 | removed_ids: set[str], |
| 573 | ) -> tuple[list[WatchEvent], set[str]]: |
| 574 | """Detect reservations that crossed their effective expiry since last check. |
| 575 | |
| 576 | Compares the set of currently-active reservation IDs (from |
| 577 | :func:`~muse.core.coordination.active_reservations`) against |
| 578 | *prev_active_ids*. Any ID that was previously active but is no longer — |
| 579 | AND was not explicitly removed from disk this cycle — is treated as |
| 580 | expired. |
| 581 | |
| 582 | Parameters |
| 583 | ---------- |
| 584 | root: |
| 585 | Repository root. |
| 586 | prev_active_ids: |
| 587 | IDs that were active on the previous call. |
| 588 | removed_ids: |
| 589 | IDs of reservations whose JSON files were deleted this cycle. |
| 590 | These are already emitted as ``removed`` events and must not be |
| 591 | double-counted as ``expired``. |
| 592 | |
| 593 | Returns |
| 594 | ------- |
| 595 | (expiry_events, current_active_ids) |
| 596 | The list of new expiration events and the refreshed active-ID set |
| 597 | (to be passed as *prev_active_ids* on the next call). |
| 598 | """ |
| 599 | curr_active = active_reservations(root) |
| 600 | curr_active_ids: set[str] = {r.reservation_id for r in curr_active} |
| 601 | |
| 602 | events: list[WatchEvent] = [] |
| 603 | for uid in prev_active_ids - curr_active_ids: |
| 604 | if uid in removed_ids: |
| 605 | continue # GC'd reservation — already covered by a removed event. |
| 606 | data = _load_record(root, "reservations", uid) or {} |
| 607 | events.append(_make_event("expired", "reservations", uid, data)) |
| 608 | |
| 609 | return events, curr_active_ids |
| 610 | |
| 611 | |
| 612 | # ── Core watch loop ─────────────────────────────────────────────────────────── |
| 613 | |
| 614 | |
| 615 | def _watch_loop( |
| 616 | root: pathlib.Path, |
| 617 | backend: _Backend, |
| 618 | *, |
| 619 | kind_filter: str | None, |
| 620 | run_id_filter: str | None, |
| 621 | branch_filter: str | None, |
| 622 | as_json: bool, |
| 623 | once: bool, |
| 624 | timeout: float | None, |
| 625 | poll_interval: float, |
| 626 | max_events: int | None = None, |
| 627 | ) -> None: |
| 628 | """Core event loop — scan, diff, emit, repeat. |
| 629 | |
| 630 | This function is deliberately separated from :func:`run` so that tests can |
| 631 | inject a mock backend without subprocess overhead. |
| 632 | |
| 633 | Algorithm |
| 634 | --------- |
| 635 | 1. **Snapshot**: scan all four coord dirs and build |
| 636 | ``{kind: {uuid: (mtime_ns, size)}}``. |
| 637 | 2. **Snapshot events**: emit one ``snapshot`` event per existing record |
| 638 | (filtered). This gives agents an accurate initial picture of swarm |
| 639 | state before any deltas. |
| 640 | 3. **Record cache**: store last-known data for every record in memory so |
| 641 | ``removed`` events carry the record's content even after the file is |
| 642 | gone. This is critical for ``--run-id`` / ``--branch`` filters to |
| 643 | work correctly on removal events. |
| 644 | 4. **Loop**: wait for the backend (kqueue event or poll interval), rescan, |
| 645 | diff against previous snapshot, emit change events, check expirations. |
| 646 | 5. **Expiration**: after each loop iteration, compare the set of active |
| 647 | reservation IDs against the previous set. IDs that left the active set |
| 648 | without being explicitly removed fire ``expired`` events. |
| 649 | 6. **Max-events cap**: if *max_events* is set, the loop exits as soon as |
| 650 | the total number of emitted events (snapshot + change) reaches the cap. |
| 651 | |
| 652 | Parameters |
| 653 | ---------- |
| 654 | root: |
| 655 | Repository root (directory containing ``.muse/``). |
| 656 | backend: |
| 657 | Initialised :class:`_KqueueBackend` or :class:`_PollingBackend`. |
| 658 | kind_filter: |
| 659 | If set, only emit events for this kind. |
| 660 | run_id_filter: |
| 661 | If set, only emit events where ``data["run_id"]`` matches. |
| 662 | branch_filter: |
| 663 | If set, only emit events where ``data["branch"]`` matches. |
| 664 | as_json: |
| 665 | If True, output NDJSON; otherwise human-readable text. |
| 666 | once: |
| 667 | If True, emit snapshot events then return immediately (no loop). |
| 668 | timeout: |
| 669 | Stop looping after this many seconds. ``None`` = run forever. |
| 670 | poll_interval: |
| 671 | Maximum time to block in :meth:`_Backend.wait` (also the kqueue |
| 672 | timeout so expirations are checked at least this often). |
| 673 | max_events: |
| 674 | If set, exit after emitting this many events (across snapshot and |
| 675 | change events). ``None`` = no cap. |
| 676 | """ |
| 677 | emitted = 0 # Total events emitted so far (for max_events enforcement). |
| 678 | |
| 679 | def _emit(ev: WatchEvent) -> bool: |
| 680 | """Emit *ev* and return True; return False if the cap has been reached.""" |
| 681 | nonlocal emitted |
| 682 | _emit_event(ev, as_json) |
| 683 | emitted += 1 |
| 684 | return max_events is None or emitted < max_events |
| 685 | |
| 686 | # Snapshot initial state. |
| 687 | snap = _scan_dirs(root) |
| 688 | |
| 689 | # Record cache: (kind, uuid) → last-known parsed data. |
| 690 | record_cache: dict[tuple[str, str], JsonDict] = {} |
| 691 | |
| 692 | # Emit initial state as snapshot events. |
| 693 | for sub in _SUBDIRS: |
| 694 | for uid in sorted(snap.get(sub, {})): |
| 695 | data = _load_record(root, sub, uid) or {} |
| 696 | record_cache[(sub, uid)] = data |
| 697 | if _passes_filters(sub, data, kind_filter, run_id_filter, branch_filter): |
| 698 | ev = _make_event("snapshot", sub, uid, data) |
| 699 | if not _emit(ev): |
| 700 | return |
| 701 | |
| 702 | if once: |
| 703 | return |
| 704 | |
| 705 | # Seed expiration tracker with currently-active reservations. |
| 706 | prev_active_ids: set[str] = { |
| 707 | r.reservation_id for r in active_reservations(root) |
| 708 | } |
| 709 | |
| 710 | deadline = time.monotonic() + timeout if timeout is not None else None |
| 711 | |
| 712 | while True: |
| 713 | remaining = (deadline - time.monotonic()) if deadline is not None else None |
| 714 | if remaining is not None and remaining <= 0.0: |
| 715 | break |
| 716 | |
| 717 | # Block up to poll_interval (or remaining timeout, whichever is less). |
| 718 | wait_sec = min( |
| 719 | remaining if remaining is not None else poll_interval, |
| 720 | poll_interval, |
| 721 | ) |
| 722 | backend.wait(wait_sec) |
| 723 | |
| 724 | # Rescan unconditionally — handles both FS events and timeout wakeups |
| 725 | # (needed for expiration checks even when no files changed). |
| 726 | new_snap = _scan_dirs(root) |
| 727 | diff = _diff_snapshots(snap, new_snap) |
| 728 | snap = new_snap |
| 729 | |
| 730 | # Track which reservation files were deleted this cycle (for expiry dedup). |
| 731 | removed_reservation_ids: set[str] = set() |
| 732 | |
| 733 | for event_type, kind, uid in diff: |
| 734 | if event_type == "removed": |
| 735 | # Use cached data so filters work even after the file is gone. |
| 736 | data = record_cache.pop((kind, uid), {}) |
| 737 | if kind == "reservations": |
| 738 | removed_reservation_ids.add(uid) |
| 739 | else: |
| 740 | data = _load_record(root, kind, uid) or {} |
| 741 | if data: |
| 742 | record_cache[(kind, uid)] = data |
| 743 | |
| 744 | if _passes_filters(kind, data, kind_filter, run_id_filter, branch_filter): |
| 745 | ev = _make_event(event_type, kind, uid, data) |
| 746 | if not _emit(ev): |
| 747 | return |
| 748 | |
| 749 | # Check for expired reservations (independent of FS changes). |
| 750 | exp_events, prev_active_ids = _check_expirations( |
| 751 | root, prev_active_ids, removed_reservation_ids |
| 752 | ) |
| 753 | for ev in exp_events: |
| 754 | if _passes_filters( |
| 755 | ev.kind, ev.data, kind_filter, run_id_filter, branch_filter |
| 756 | ): |
| 757 | if not _emit(ev): |
| 758 | return |
| 759 | |
| 760 | |
| 761 | # ── CLI registration ────────────────────────────────────────────────────────── |
| 762 | |
| 763 | |
| 764 | def register( |
| 765 | subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", |
| 766 | ) -> None: |
| 767 | """Register the ``watch`` subcommand on *subparsers* (under ``muse coord``). |
| 768 | |
| 769 | Wires all flags with their defaults, choices, and help text so that |
| 770 | ``--help`` output is accurate. Sets ``func`` to :func:`run`. |
| 771 | """ |
| 772 | parser = subparsers.add_parser( |
| 773 | "watch", |
| 774 | help="Stream coordination events in real time (kqueue/polling).", |
| 775 | description=__doc__, |
| 776 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 777 | ) |
| 778 | parser.add_argument( |
| 779 | "--once", |
| 780 | action="store_true", |
| 781 | default=False, |
| 782 | help="Emit current state as snapshot events, then exit.", |
| 783 | ) |
| 784 | parser.add_argument( |
| 785 | "--timeout", |
| 786 | type=float, |
| 787 | default=None, |
| 788 | metavar="SECONDS", |
| 789 | help=( |
| 790 | "Stop watching after SECONDS seconds (>= 0). " |
| 791 | "0 is equivalent to --once." |
| 792 | ), |
| 793 | ) |
| 794 | parser.add_argument( |
| 795 | "--max-events", |
| 796 | type=int, |
| 797 | default=None, |
| 798 | metavar="N", |
| 799 | dest="max_events", |
| 800 | help=( |
| 801 | "Exit after emitting N events (>= 1). Useful for batch " |
| 802 | "processing without a time-based timeout." |
| 803 | ), |
| 804 | ) |
| 805 | parser.add_argument( |
| 806 | "--poll-interval", |
| 807 | type=float, |
| 808 | default=1.0, |
| 809 | metavar="SECS", |
| 810 | dest="poll_interval", |
| 811 | help=( |
| 812 | f"Polling/kqueue-timeout interval in seconds " |
| 813 | f"(default: 1.0; range: [{_MIN_POLL_INTERVAL}, {_MAX_POLL_INTERVAL}])." |
| 814 | ), |
| 815 | ) |
| 816 | parser.add_argument( |
| 817 | "--run-id", |
| 818 | default=None, |
| 819 | metavar="RUNID", |
| 820 | dest="run_id", |
| 821 | help=( |
| 822 | "Only emit events where data.run_id matches. " |
| 823 | f"Maximum {_MAX_RUN_ID_LEN} characters." |
| 824 | ), |
| 825 | ) |
| 826 | parser.add_argument( |
| 827 | "--branch", "-b", |
| 828 | default=None, |
| 829 | metavar="BRANCH", |
| 830 | dest="branch_filter", |
| 831 | help="Only emit events where data.branch matches.", |
| 832 | ) |
| 833 | parser.add_argument( |
| 834 | "--kind", |
| 835 | default=None, |
| 836 | choices=sorted(_VALID_KINDS), |
| 837 | metavar="KIND", |
| 838 | help=( |
| 839 | "Only emit events for this kind " |
| 840 | "(" + " | ".join(sorted(_VALID_KINDS)) + ")." |
| 841 | ), |
| 842 | ) |
| 843 | parser.add_argument( |
| 844 | "--format", "-f", |
| 845 | default="text", |
| 846 | dest="fmt", |
| 847 | choices=("text", "json"), |
| 848 | help="Output format: text (default) or json.", |
| 849 | ) |
| 850 | parser.add_argument( |
| 851 | "--json", |
| 852 | action="store_const", |
| 853 | const="json", |
| 854 | dest="fmt", |
| 855 | help="Shorthand for --format json.", |
| 856 | ) |
| 857 | parser.set_defaults(func=run) |
| 858 | |
| 859 | |
| 860 | # ── Command entry point ─────────────────────────────────────────────────────── |
| 861 | |
| 862 | |
| 863 | def run(args: argparse.Namespace) -> None: |
| 864 | """Stream coordination events from ``.muse/coordination/``. |
| 865 | |
| 866 | Execution order |
| 867 | --------------- |
| 868 | 1. **Validate inputs** — ``--run-id`` length, ``--poll-interval`` range, |
| 869 | ``--timeout`` sign, ``--max-events`` sign. All validation fires before |
| 870 | ``require_repo()`` or any file I/O; failures exit |
| 871 | :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message to |
| 872 | *stderr* (or compact JSON when ``--json`` is set). |
| 873 | 2. **Resolve repo** — ``require_repo()`` finds ``.muse/``. |
| 874 | 3. **Create coord dirs** — ``_ensure_coord_dirs()`` creates any missing |
| 875 | subdirectories so the backend can open them immediately. |
| 876 | 4. **Initialise backend** — kqueue on macOS/BSD, polling everywhere else. |
| 877 | 5. **Register SIGTERM** — maps to ``SystemExit(0)`` for clean process- |
| 878 | manager shutdown. |
| 879 | 6. **Run watch loop** — ``_watch_loop()`` handles scan → diff → emit. |
| 880 | |
| 881 | Backend selection |
| 882 | ----------------- |
| 883 | On macOS/BSD (where ``select.kqueue`` is available), a kqueue backend is |
| 884 | used: the process blocks in the kernel until a file is added, replaced, or |
| 885 | deleted in one of the four coordination subdirectories. CPU usage between |
| 886 | events is effectively zero. |
| 887 | |
| 888 | On Linux and other platforms, a polling backend is used: the process |
| 889 | sleeps for ``--poll-interval`` seconds and then rescans all directories. |
| 890 | |
| 891 | Both backends share the exact same scan → diff → emit logic. |
| 892 | |
| 893 | Signal handling |
| 894 | --------------- |
| 895 | ``SIGINT`` (Ctrl-C) raises :exc:`KeyboardInterrupt` which exits the loop |
| 896 | cleanly. ``SIGTERM`` (sent by process managers or test harnesses) is |
| 897 | mapped to :exc:`SystemExit(0)` via a signal handler registered in this |
| 898 | function and restored to ``SIG_DFL`` on exit. |
| 899 | |
| 900 | Security |
| 901 | -------- |
| 902 | * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` to bound memory use. |
| 903 | * ``--poll-interval`` is validated to |
| 904 | ``[_MIN_POLL_INTERVAL, _MAX_POLL_INTERVAL]`` before any I/O. |
| 905 | * All strings printed to stdout are passed through |
| 906 | :func:`~muse.core.validation.sanitize_display`. |
| 907 | * The kqueue backend checks each coord directory for symlinks before |
| 908 | opening any fd. |
| 909 | * ``--kind`` is validated against a frozenset allowlist by argparse. |
| 910 | |
| 911 | Performance |
| 912 | ----------- |
| 913 | kqueue: wakes only on FS changes — no CPU overhead between events. |
| 914 | polling: ``--poll-interval`` controls wake frequency; lower values give |
| 915 | faster detection at the cost of more directory I/O. |
| 916 | |
| 917 | Args: |
| 918 | args: Parsed ``argparse.Namespace`` with attributes ``once``, |
| 919 | ``timeout``, ``max_events``, ``poll_interval``, ``run_id``, |
| 920 | ``branch_filter``, ``kind``, and ``fmt``. |
| 921 | |
| 922 | Exit codes: |
| 923 | 0 — normal exit (``--once``, ``--timeout``, ``--max-events``, |
| 924 | SIGINT, or SIGTERM). |
| 925 | 1 — bad arguments or backend initialisation failure. |
| 926 | """ |
| 927 | as_json: bool = args.fmt == "json" |
| 928 | kind_filter: str | None = args.kind |
| 929 | run_id_filter: str | None = args.run_id |
| 930 | branch_filter: str | None = args.branch_filter |
| 931 | once: bool = args.once |
| 932 | timeout: float | None = args.timeout |
| 933 | max_events: int | None = args.max_events |
| 934 | poll_interval: float = args.poll_interval |
| 935 | |
| 936 | # ── Input validation (before any file I/O) ──────────────────────────────── |
| 937 | |
| 938 | if run_id_filter is not None and len(run_id_filter) > _MAX_RUN_ID_LEN: |
| 939 | msg = f"--run-id is too long ({len(run_id_filter)} chars; max {_MAX_RUN_ID_LEN})" |
| 940 | if as_json: |
| 941 | print(json.dumps({"error": msg, "status": "bad_args"})) |
| 942 | else: |
| 943 | print(f"❌ {msg}", file=sys.stderr) |
| 944 | raise SystemExit(ExitCode.USER_ERROR) |
| 945 | |
| 946 | if not (_MIN_POLL_INTERVAL <= poll_interval <= _MAX_POLL_INTERVAL): |
| 947 | msg = ( |
| 948 | f"--poll-interval must be in [{_MIN_POLL_INTERVAL}, {_MAX_POLL_INTERVAL}]," |
| 949 | f" got {poll_interval}" |
| 950 | ) |
| 951 | if as_json: |
| 952 | print(json.dumps({"error": msg, "status": "bad_args"})) |
| 953 | else: |
| 954 | print(f"❌ {msg}", file=sys.stderr) |
| 955 | raise SystemExit(ExitCode.USER_ERROR) |
| 956 | |
| 957 | if timeout is not None and timeout < 0.0: |
| 958 | msg = f"--timeout must be >= 0, got {timeout}" |
| 959 | if as_json: |
| 960 | print(json.dumps({"error": msg, "status": "bad_args"})) |
| 961 | else: |
| 962 | print(f"❌ {msg}", file=sys.stderr) |
| 963 | raise SystemExit(ExitCode.USER_ERROR) |
| 964 | |
| 965 | if max_events is not None and max_events < 1: |
| 966 | msg = f"--max-events must be >= 1, got {max_events}" |
| 967 | if as_json: |
| 968 | print(json.dumps({"error": msg, "status": "bad_args"})) |
| 969 | else: |
| 970 | print(f"❌ {msg}", file=sys.stderr) |
| 971 | raise SystemExit(ExitCode.USER_ERROR) |
| 972 | |
| 973 | # --timeout 0 is equivalent to --once. |
| 974 | if timeout is not None and timeout == 0.0: |
| 975 | once = True |
| 976 | timeout = None |
| 977 | |
| 978 | root = require_repo() |
| 979 | |
| 980 | # Ensure all coord dirs exist before we try to open or watch them. |
| 981 | _ensure_coord_dirs(root) |
| 982 | |
| 983 | # Select and initialise the backend. |
| 984 | try: |
| 985 | if hasattr(select, "kqueue"): |
| 986 | dirs = [_coord_dir(root) / sub for sub in _SUBDIRS] |
| 987 | backend: _Backend = _KqueueBackend(dirs) |
| 988 | else: |
| 989 | backend = _PollingBackend(poll_interval) |
| 990 | except (OSError, ValueError) as exc: |
| 991 | _err(f"Failed to initialise watch backend: {exc}") |
| 992 | raise SystemExit(ExitCode.USER_ERROR) |
| 993 | |
| 994 | # Register SIGTERM → SystemExit(0) so process managers can stop us cleanly. |
| 995 | prev_sigterm = signal.getsignal(signal.SIGTERM) |
| 996 | |
| 997 | def _handle_sigterm(sig: int, frame: types.FrameType | None) -> None: # noqa: ARG001 |
| 998 | raise SystemExit(0) |
| 999 | |
| 1000 | signal.signal(signal.SIGTERM, _handle_sigterm) |
| 1001 | |
| 1002 | if not as_json: |
| 1003 | backend_label = backend.name |
| 1004 | print( |
| 1005 | f"\nmuse coord watch — watching .muse/coordination/ ({backend_label})" |
| 1006 | ) |
| 1007 | parts: list[str] = [] |
| 1008 | if kind_filter: |
| 1009 | parts.append(f"kind={kind_filter}") |
| 1010 | if run_id_filter: |
| 1011 | parts.append(f"run-id={sanitize_display(run_id_filter)}") |
| 1012 | if branch_filter: |
| 1013 | parts.append(f"branch={sanitize_display(branch_filter)}") |
| 1014 | if max_events is not None: |
| 1015 | parts.append(f"max-events={max_events}") |
| 1016 | if parts: |
| 1017 | print(f" filter: {', '.join(parts)}") |
| 1018 | print("─" * 62) |
| 1019 | |
| 1020 | try: |
| 1021 | _watch_loop( |
| 1022 | root, |
| 1023 | backend, |
| 1024 | kind_filter=kind_filter, |
| 1025 | run_id_filter=run_id_filter, |
| 1026 | branch_filter=branch_filter, |
| 1027 | as_json=as_json, |
| 1028 | once=once, |
| 1029 | timeout=timeout, |
| 1030 | poll_interval=poll_interval, |
| 1031 | max_events=max_events, |
| 1032 | ) |
| 1033 | except KeyboardInterrupt: |
| 1034 | pass |
| 1035 | finally: |
| 1036 | backend.close() |
| 1037 | signal.signal(signal.SIGTERM, prev_sigterm) |
| 1038 | if not as_json: |
| 1039 | print("\n (watch ended)", flush=True) |
| 1040 | |
| 1041 | |
| 1042 | # ── Helpers ─────────────────────────────────────────────────────────────────── |
| 1043 | |
| 1044 | |
| 1045 | def _err(msg: str) -> None: |
| 1046 | """Print an error message to *stderr* with the ``❌`` prefix. |
| 1047 | |
| 1048 | Used for backend initialisation failures and other pre-loop errors. |
| 1049 | Validation failures in :func:`run` print their own messages before calling |
| 1050 | ``raise SystemExit``. |
| 1051 | """ |
| 1052 | print(f"❌ {msg}", file=sys.stderr) |
File History
1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8
docs(KD-STAGING): sync governance after KD-6b DONE
Human
12 days ago