"""Watch mode: poll for file changes and auto-re-index.

Output formats: text (default), ``--json``. SARIF is deliberately NOT
emitted because ``roam watch`` is a long-running process, not a stateless
invocation; SARIF emission is incompatible with the stateless one-shot
schema. The watcher's signals are consumed by other roam commands +
agent runtimes directly from disk on the next invocation; SARIF would
be redundant. See action.yml _SUPPORTED_SARIF allowlist + W1181-audit
memo.
"""

from __future__ import annotations

import json
import queue
import sqlite3
import threading
import time
import urllib.parse
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import click

from roam.capability import roam_capability
from roam.commands.resolve import ensure_index
from roam.db.connection import db_exists, find_project_root, open_db


def load_tracked_files(project_root: Path) -> dict[str, float]:
    """Return {relative_path: mtime} for all files currently in the DB.

    Returns an empty dict when no index exists.
    """
    if not db_exists(project_root):
        return {}
    try:
        with open_db(readonly=True, project_root=project_root) as conn:
            rows = conn.execute("SELECT path, mtime FROM files").fetchall()
        return {row[0]: (row[1] or 0.0) for row in rows}
    except (click.ClickException, sqlite3.Error):
        return {}


def scan_disk_mtimes(
    file_paths: list[str],
    project_root: Path,
) -> dict[str, float]:
    """Return {relative_path: mtime} for paths that exist on disk.

    Missing files are omitted from the result.
    """
    result: dict[str, float] = {}
    for rel in file_paths:
        try:
            result[rel] = (project_root / rel).stat().st_mtime
        except OSError:
            pass
    return result


def detect_changes(
    tracked: dict[str, float],
    current_disk: dict[str, float],
) -> tuple[list[str], list[str], list[str]]:
    """Compare tracked DB state with current disk state.

    Args:
        tracked:      {path: stored_mtime}  -- from the index DB.
        current_disk: {path: disk_mtime}    -- from the filesystem right now.

    Returns:
        (added, modified, removed) -- three lists of relative paths.
        added    = paths on disk but not in index.
        modified = paths in both but mtime differs by > 0.001 s.
        removed  = paths in index but no longer on disk.
    """
    tracked_set = set(tracked)
    disk_set = set(current_disk)

    added = sorted(disk_set - tracked_set)
    removed = sorted(tracked_set - disk_set)
    modified = sorted(p for p in tracked_set & disk_set if abs(current_disk[p] - tracked[p]) > 0.001)
    return added, modified, removed


def discover_current_files(project_root: Path) -> list[str]:
    """Discover source files on disk via the same logic as the indexer.

    Falls back to an empty list on filesystem errors so the watcher degrades
    gracefully when called before the project is fully initialised.
    """
    try:
        from roam.index.discovery import discover_files

        return discover_files(project_root)
    except OSError:
        return []


def run_incremental_index(project_root: Path, quiet: bool, force: bool = False) -> None:
    """Trigger an index refresh.

    Args:
        project_root: repository root.
        quiet: suppress index progress output.
        force: run a full rebuild when True.
    """
    from roam.index.indexer import Indexer

    indexer = Indexer(project_root=project_root)
    indexer.run(force=force, quiet=quiet)


def _guardian_drift_summary(
    conn,
    project_root: Path,
    threshold: float = 0.5,
    max_files: int = 250,
) -> dict:
    """Return compact ownership-drift summary for architecture guardian mode."""
    from roam.commands.cmd_drift import compute_drift_score, compute_file_ownership
    from roam.commands.codeowners_helpers import find_codeowners, parse_codeowners, resolve_owners

    co_path = find_codeowners(project_root)
    if co_path is None:
        return {
            "codeowners_found": False,
            "owned_files": 0,
            "drift_files": 0,
            "drift_pct": 0.0,
            "threshold": threshold,
        }

    rules = parse_codeowners(co_path)
    file_rows = conn.execute(
        """
        SELECT f.id, f.path
        FROM files f
        LEFT JOIN file_stats fs ON fs.file_id = f.id
        ORDER BY COALESCE(fs.total_churn, 0) DESC, f.path
        LIMIT ?
        """,
        (max_files,),
    ).fetchall()
    if not file_rows:
        return {
            "codeowners_found": True,
            "owned_files": 0,
            "drift_files": 0,
            "drift_pct": 0.0,
            "threshold": threshold,
        }

    now_ts = int(time.time())
    owned = 0
    drift = 0
    for row in file_rows:
        path = str(row["path"]).replace("\\", "/")
        owners = resolve_owners(rules, path)
        if not owners:
            continue
        owned += 1
        shares = compute_file_ownership(conn, int(row["id"]), now_ts=now_ts)
        if not shares:
            continue
        score = compute_drift_score(owners, shares)
        if score >= threshold:
            drift += 1

    pct = round((drift * 100.0 / owned), 1) if owned else 0.0
    return {
        "codeowners_found": True,
        "owned_files": owned,
        "drift_files": drift,
        "drift_pct": pct,
        "threshold": threshold,
    }


def collect_guardian_snapshot(
    project_root: Path,
    *,
    health_gate: float = 70.0,
    drift_threshold: float = 0.5,
) -> dict:
    """Collect a continuous architecture-guardian snapshot."""
    from roam.commands.cmd_trends import _analyze_trends, _trend_verdict
    from roam.commands.metrics_history import append_snapshot, get_snapshots

    with open_db(readonly=False, project_root=project_root) as conn:
        current = append_snapshot(conn, source="watch-guardian")
        snaps = get_snapshots(conn, limit=30)
        drift = _guardian_drift_summary(
            conn,
            project_root=project_root,
            threshold=drift_threshold,
        )

    chrono = []
    for snap in reversed(snaps):
        chrono.append(
            {
                "files": snap["files"],
                "symbols": snap["symbols"],
                "edges": snap["edges"],
                "cycles": snap["cycles"],
                "god_components": snap["god_components"],
                "bottlenecks": snap["bottlenecks"],
                "dead_exports": snap["dead_exports"],
                "layer_violations": snap["layer_violations"],
                "health_score": snap["health_score"],
            }
        )

    trend = {
        "verdict": "insufficient-data",
        "anomaly_count": 0,
        "significant_trends": 0,
    }
    if len(chrono) >= 4:
        analysis = _analyze_trends(chrono, sensitivity="medium")
        trend = {
            "verdict": _trend_verdict(analysis),
            "anomaly_count": len(analysis.get("anomalies", [])),
            "significant_trends": len([t for t in analysis.get("trends", []) if t.get("direction") != "stable"]),
        }

    health_score = int(current.get("health_score", 0) or 0)
    gates = {
        "health_gate": health_gate,
        "health_gate_pass": health_score >= health_gate,
    }
    return {
        "timestamp": (datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")),
        "source": "watch-guardian",
        "current": {
            "health_score": health_score,
            "cycles": int(current.get("cycles", 0) or 0),
            "layer_violations": int(current.get("layer_violations", 0) or 0),
            "dead_exports": int(current.get("dead_exports", 0) or 0),
            "tangle_ratio": float(current.get("tangle_ratio", 0.0) or 0.0),
            "avg_complexity": float(current.get("avg_complexity", 0.0) or 0.0),
            "brain_methods": int(current.get("brain_methods", 0) or 0),
        },
        "trend": trend,
        "drift": drift,
        "gates": gates,
    }


def append_guardian_report(report_path: Path, payload: dict) -> None:
    """Append guardian snapshots as JSONL for compliance/report pipelines."""
    report_path.parent.mkdir(parents=True, exist_ok=True)
    with report_path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(payload, sort_keys=True, ensure_ascii=True) + "\n")


class WebhookBridge:
    """Tiny local HTTP bridge that queues index-refresh triggers.

    Exposes:
      POST {path}  -> enqueue trigger event
      GET  /health -> daemon/webhook status
    """

    def __init__(
        self,
        host: str,
        port: int,
        path: str = "/roam/reindex",
        secret: str = "",
    ) -> None:
        self.host = host
        self.port = port
        self.path = path if path.startswith("/") else f"/{path}"
        self.secret = secret.strip()
        self._events: queue.Queue[dict] = queue.Queue()
        self._lock = threading.Lock()
        self._accepted = 0
        self._started_at = time.time()
        self._server: ThreadingHTTPServer | None = None
        self._thread: threading.Thread | None = None

    def _auth_ok(self, headers: dict[str, str]) -> bool:
        """Validate webhook secret (if configured)."""
        if not self.secret:
            return True
        token = headers.get("x-roam-secret", "").strip()
        if token == self.secret:
            return True
        bearer = headers.get("authorization", "").strip()
        if bearer.lower().startswith("bearer "):
            return bearer[7:].strip() == self.secret
        return False

    def _enqueue(self, event: str, force: bool = False) -> None:
        with self._lock:
            self._accepted += 1
        self._events.put(
            {
                "event": event or "webhook",
                "force": bool(force),
                "received_at": time.time(),
            }
        )

    def drain_events(self) -> list[dict]:
        """Drain queued trigger events."""
        out: list[dict] = []
        while True:
            try:
                out.append(self._events.get_nowait())
            except queue.Empty:
                break
        return out

    def bridge_stats(self) -> dict:
        with self._lock:
            accepted = self._accepted
        return {
            "host": self.host,
            "port": self.port,
            "path": self.path,
            "queued_events": self._events.qsize(),
            "accepted_events": accepted,
            "uptime_s": round(max(0.0, time.time() - self._started_at), 3),
            "secret_required": bool(self.secret),
        }

    def _build_handler(self):
        bridge = self

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, fmt, *args):  # noqa: D401, ANN001
                # Keep daemon mode quiet unless the user asked for verbose logs.
                return

            def send_response(self, status: int) -> None:
                # Explicit wrappers so these inherited stdlib methods resolve.
                super().send_response(status)

            def send_header(self, keyword: str, value: str) -> None:
                super().send_header(keyword, value)

            def end_headers(self) -> None:
                super().end_headers()

            def _json(self, status: int, payload: dict) -> None:
                body = json.dumps(payload, sort_keys=True).encode("utf-8")
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

            def do_GET(self):  # noqa: N802
                parsed = urllib.parse.urlparse(self.path)
                if parsed.path == "/health":
                    payload = {"ok": True, **bridge.bridge_stats()}
                    self._json(200, payload)
                    return
                self._json(404, {"ok": False, "error": "not found"})

            def do_POST(self):  # noqa: N802
                parsed = urllib.parse.urlparse(self.path)
                if parsed.path != bridge.path:
                    self._json(404, {"ok": False, "error": "not found"})
                    return

                headers = {k.lower(): v for k, v in self.headers.items()}
                if not bridge._auth_ok(headers):
                    self._json(401, {"ok": False, "error": "unauthorized"})
                    return

                length = 0
                try:
                    length = int(self.headers.get("Content-Length", "0"))
                except (TypeError, ValueError):
                    length = 0
                raw = self.rfile.read(max(0, min(length, 1_000_000))) if length > 0 else b""

                payload: dict = {}
                if raw:
                    try:
                        decoded = json.loads(raw.decode("utf-8"))
                        if isinstance(decoded, dict):
                            payload = decoded
                    except (UnicodeDecodeError, json.JSONDecodeError):
                        payload = {}

                event = (
                    headers.get("x-roam-event")
                    or headers.get("x-github-event")
                    or str(payload.get("event", "")).strip()
                    or "webhook"
                )
                force = bool(payload.get("force", False))
                bridge._enqueue(event=event, force=force)
                self._json(
                    202,
                    {
                        "ok": True,
                        "accepted": True,
                        "event": event,
                        "force": force,
                        "queued_events": bridge._events.qsize(),
                    },
                )

        return Handler

    def start(self) -> None:
        """Start the local webhook HTTP server in a background thread."""
        if self._server is not None:
            return
        handler = self._build_handler()
        self._server = ThreadingHTTPServer((self.host, self.port), handler)
        self.port = self._server.server_address[1]
        self._thread = threading.Thread(
            target=self._server.serve_forever,
            name="roam-webhook-bridge",
            daemon=True,
        )
        self._thread.start()

    def shutdown(self) -> None:
        """Stop the background webhook server."""
        if self._server is None:
            return
        try:
            self._server.shutdown()
            self._server.server_close()
        finally:
            self._server = None
        if self._thread is not None:
            self._thread.join(timeout=2.0)
            self._thread = None


class DebounceAccumulator:
    """Accumulate file-change events and fire after a quiet period.

    Call add() with changed paths each poll cycle, then check
    should_fire() to know when the debounce window has elapsed.
    """

    def __init__(self, window: float = 1.0) -> None:
        self.window = window
        self._pending: set[str] = set()
        self._last_change: float | None = None

    def add(self, paths: list[str]) -> None:
        """Record newly changed paths and reset the quiet-period timer."""
        if paths:
            self._pending.update(paths)
            self._last_change = time.monotonic()

    def has_pending(self) -> bool:
        """Return True if there are accumulated changes waiting to fire."""
        return bool(self._pending)

    def should_fire(self, now: float) -> bool:
        """Return True when the quiet window has elapsed since the last change."""
        if not self._pending or self._last_change is None:
            return False
        return (now - self._last_change) >= self.window

    def flush(self) -> list[str]:
        """Return accumulated paths and reset internal state."""
        result = sorted(self._pending)
        self._pending.clear()
        self._last_change = None
        return result


def _need_force(webhook_events: list, webhook_force: bool) -> bool:
    """does the current event batch require a forced re-index?"""
    has_explicit = any(bool(evt.get("force")) for evt in webhook_events if isinstance(evt, dict))
    return has_explicit or (webhook_force and bool(webhook_events))


def _scan_disk_changes(_discover, project_root: Path, tracked: dict, quiet: bool):
    """discover + diff disk state, optionally print per-file lines."""
    current_paths = _discover()
    current_disk = scan_disk_mtimes(current_paths, project_root)
    added, modified, removed = detect_changes(tracked, current_disk)
    changed = added + modified + removed
    if changed and not quiet:
        for path in added:
            click.echo(f"  + {path}")
        for path in modified:
            click.echo(f"  ~ {path}")
        for path in removed:
            click.echo(f"  - {path}")
    return tracked, changed


def _label_webhook_events(webhook_events: list, quiet: bool) -> list[str]:
    """turn webhook event dicts into ``<webhook:name>`` labels."""
    if not webhook_events:
        return []
    labels = [
        f"<webhook:{str(evt.get('event', 'webhook')).strip() or 'webhook'}>"
        for evt in webhook_events
        if isinstance(evt, dict)
    ]
    if labels and not quiet:
        for label in labels:
            click.echo(f"  * {label}")
    return labels


def _refresh_tracked_after_reindex(project_root: Path, prev_file_count: int, quiet: bool):
    """reload tracked-files state and emit a status line."""
    tracked = load_tracked_files(project_root)
    new_count = len(tracked)
    if not quiet:
        click.echo(f"Re-index complete. Watching {new_count} files.")
    elif new_count != prev_file_count:
        click.echo(f"Re-indexed. Watching {new_count} files.")
    return tracked, new_count


def _run_guardian_step(_guardian_collect, _guardian_write, guardian_report: str, quiet: bool) -> None:
    """collect + write a guardian snapshot, log the verdict."""
    try:
        payload = _guardian_collect()
        if guardian_report:
            _guardian_write(payload)
        if quiet:
            return
        gate_state = "PASS" if payload.get("gates", {}).get("health_gate_pass") else "FAIL"
        click.echo(
            "Guardian: health={} trend={} drift={} ({})".format(
                payload.get("current", {}).get("health_score", "n/a"),
                payload.get("trend", {}).get("verdict", "n/a"),
                payload.get("drift", {}).get("drift_files", "n/a"),
                gate_state,
            )
        )
    except Exception as exc:
        if not quiet:
            click.echo(f"Guardian update failed: {exc}")


@dataclass
class _PollDeps:
    """Injectable callables for poll_loop.

    Build with for_project() for production; pass stubs directly in tests.
    """

    sleep: Callable = time.sleep
    discover: Callable = lambda: []
    reindex: Callable = lambda force=False: None
    external_events: Callable = lambda: []
    guardian_collect: Callable = lambda: {}
    guardian_write: Callable = lambda payload: None

    @classmethod
    def for_project(
        cls,
        project_root: Path,
        guardian_health_gate: float,
        guardian_drift_threshold: float,
        guardian_report: str,
        *,
        external_events: Callable | None = None,
    ) -> _PollDeps:
        return cls(
            sleep=time.sleep,
            discover=lambda: discover_current_files(project_root),
            reindex=lambda force=False: run_incremental_index(project_root, quiet=True, force=force),
            external_events=external_events or (lambda: []),
            guardian_collect=lambda: collect_guardian_snapshot(
                project_root=project_root,
                health_gate=guardian_health_gate,
                drift_threshold=guardian_drift_threshold,
            ),
            guardian_write=lambda payload: (
                append_guardian_report(Path(guardian_report), payload) if guardian_report else None
            ),
        )


def _on_debounce_fire(
    acc: DebounceAccumulator,
    pending_force: bool,
    quiet: bool,
    guardian: bool,
    guardian_report: str,
    project_root: Path,
    file_count: int,
    deps: _PollDeps,
) -> tuple[dict[str, float], int, bool]:
    """Execute a debounced re-index + optional guardian snapshot. Returns (tracked, file_count, pending_force=False)."""
    batch = acc.flush()
    if not quiet:
        mode_label = "force re-indexing" if pending_force else "re-indexing"
        click.echo(f"Changed: {len(batch)} event(s) -- {mode_label}...")
    deps.reindex(force=pending_force)
    tracked, file_count = _refresh_tracked_after_reindex(project_root, file_count, quiet)
    if guardian or guardian_report:
        _run_guardian_step(deps.guardian_collect, deps.guardian_write, guardian_report, quiet)
    return tracked, file_count, False


def poll_loop(
    project_root: Path,
    interval: float,
    debounce: float,
    quiet: bool,
    webhook_only: bool = False,
    webhook_force: bool = False,
    guardian: bool = False,
    guardian_report: str = "",
    *,
    deps: _PollDeps | None = None,
) -> None:
    """Run the watch loop until interrupted.

    Args:
        project_root:  Root directory to watch.
        interval:      Poll interval in seconds.
        debounce:      Quiet-period window in seconds before triggering re-index.
        quiet:         Suppress per-file change messages when True.
        webhook_only:  When True, skip file polling and only process webhook triggers.
        webhook_force: Force full rebuild for webhook-triggered refreshes.
        guardian:      Enable continuous architecture guardian snapshots.
        guardian_report:
                       Optional JSONL path for compliance-ready guardian artifacts.
        deps:          Injectable callables; build with _PollDeps.for_project() for
                       production, or pass _PollDeps with stubs in tests.
    """
    if deps is None:
        deps = _PollDeps.for_project(project_root, 70.0, 0.5, guardian_report)

    acc = DebounceAccumulator(window=debounce)

    # Seed initial state from what is already tracked in the DB
    tracked = load_tracked_files(project_root)

    file_count = len(tracked)
    mode = "webhook-only" if webhook_only else "poll+webhook"
    click.echo(f"Watching {file_count} files... (interval={interval}s, debounce={debounce}s, mode={mode})")
    click.echo("Press Ctrl+C to stop.")
    pending_force = False

    while True:
        deps.sleep(interval)

        webhook_events = deps.external_events() or []
        pending_force = pending_force or _need_force(webhook_events, webhook_force)

        changed: list[str] = []
        if not webhook_only:
            tracked, disk_changes = _scan_disk_changes(deps.discover, project_root, tracked, quiet)
            changed.extend(disk_changes)
        changed.extend(_label_webhook_events(webhook_events, quiet))

        if changed:
            acc.add(changed)

        if acc.should_fire(time.monotonic()):
            tracked, file_count, pending_force = _on_debounce_fire(
                acc,
                pending_force,
                quiet,
                guardian,
                guardian_report,
                project_root,
                file_count,
                deps,
            )


@roam_capability(
    name="watch",
    category="getting-started",
    summary="Watch for file changes and auto-re-index incrementally",
    maturity="stable",
    mcp_expose=False,
    mcp_preset=("core",),
    side_effect=True,
    task_required=False,
    destructive=False,
    stale_sensitive=False,
    ai_safe=True,
    requires_index=False,
)
@click.command("watch")
@click.option(
    "--interval",
    "-i",
    default=2.0,
    show_default=True,
    type=float,
    help="Poll interval in seconds.",
)
@click.option(
    "--debounce",
    "-d",
    default=1.0,
    show_default=True,
    type=float,
    help="Quiet-period window before triggering re-index (seconds).",
)
@click.option(
    "--quiet",
    "-q",
    is_flag=True,
    help="Suppress per-file change messages.",
)
@click.option(
    "--webhook-port",
    default=0,
    show_default=True,
    type=int,
    help="Enable webhook bridge HTTP listener on this port (0=disabled).",
)
@click.option(
    "--webhook-host",
    default="127.0.0.1",
    show_default=True,
    help="Webhook bridge bind host.",
)
@click.option(
    "--webhook-path",
    default="/roam/reindex",
    show_default=True,
    help="Webhook POST path used to trigger re-index.",
)
@click.option(
    "--webhook-secret",
    default="",
    help="Optional shared secret (X-Roam-Secret or Bearer token).",
)
@click.option(
    "--webhook-only",
    is_flag=True,
    help="Disable file polling and only react to webhook triggers.",
)
@click.option(
    "--webhook-force",
    is_flag=True,
    help="Force full reindex for webhook-triggered events.",
)
@click.option(
    "--guardian",
    is_flag=True,
    help="Enable continuous architecture guardian snapshots after each re-index.",
)
@click.option(
    "--guardian-report",
    default="",
    help="Optional JSONL file path to append guardian compliance snapshots.",
)
@click.option(
    "--guardian-health-gate",
    default=70.0,
    show_default=True,
    type=float,
    help="Guardian health score gate threshold (tracked in snapshots).",
)
@click.option(
    "--guardian-drift-threshold",
    default=0.5,
    show_default=True,
    type=float,
    help="Ownership drift threshold used in guardian summaries.",
)
@click.pass_context
def watch(
    ctx,
    interval,
    debounce,
    quiet,
    webhook_port,
    webhook_host,
    webhook_path,
    webhook_secret,
    webhook_only,
    webhook_force,
    guardian,
    guardian_report,
    guardian_health_gate,
    guardian_drift_threshold,
):
    """Watch for file changes and auto-re-index incrementally.

    Uses polling (no external dependencies).  Debounces rapid bursts of
    changes and only re-indexes what actually changed.
    Optional webhook bridge mode allows CI/webhook systems to trigger
    refreshes without shelling into the process.

    Use ``--guardian`` for continuous architecture health snapshots
    (JSONL audit trail of health score, cycles, drift, and layer
    violations after each re-index).

    Press Ctrl+C to stop.

    \b
    Examples:
      roam watch
      roam watch --interval 5 --debounce 2
      roam watch --guardian --guardian-health-gate 80
      roam watch --webhook-port 8765 --webhook-secret mytoken

    See also ``index`` (one-shot reindex), ``hooks`` (git hook setup),
    and ``health`` (the score the guardian gate enforces).
    """
    project_root = find_project_root()
    bridge: WebhookBridge | None = None

    # Ensure an index exists before we start watching.
    # In webhook-only mode we still ensure an index exists so triggers
    # can refresh an already valid baseline.
    ensure_index(quiet=quiet)

    if webhook_port:
        bridge = WebhookBridge(
            host=webhook_host,
            port=webhook_port,
            path=webhook_path,
            secret=webhook_secret,
        )
        bridge.start()
        if not quiet:
            click.echo(f"Webhook bridge listening on http://{webhook_host}:{bridge.port}{bridge.path}")
            if webhook_secret:
                click.echo("Webhook auth: secret required via X-Roam-Secret or Bearer token.")
    if (guardian or guardian_report) and not quiet:
        click.echo(
            "Architecture guardian enabled "
            f"(health_gate={guardian_health_gate}, drift_threshold={guardian_drift_threshold})"
        )
        if guardian_report:
            click.echo(f"Guardian report: {guardian_report}")

    try:
        poll_loop(
            project_root=project_root,
            interval=interval,
            debounce=debounce,
            quiet=quiet,
            webhook_only=webhook_only,
            webhook_force=webhook_force,
            guardian=guardian,
            guardian_report=guardian_report,
            deps=_PollDeps.for_project(
                project_root,
                guardian_health_gate,
                guardian_drift_threshold,
                guardian_report,
                external_events=(bridge.drain_events if bridge else None),
            ),
        )
    except KeyboardInterrupt:
        click.echo("\nWatch stopped.")
    finally:
        if bridge is not None:
            bridge.shutdown()
