"""``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/<uuid>.json``
``intent``        — ``.muse/coordination/intents/<uuid>.json``
``release``       — ``.muse/coordination/releases/<uuid>.json``
``heartbeat``     — ``.muse/coordination/heartbeats/<uuid>.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,       // UUID 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
"""

from __future__ import annotations

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

from muse._version import __version__
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 root / ".muse" / "coordination"


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 → {uuid → (mtime_ns, size)}
_Snapshot = dict[str, dict[str, tuple[int, int]]]


@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:
        UUID 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: {uuid: (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, uuid)`` tuples:

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

    Results are sorted by kind (canonical ``_SUBDIRS`` order) and then by uuid
    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"
    try:
        return json.loads(path.read_text())
    except (OSError, json.JSONDecodeError):
        return None


# ── 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=datetime.datetime.now(datetime.timezone.utc).isoformat(),
        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 UUID 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_short = sanitize_display(ev.id[:8])
    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_short}", 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_short}  [{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_short}  {op}  {first_addr}",
            flush=True,
        )

    elif ev.kind == "release":
        reason = sanitize_display(data.get("reason", "?"))
        res_id = sanitize_display((data.get("reservation_id") or "")[:8])
        print(
            f"{icon} {kind_col} {run_id:<32}  {uid_short}  {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 "")[:8])
        print(
            f"{icon} {kind_col} {run_id:<32}  {uid_short}  extends to {ext}  (res: {res_id})",
            flush=True,
        )

    else:
        print(f"{icon} {kind_col} {agent_str:<32}  {uid_short}", 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: {uuid: (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, uuid) → 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 "
            "(" + " | ".join(sorted(_VALID_KINDS)) + ")."
        ),
    )
    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 entry point ───────────────────────────────────────────────────────


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

    Execution order
    ---------------
    1. **Validate inputs** — ``--run-id`` length, ``--poll-interval`` range,
       ``--timeout`` sign, ``--max-events`` sign.  All validation fires before
       ``require_repo()`` or any file I/O; failures exit
       :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message to
       *stderr* (or compact JSON when ``--json`` is set).
    2. **Resolve repo** — ``require_repo()`` finds ``.muse/``.
    3. **Create coord dirs** — ``_ensure_coord_dirs()`` creates any missing
       subdirectories so the backend can open them immediately.
    4. **Initialise backend** — kqueue on macOS/BSD, polling everywhere else.
    5. **Register SIGTERM** — maps to ``SystemExit(0)`` for clean process-
       manager shutdown.
    6. **Run watch loop** — ``_watch_loop()`` handles scan → diff → emit.

    Backend selection
    -----------------
    On macOS/BSD (where ``select.kqueue`` is available), a kqueue backend is
    used: the process blocks in the kernel until a file is added, replaced, or
    deleted in one of the four coordination subdirectories.  CPU usage between
    events is effectively zero.

    On Linux and other platforms, a polling backend is used: the process
    sleeps for ``--poll-interval`` seconds and then rescans all directories.

    Both backends share the exact same scan → diff → emit logic.

    Signal handling
    ---------------
    ``SIGINT`` (Ctrl-C) raises :exc:`KeyboardInterrupt` which exits the loop
    cleanly.  ``SIGTERM`` (sent by process managers or test harnesses) is
    mapped to :exc:`SystemExit(0)` via a signal handler registered in this
    function and restored to ``SIG_DFL`` on exit.

    Security
    --------
    * ``--run-id`` is capped at :data:`_MAX_RUN_ID_LEN` to bound memory use.
    * ``--poll-interval`` is validated to
      ``[_MIN_POLL_INTERVAL, _MAX_POLL_INTERVAL]`` before any I/O.
    * All strings printed to stdout are passed through
      :func:`~muse.core.validation.sanitize_display`.
    * The kqueue backend checks each coord directory for symlinks before
      opening any fd.
    * ``--kind`` is validated against a frozenset allowlist by argparse.

    Performance
    -----------
    kqueue: wakes only on FS changes — no CPU overhead between events.
    polling: ``--poll-interval`` controls wake frequency; lower values give
    faster detection at the cost of more directory I/O.

    Args:
        args: Parsed ``argparse.Namespace`` with attributes ``once``,
            ``timeout``, ``max_events``, ``poll_interval``, ``run_id``,
            ``branch_filter``, ``kind``, and ``fmt``.

    Exit codes:
        0 — normal exit (``--once``, ``--timeout``, ``--max-events``,
            SIGINT, or SIGTERM).
        1 — bad arguments or backend initialisation failure.
    """
    as_json: bool = args.fmt == "json"
    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)