"""``muse coord watch`` — stream coordination events in real time.

Watches ``.muse/coordination/`` for file-system changes and emits a structured
event for every coordination record that is added, modified, removed, or
expired.  On macOS/BSD the kqueue kernel interface is used, so the process
sleeps in the kernel and wakes **only** when a file actually changes — zero
CPU overhead between events.  On Linux and other platforms a configurable
polling fallback is used instead.

This replaces the polling anti-pattern where agents repeatedly call
``muse coord list`` to detect swarm state changes.  Piping
``muse coord watch --json`` gives an agent an event-driven coordination bus.

Event types
-----------
``snapshot``
    Emitted once per existing record immediately after startup, before any
    change events.  Agents should consume all ``snapshot`` events first to
    build an accurate initial picture of the swarm, then react to deltas.

``added``
    A new coordination file appeared on disk (e.g. another agent just called
    ``muse coord reserve``).

``modified``
    An existing file changed (e.g. a heartbeat that extended a reservation's
    TTL, or an intent that was amended).

``removed``
    A file was deleted from disk (e.g. GC cleaned up an expired reservation
    with ``muse coord gc --execute``).  The event carries the last-known data
    for the record so filters still work correctly.

``expired``
    A reservation that was previously active has crossed its effective
    expiry timestamp.  The file may still exist on disk (not GC'd yet) —
    this event fires when ``active_reservations()`` stops returning it.

Event kinds
-----------
``reservation``   — ``.muse/coordination/reservations/<id>.json``
``intent``        — ``.muse/coordination/intents/<id>.json``
``release``       — ``.muse/coordination/releases/<id>.json``
``heartbeat``     — ``.muse/coordination/heartbeats/<id>.json``

Usage::

    muse coord watch                        # stream all events, run forever
    muse coord watch --once                 # emit current state as snapshots, then exit
    muse coord watch --timeout 60           # run for 60 seconds then exit
    muse coord watch --max-events 10        # exit after receiving 10 events
    muse coord watch --kind reservation     # only reservation events
    muse coord watch --run-id agent-42      # only events from agent-42
    muse coord watch --branch feature/x    # only events on feature/x
    muse coord watch --json                 # NDJSON output (one event per line)
    muse coord watch --poll-interval 0.5   # faster fallback poll cadence

JSON output schema (NDJSON — one JSON object per line)::

    {
      "schema_version": str,
      "event_type":     "snapshot" | "added" | "modified" | "removed" | "expired",
      "kind":           "reservation" | "intent" | "release" | "heartbeat",
      "id":             str,       // content-addressed ID of the changed record
      "timestamp":      str,       // ISO 8601 UTC — when the event was generated
      "data":           dict | {}  // parsed record content; empty for "removed"
                                   // if the cache had no prior data
    }

Text output example::

    muse coord watch — watching .muse/coordination/ (kqueue)
    ──────────────────────────────────────────────────────────────
    · reservation   agent-42@feature/billing  a1b2c3d4  [billing.py::compute_total]
    + reservation   agent-99@main             e5f6a7b8  [auth.py::verify_token]
    ~ heartbeat     agent-42                  a1b2c3d4  extends to 2026-03-30T15:00Z
    ✓ release       agent-42                  a1b2c3d4  completed  (res: a1b2c3d4)
    ! expired       agent-99@main             e5f6a7b8  [auth.py::verify_token]

Flags::

    --once                  Emit current state as snapshot events, then exit.
    --timeout SECONDS       Stop watching after N seconds (>= 0; 0 = --once).
    --max-events N          Exit after emitting N events (>= 1).  Useful for
                            batch processing without a time-based timeout.
    --poll-interval SECS    Polling/kqueue-timeout interval in seconds
                            (default 1.0; range [0.01, 3600]).
    --run-id RUNID          Only emit events where data.run_id == RUNID.
                            Maximum 256 characters.
    --branch BRANCH         Only emit events where data.branch == BRANCH.
    --kind KIND             Only emit events for this kind
                            (reservation | intent | release | heartbeat).
    --format / -f           Output format: text (default) or json.
    --json                  Shorthand for --format json.

Security::

    All agent-supplied strings are passed through
    :func:`~muse.core.validation.sanitize_display` before printing to strip
    ANSI escape sequences and terminal control characters.

    The ``--kind`` argument is validated against an allowlist before use.
    The coordination directories are checked for symlink attacks before any
    kqueue fd is opened.

Performance::

    kqueue backend   O(0) CPU between events — woken by the kernel only when
                     a directory changes.  Handles ``write_text_atomic``'s
                     mkstemp+rename pattern correctly (confirmed via testing).

    polling backend  O(n) scan every ``--poll-interval`` seconds where n is
                     the total number of JSON files across all 4 coord dirs.

    Record cache     The last-known data for every live record is kept in
                     memory so ``removed`` events can carry the record's data
                     even after the file is gone.  The cache is bounded by the
                     number of live coordination records (typically ≪ 10 000).

Exit codes::

    0  — normal exit (--once, --timeout, --max-events, SIGINT, or SIGTERM)
    1  — bad arguments or backend initialisation failure
"""

import argparse
import dataclasses
import json
import logging
import os
import pathlib
import select
import signal
import sys
import time
import types

from muse import __version__
from muse.core.types import load_json_file, now_utc_iso
from muse.core.paths import coordination_dir as _coordination_dir
from muse.core.coordination import active_reservations
from muse.core.errors import ExitCode
from muse.core.repo import require_repo
from muse.core.validation import sanitize_display

type _IconMap = dict[str, str]
logger = logging.getLogger(__name__)

# JSON-compatible value type for parsed coordination record payloads.
JsonValue = str | int | float | bool | None | list["JsonValue"] | "JsonDict"
JsonDict = dict[str, JsonValue]

# ── Input constraints ─────────────────────────────────────────────────────────

#: Maximum byte-length of the ``--run-id`` filter value.  Matches the cap
#: applied to run-id in all other coordination commands.
_MAX_RUN_ID_LEN: int = 256

#: Allowed range for ``--poll-interval`` (seconds).
_MIN_POLL_INTERVAL: float = 0.01
_MAX_POLL_INTERVAL: float = 3600.0

# ── Directory layout ──────────────────────────────────────────────────────────

_SUBDIRS: tuple[str, ...] = ("reservations", "intents", "releases", "heartbeats", "tasks", "claims", "dependencies")
_VALID_KINDS: frozenset[str] = frozenset(_SUBDIRS)

def _coord_dir(root: pathlib.Path) -> pathlib.Path:
    """Return ``.muse/coordination/`` for *root*."""
    return _coordination_dir(root)

def _ensure_coord_dirs(root: pathlib.Path) -> None:
    """Create all four coordination subdirectories if they do not exist."""
    for sub in _SUBDIRS:
        (_coord_dir(root) / sub).mkdir(parents=True, exist_ok=True)

# ── Data model ────────────────────────────────────────────────────────────────

# Internal type alias: kind → {record_id → (mtime_ns, size)}
_SnapshotInner = dict[str, tuple[int, int]]
_Snapshot = dict[str, _SnapshotInner]

@dataclasses.dataclass
class WatchEvent:
    """A single coordination event emitted by the watch loop.

    Attributes
    ----------
    event_type:
        One of ``"snapshot"``, ``"added"``, ``"modified"``, ``"removed"``,
        or ``"expired"``.
    kind:
        One of ``"reservation"``, ``"intent"``, ``"release"``,
        or ``"heartbeat"``.
    id:
        content-addressed ID of the coordination record that changed.
    timestamp:
        ISO 8601 UTC string — when this event was generated (not when the
        underlying file change occurred).
    data:
        Parsed JSON content of the record.  Empty dict for ``"removed"``
        events when no cached data exists.
    """

    event_type: str
    kind: str
    id: str
    timestamp: str
    data: JsonDict

    def to_dict(self) -> JsonDict:
        """Serialise to the public JSON schema."""
        return {
            "schema_version": __version__,
            "event_type": self.event_type,
            "kind": self.kind,
            "id": self.id,
            "timestamp": self.timestamp,
            "data": self.data,
        }

# ── Backend ABC ───────────────────────────────────────────────────────────────

class _Backend:
    """Abstract base for kqueue and polling backends.

    Subclasses implement :meth:`wait` and :meth:`close`.  The ``name``
    attribute identifies the backend for diagnostic output.
    """

    name: str = "unknown"

    def wait(self, timeout: float) -> bool:
        """Block up to *timeout* seconds.

        Returns ``True`` if at least one FS change was detected (kqueue), or
        ``True`` unconditionally after sleeping (polling).  Returns ``False``
        only if the backend times out with certainty that nothing changed
        (never happens in the polling backend).
        """
        raise NotImplementedError

    def close(self) -> None:
        """Release all OS resources held by this backend."""

# ── kqueue backend (macOS / BSD) ──────────────────────────────────────────────

class _KqueueBackend(_Backend):
    """File-system event watcher using BSD kqueue.

    Opens one ``O_RDONLY`` file descriptor per coordination subdirectory and
    registers a ``KQ_FILTER_VNODE / KQ_NOTE_WRITE`` interest.  The ``wait``
    call blocks in the kernel until a directory entry is added, removed, or
    replaced (which ``write_text_atomic``'s mkstemp+rename pattern triggers).

    Parameters
    ----------
    dirs:
        List of directory paths to watch.  Each directory is created if it
        does not exist.  A symlink-swap attack (symlinked directory pointing
        outside the repo) raises :exc:`ValueError` before any fd is opened.

    Raises
    ------
    ValueError
        If any watched directory is a symlink.
    OSError
        If a directory cannot be opened for watching.
    """

    name = "kqueue"

    def __init__(self, dirs: list[pathlib.Path]) -> None:
        self._kq = select.kqueue()
        self._fds: list[int] = []
        kevents = []
        for d in dirs:
            d.mkdir(parents=True, exist_ok=True)
            # Symlink-swap guard: reject symlinked directories.
            if d.is_symlink():
                self._kq.close()
                raise ValueError(
                    f"Coordination directory is a symlink — refusing to watch: {d}"
                )
            fd = os.open(str(d), os.O_RDONLY)
            self._fds.append(fd)
            kevents.append(select.kevent(
                fd,
                filter=select.KQ_FILTER_VNODE,
                flags=(
                    select.KQ_EV_ADD
                    | select.KQ_EV_ENABLE
                    | select.KQ_EV_CLEAR
                ),
                fflags=(
                    select.KQ_NOTE_WRITE
                    | select.KQ_NOTE_DELETE
                    | select.KQ_NOTE_EXTEND
                    | select.KQ_NOTE_RENAME
                ),
            ))
        # Register all events in one syscall.
        self._kq.control(kevents, 0)

    def wait(self, timeout: float) -> bool:
        """Block up to *timeout* seconds; return True if any FS change fired."""
        events = self._kq.control(None, 64, max(0.0, timeout))
        return len(events) > 0

    def close(self) -> None:
        """Close the kqueue and all watched directory fds."""
        try:
            self._kq.close()
        except OSError:
            pass
        for fd in self._fds:
            try:
                os.close(fd)
            except OSError:
                pass
        self._fds.clear()

# ── Polling backend (Linux / other) ──────────────────────────────────────────

class _PollingBackend(_Backend):
    """mtime-scan polling backend for platforms without kqueue.

    Sleeps for *interval* seconds between directory scans.  Always returns
    ``True`` from :meth:`wait` so the caller always runs a full diff.

    Parameters
    ----------
    interval:
        Polling interval in seconds (clamped to [0.01, 3600]).
    """

    name = "polling"

    def __init__(self, interval: float) -> None:
        self._interval = max(0.01, min(3600.0, interval))

    def wait(self, timeout: float) -> bool:
        """Sleep for ``min(interval, timeout)`` seconds; always returns True."""
        time.sleep(min(self._interval, max(0.0, timeout)))
        return True

    def close(self) -> None:
        pass

# ── Core scanning and diffing ─────────────────────────────────────────────────

def _scan_dirs(root: pathlib.Path) -> _Snapshot:
    """Snapshot the mtime_ns and size of every ``.json`` file in all coord dirs.

    Returns a mapping ``{kind: {record_id: (mtime_ns, size_bytes)}}``.  Directories
    that do not exist yet are returned as empty dicts.  Files that disappear
    between the glob and the stat call are silently skipped.

    This function is O(n) where n is the total number of JSON files across the
    four coordination subdirectories.  On a typical swarm with hundreds of
    agents this is a single directory scan per kind, which is very fast.
    """
    snap: _Snapshot = {sub: {} for sub in _SUBDIRS}
    coord = _coord_dir(root)
    for sub in _SUBDIRS:
        d = coord / sub
        if not d.exists():
            continue
        for path in d.glob("*.json"):
            try:
                st = path.stat()
                snap[sub][path.stem] = (st.st_mtime_ns, st.st_size)
            except OSError:
                pass  # Deleted between glob and stat — skip gracefully.
    return snap

def _diff_snapshots(
    old: _Snapshot,
    new: _Snapshot,
) -> list[tuple[str, str, str]]:
    """Compute the delta between two snapshots.

    Returns a list of ``(event_type, kind, record_id)`` tuples:

    * ``"added"``    — record_id exists in *new* but not *old*
    * ``"removed"``  — record_id exists in *old* but not *new*
    * ``"modified"`` — record_id exists in both but ``(mtime_ns, size)`` changed

    Results are sorted by kind (canonical ``_SUBDIRS`` order) and then by record_id
    within each kind for deterministic test output.
    """
    events: list[tuple[str, str, str]] = []
    for sub in _SUBDIRS:
        old_sub = old.get(sub, {})
        new_sub = new.get(sub, {})
        old_ids = set(old_sub)
        new_ids = set(new_sub)
        for uid in sorted(new_ids - old_ids):
            events.append(("added", sub, uid))
        for uid in sorted(old_ids - new_ids):
            events.append(("removed", sub, uid))
        for uid in sorted(old_ids & new_ids):
            if old_sub[uid] != new_sub[uid]:
                events.append(("modified", sub, uid))
    return events

def _load_record(root: pathlib.Path, kind: str, uid: str) -> JsonDict | None:
    """Load and parse the JSON record for *kind*/*uid*.

    Returns ``None`` if the file does not exist or cannot be parsed.  This
    handles the TOCTOU window between snapshot and load gracefully.
    """
    path = _coord_dir(root) / kind / f"{uid}.json"
    return load_json_file(path)

# ── Filtering ─────────────────────────────────────────────────────────────────

def _passes_filters(
    kind: str,
    data: JsonDict,
    kind_filter: str | None,
    run_id_filter: str | None,
    branch_filter: str | None,
) -> bool:
    """Return True if an event with the given *kind* and *data* passes all filters.

    Filters compose with AND semantics:

    * ``kind_filter``     — exact match on the event kind string.
    * ``run_id_filter``   — exact match on ``data["run_id"]``.
    * ``branch_filter``   — exact match on ``data["branch"]``.

    An event with no ``run_id`` or ``branch`` field in *data* (e.g. a
    ``removed`` event with empty data) will NOT pass run_id/branch filters.
    This is intentional — the record cache ensures removed events carry cached
    data so filters still work for known records.
    """
    if kind_filter and kind != kind_filter:
        return False
    if run_id_filter and data.get("run_id") != run_id_filter:
        return False
    if branch_filter and data.get("branch") != branch_filter:
        return False
    return True

# ── Event emission ────────────────────────────────────────────────────────────

def _make_event(
    event_type: str,
    kind: str,
    uid: str,
    data: JsonDict,
) -> WatchEvent:
    """Construct a :class:`WatchEvent` stamped with the current UTC time."""
    return WatchEvent(
        event_type=event_type,
        kind=kind,
        id=uid,
        timestamp=now_utc_iso(),
        data=data,
    )

# Icon prefixes for text output.
_TEXT_ICONS: _IconMap = {
    "snapshot": "·",
    "added":    "+",
    "modified": "~",
    "removed":  "-",
    "expired":  "!",
}

def _emit_event(ev: WatchEvent, as_json: bool) -> None:
    """Print *ev* to stdout in JSON (NDJSON) or human-readable text format.

    **JSON mode**: one JSON object per line, flushed immediately so agents
    reading from a pipe receive events without buffering.

    **Text mode**: each line is ``<icon> <kind> <summary>`` — concise enough
    to scan in a terminal but includes the ID prefix so events can be
    correlated with ``muse coord list`` output.

    All agent-supplied strings (run_id, branch, addresses, reason) are passed
    through :func:`~muse.core.validation.sanitize_display` before printing.
    """
    if as_json:
        print(json.dumps(ev.to_dict()), flush=True)
        return

    icon = _TEXT_ICONS.get(ev.event_type, "?")
    uid = sanitize_display(ev.id)
    kind_col = f"{ev.kind:<12}"

    data = ev.data
    if not data:
        # Removed event with empty cache or expiration with no data.
        print(f"{icon} {kind_col} {uid}", flush=True)
        return

    run_id = sanitize_display(data.get("run_id", "?"))
    branch = sanitize_display(data.get("branch") or "")
    agent_str = f"{run_id}@{branch}" if branch else run_id

    if ev.kind == "reservation":
        addrs: list[str] = data.get("addresses", [])
        addr_parts = [sanitize_display(a) for a in addrs[:3]]
        suffix = f" +{len(addrs) - 3} more" if len(addrs) > 3 else ""
        addr_str = ", ".join(addr_parts) + suffix
        print(
            f"{icon} {kind_col} {agent_str:<32}  {uid}  [{addr_str}]",
            flush=True,
        )

    elif ev.kind == "intent":
        op = sanitize_display(data.get("operation", "?"))
        addrs = data.get("addresses", [])
        first_addr = sanitize_display(addrs[0]) if addrs else "?"
        print(
            f"{icon} {kind_col} {agent_str:<32}  {uid}  {op}  {first_addr}",
            flush=True,
        )

    elif ev.kind == "release":
        reason = sanitize_display(data.get("reason", "?"))
        res_id = sanitize_display(data.get("reservation_id") or "")
        print(
            f"{icon} {kind_col} {run_id:<32}  {uid}  {reason}  (res: {res_id})",
            flush=True,
        )

    elif ev.kind == "heartbeat":
        ext = sanitize_display(data.get("extended_expires_at", "?"))
        res_id = sanitize_display(data.get("reservation_id") or "")
        print(
            f"{icon} {kind_col} {run_id:<32}  {uid}  extends to {ext}  (res: {res_id})",
            flush=True,
        )

    else:
        print(f"{icon} {kind_col} {agent_str:<32}  {uid}", flush=True)

# ── Expiration detection ──────────────────────────────────────────────────────

def _check_expirations(
    root: pathlib.Path,
    prev_active_ids: set[str],
    removed_ids: set[str],
) -> tuple[list[WatchEvent], set[str]]:
    """Detect reservations that crossed their effective expiry since last check.

    Compares the set of currently-active reservation IDs (from
    :func:`~muse.core.coordination.active_reservations`) against
    *prev_active_ids*.  Any ID that was previously active but is no longer —
    AND was not explicitly removed from disk this cycle — is treated as
    expired.

    Parameters
    ----------
    root:
        Repository root.
    prev_active_ids:
        IDs that were active on the previous call.
    removed_ids:
        IDs of reservations whose JSON files were deleted this cycle.
        These are already emitted as ``removed`` events and must not be
        double-counted as ``expired``.

    Returns
    -------
    (expiry_events, current_active_ids)
        The list of new expiration events and the refreshed active-ID set
        (to be passed as *prev_active_ids* on the next call).
    """
    curr_active = active_reservations(root)
    curr_active_ids: set[str] = {r.reservation_id for r in curr_active}

    events: list[WatchEvent] = []
    for uid in prev_active_ids - curr_active_ids:
        if uid in removed_ids:
            continue  # GC'd reservation — already covered by a removed event.
        data = _load_record(root, "reservations", uid) or {}
        events.append(_make_event("expired", "reservations", uid, data))

    return events, curr_active_ids

# ── Core watch loop ───────────────────────────────────────────────────────────

def _watch_loop(
    root: pathlib.Path,
    backend: _Backend,
    *,
    kind_filter: str | None,
    run_id_filter: str | None,
    branch_filter: str | None,
    as_json: bool,
    once: bool,
    timeout: float | None,
    poll_interval: float,
    max_events: int | None = None,
) -> None:
    """Core event loop — scan, diff, emit, repeat.

    This function is deliberately separated from :func:`run` so that tests can
    inject a mock backend without subprocess overhead.

    Algorithm
    ---------
    1. **Snapshot**: scan all four coord dirs and build
       ``{kind: {record_id: (mtime_ns, size)}}``.
    2. **Snapshot events**: emit one ``snapshot`` event per existing record
       (filtered).  This gives agents an accurate initial picture of swarm
       state before any deltas.
    3. **Record cache**: store last-known data for every record in memory so
       ``removed`` events carry the record's content even after the file is
       gone.  This is critical for ``--run-id`` / ``--branch`` filters to
       work correctly on removal events.
    4. **Loop**: wait for the backend (kqueue event or poll interval), rescan,
       diff against previous snapshot, emit change events, check expirations.
    5. **Expiration**: after each loop iteration, compare the set of active
       reservation IDs against the previous set.  IDs that left the active set
       without being explicitly removed fire ``expired`` events.
    6. **Max-events cap**: if *max_events* is set, the loop exits as soon as
       the total number of emitted events (snapshot + change) reaches the cap.

    Parameters
    ----------
    root:
        Repository root (directory containing ``.muse/``).
    backend:
        Initialised :class:`_KqueueBackend` or :class:`_PollingBackend`.
    kind_filter:
        If set, only emit events for this kind.
    run_id_filter:
        If set, only emit events where ``data["run_id"]`` matches.
    branch_filter:
        If set, only emit events where ``data["branch"]`` matches.
    as_json:
        If True, output NDJSON; otherwise human-readable text.
    once:
        If True, emit snapshot events then return immediately (no loop).
    timeout:
        Stop looping after this many seconds.  ``None`` = run forever.
    poll_interval:
        Maximum time to block in :meth:`_Backend.wait` (also the kqueue
        timeout so expirations are checked at least this often).
    max_events:
        If set, exit after emitting this many events (across snapshot and
        change events).  ``None`` = no cap.
    """
    emitted = 0  # Total events emitted so far (for max_events enforcement).

    def _emit(ev: WatchEvent) -> bool:
        """Emit *ev* and return True; return False if the cap has been reached."""
        nonlocal emitted
        _emit_event(ev, as_json)
        emitted += 1
        return max_events is None or emitted < max_events

    # Snapshot initial state.
    snap = _scan_dirs(root)

    # Record cache: (kind, record_id) → last-known parsed data.
    record_cache: dict[tuple[str, str], JsonDict] = {}

    # Emit initial state as snapshot events.
    for sub in _SUBDIRS:
        for uid in sorted(snap.get(sub, {})):
            data = _load_record(root, sub, uid) or {}
            record_cache[(sub, uid)] = data
            if _passes_filters(sub, data, kind_filter, run_id_filter, branch_filter):
                ev = _make_event("snapshot", sub, uid, data)
                if not _emit(ev):
                    return

    if once:
        return

    # Seed expiration tracker with currently-active reservations.
    prev_active_ids: set[str] = {
        r.reservation_id for r in active_reservations(root)
    }

    deadline = time.monotonic() + timeout if timeout is not None else None

    while True:
        remaining = (deadline - time.monotonic()) if deadline is not None else None
        if remaining is not None and remaining <= 0.0:
            break

        # Block up to poll_interval (or remaining timeout, whichever is less).
        wait_sec = min(
            remaining if remaining is not None else poll_interval,
            poll_interval,
        )
        backend.wait(wait_sec)

        # Rescan unconditionally — handles both FS events and timeout wakeups
        # (needed for expiration checks even when no files changed).
        new_snap = _scan_dirs(root)
        diff = _diff_snapshots(snap, new_snap)
        snap = new_snap

        # Track which reservation files were deleted this cycle (for expiry dedup).
        removed_reservation_ids: set[str] = set()

        for event_type, kind, uid in diff:
            if event_type == "removed":
                # Use cached data so filters work even after the file is gone.
                data = record_cache.pop((kind, uid), {})
                if kind == "reservations":
                    removed_reservation_ids.add(uid)
            else:
                data = _load_record(root, kind, uid) or {}
                if data:
                    record_cache[(kind, uid)] = data

            if _passes_filters(kind, data, kind_filter, run_id_filter, branch_filter):
                ev = _make_event(event_type, kind, uid, data)
                if not _emit(ev):
                    return

        # Check for expired reservations (independent of FS changes).
        exp_events, prev_active_ids = _check_expirations(
            root, prev_active_ids, removed_reservation_ids
        )
        for ev in exp_events:
            if _passes_filters(
                ev.kind, ev.data, kind_filter, run_id_filter, branch_filter
            ):
                if not _emit(ev):
                    return

# ── CLI registration ──────────────────────────────────────────────────────────

def register(
    subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
) -> None:
    """Register the ``watch`` subcommand 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`.
    """
    parser = subparsers.add_parser(
        "watch",
        help="Stream coordination events in real time (kqueue/polling).",
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "--once",
        action="store_true",
        default=False,
        help="Emit current state as snapshot events, then exit.",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=None,
        metavar="SECONDS",
        help=(
            "Stop watching after SECONDS seconds (>= 0).  "
            "0 is equivalent to --once."
        ),
    )
    parser.add_argument(
        "--max-events",
        type=int,
        default=None,
        metavar="N",
        dest="max_events",
        help=(
            "Exit after emitting N events (>= 1).  Useful for batch "
            "processing without a time-based timeout."
        ),
    )
    parser.add_argument(
        "--poll-interval",
        type=float,
        default=1.0,
        metavar="SECS",
        dest="poll_interval",
        help=(
            f"Polling/kqueue-timeout interval in seconds "
            f"(default: 1.0; range: [{_MIN_POLL_INTERVAL}, {_MAX_POLL_INTERVAL}])."
        ),
    )
    parser.add_argument(
        "--run-id",
        default=None,
        metavar="RUNID",
        dest="run_id",
        help=(
            "Only emit events where data.run_id matches.  "
            f"Maximum {_MAX_RUN_ID_LEN} characters."
        ),
    )
    parser.add_argument(
        "--branch", "-b",
        default=None,
        metavar="BRANCH",
        dest="branch_filter",
        help="Only emit events where data.branch matches.",
    )
    parser.add_argument(
        "--kind",
        default=None,
        choices=sorted(_VALID_KINDS),
        metavar="KIND",
        help=(
            "Only emit events for this kind "
            f"({' | '.join(sorted(_VALID_KINDS))})."
        ),
    )
    parser.add_argument(
        "--json", "-j",
        action="store_true",
        dest="json_out",
        help="Emit machine-readable JSON events on stdout.",
    )
    parser.set_defaults(func=run, json_out=False)

# ── Command entry point ───────────────────────────────────────────────────────

def run(args: argparse.Namespace) -> None:
    """Stream coordination events from ``.muse/coordination/`` as NDJSON.

    Emits one JSON object per line for every coordination record that is added,
    modified, removed, or expired.  Uses kqueue on macOS/BSD (zero CPU between
    events) and a configurable polling fallback on Linux.  SIGTERM maps to
    SystemExit(0) for clean process-manager shutdown.

    Agent quickstart::

        muse coord watch --json                       # stream all events forever
        muse coord watch --once --json                # snapshot current state, exit
        muse coord watch --kind reservation --json    # reservation events only
        muse coord watch --run-id agent-42 --json     # filter by agent run-id
        muse coord watch --timeout 60 --json          # exit after 60 seconds

    JSON fields (NDJSON — one object per line)::

        schema_version  Muse version string that produced this event.
        event_type      "snapshot", "added", "modified", "removed", or "expired".
        kind            "reservation", "intent", "release", or "heartbeat".
        id              content-addressed ID of the changed coordination record.
        timestamp       ISO-8601 UTC timestamp when the event was generated.
        data            Parsed record content; empty dict for removed events with no cache.

    Exit codes::

        0  Normal exit (--once, --timeout, --max-events, SIGINT, or SIGTERM).
        1  Bad arguments or backend initialisation failure.
    """
    as_json: bool = args.json_out
    kind_filter: str | None = args.kind
    run_id_filter: str | None = args.run_id
    branch_filter: str | None = args.branch_filter
    once: bool = args.once
    timeout: float | None = args.timeout
    max_events: int | None = args.max_events
    poll_interval: float = args.poll_interval

    # ── Input validation (before any file I/O) ────────────────────────────────

    if run_id_filter is not None and len(run_id_filter) > _MAX_RUN_ID_LEN:
        msg = f"--run-id is too long ({len(run_id_filter)} chars; max {_MAX_RUN_ID_LEN})"
        if as_json:
            print(json.dumps({"error": msg, "status": "bad_args"}))
        else:
            print(f"❌ {msg}", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    if not (_MIN_POLL_INTERVAL <= poll_interval <= _MAX_POLL_INTERVAL):
        msg = (
            f"--poll-interval must be in [{_MIN_POLL_INTERVAL}, {_MAX_POLL_INTERVAL}],"
            f" got {poll_interval}"
        )
        if as_json:
            print(json.dumps({"error": msg, "status": "bad_args"}))
        else:
            print(f"❌ {msg}", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    if timeout is not None and timeout < 0.0:
        msg = f"--timeout must be >= 0, got {timeout}"
        if as_json:
            print(json.dumps({"error": msg, "status": "bad_args"}))
        else:
            print(f"❌ {msg}", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    if max_events is not None and max_events < 1:
        msg = f"--max-events must be >= 1, got {max_events}"
        if as_json:
            print(json.dumps({"error": msg, "status": "bad_args"}))
        else:
            print(f"❌ {msg}", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    # --timeout 0 is equivalent to --once.
    if timeout is not None and timeout == 0.0:
        once = True
        timeout = None

    root = require_repo()

    # Ensure all coord dirs exist before we try to open or watch them.
    _ensure_coord_dirs(root)

    # Select and initialise the backend.
    try:
        if hasattr(select, "kqueue"):
            dirs = [_coord_dir(root) / sub for sub in _SUBDIRS]
            backend: _Backend = _KqueueBackend(dirs)
        else:
            backend = _PollingBackend(poll_interval)
    except (OSError, ValueError) as exc:
        _err(f"Failed to initialise watch backend: {exc}")
        raise SystemExit(ExitCode.USER_ERROR)

    # Register SIGTERM → SystemExit(0) so process managers can stop us cleanly.
    prev_sigterm = signal.getsignal(signal.SIGTERM)

    def _handle_sigterm(sig: int, frame: types.FrameType | None) -> None:  # noqa: ARG001
        raise SystemExit(0)

    signal.signal(signal.SIGTERM, _handle_sigterm)

    if not as_json:
        backend_label = backend.name
        print(
            f"\nmuse coord watch — watching .muse/coordination/ ({backend_label})"
        )
        parts: list[str] = []
        if kind_filter:
            parts.append(f"kind={kind_filter}")
        if run_id_filter:
            parts.append(f"run-id={sanitize_display(run_id_filter)}")
        if branch_filter:
            parts.append(f"branch={sanitize_display(branch_filter)}")
        if max_events is not None:
            parts.append(f"max-events={max_events}")
        if parts:
            print(f"  filter: {', '.join(parts)}")
        print("─" * 62)

    try:
        _watch_loop(
            root,
            backend,
            kind_filter=kind_filter,
            run_id_filter=run_id_filter,
            branch_filter=branch_filter,
            as_json=as_json,
            once=once,
            timeout=timeout,
            poll_interval=poll_interval,
            max_events=max_events,
        )
    except KeyboardInterrupt:
        pass
    finally:
        backend.close()
        signal.signal(signal.SIGTERM, prev_sigterm)
        if not as_json:
            print("\n  (watch ended)", flush=True)

# ── Helpers ───────────────────────────────────────────────────────────────────

def _err(msg: str) -> None:
    """Print an error message to *stderr* with the ``❌`` prefix.

    Used for backend initialisation failures and other pre-loop errors.
    Validation failures in :func:`run` print their own messages before calling
    ``raise SystemExit``.
    """
    print(f"❌ {msg}", file=sys.stderr)