fix(gateway): single-owner gate + zero-sub early exit for kanban notifier

gateway/kanban_watchers.py

@@ -58,15 +58,17 @@ def _resolve_auto_decompose_settings(
 
 
 def _acquire_singleton_lock(lock_path) -> "tuple[Optional[object], str]":
-    """Take an exclusive, non-blocking advisory lock for the sole dispatcher.
+    """Take an exclusive, non-blocking advisory lock for a singleton loop.
 
     Only one gateway process machine-wide may run the embedded kanban
-    dispatcher: concurrent dispatchers double the reclaim frequency (each
-    runs its own ``release_stale_claims`` → promote → dispatch loop), double
-    claim-attempt events in the event log, and — with ``wal_autocheckpoint=0`` —
-    concurrent manual WAL checkpoints can corrupt index pages. The
-    ``dispatch_in_gateway`` config flag is the primary control; this lock is the
-    backstop that survives config drift and same-profile restart races.
+    dispatcher (``.dispatcher.lock``): concurrent dispatchers double the reclaim
+    frequency (each runs its own ``release_stale_claims`` → promote → dispatch
+    loop), double claim-attempt events in the event log, and — with
+    ``wal_autocheckpoint=0`` — concurrent manual WAL checkpoints can corrupt
+    index pages. The notifier uses the same helper with its own
+    ``.notifier.lock`` so only one gateway polls board DBs for notifications.
+    The ``dispatch_in_gateway`` config flag is the primary control; the lock is
+    the backstop that survives config drift and same-profile restart races.
 
     Delegates to :func:`gateway.status._try_acquire_file_lock` (``fcntl`` on
     POSIX, ``msvcrt`` on Windows) so the guard is cross-platform.
@@ -95,7 +97,7 @@ def _acquire_singleton_lock(lock_path) -> "tuple[Optional[object], str]":
 
 
 def _release_singleton_lock(handle) -> None:
-    """Release a dispatcher singleton lock acquired via :func:`_acquire_singleton_lock`."""
+    """Release a singleton lock acquired via :func:`_acquire_singleton_lock`."""
     if handle is None:
         return
     try:
@@ -131,10 +133,13 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None:
         cross boards, so delivery semantics are unchanged — this is
         purely a fan-out of the single-DB poll.
         """
-        # Gate: only the dispatch-owning gateway opens kanban DBs for notifier polling.
-        # Non-dispatch gateways have no subscriptions to deliver — all kanban state lives
-        # in the dispatch owner's per-board DBs. This prevents N-gateway -shm contention.
-        # TODO: gate per-board when per-board dispatcher_owner tracking lands.
+        # Gate order mirrors `_kanban_dispatcher_watcher`: the
+        # HERMES_KANBAN_DISPATCH_IN_GATEWAY env override, then the
+        # `kanban.dispatch_in_gateway` config flag, then the machine-global
+        # singleton `.notifier.lock` below — so only ONE gateway machine-wide
+        # polls kanban DBs for notifications. Per-board work is further gated
+        # by a read-only subscription probe, so boards with zero subscriptions
+        # are never opened writable (no schema migration, no checkpoints).
         try:
             from hermes_cli.config import load_config as _load_config
         except Exception:
@@ -162,6 +167,32 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None:
             logger.warning("kanban notifier: kanban_db not importable; notifier disabled")
             return
 
+        # Single-notifier backstop, mirroring the dispatcher's singleton lock:
+        # `dispatch_in_gateway` defaults to true, so without it every gateway
+        # polls every board DB each tick — the exact N-gateway -shm/-wal
+        # contention this gate exists to prevent. The lock lives at the
+        # machine-global kanban root (shared across profiles by design), so it
+        # serialises ALL gateways; a `.notifier.lock` separate from
+        # `.dispatcher.lock` lets a gateway notify without dispatching.
+        self._kanban_notifier_lock_handle = None
+        _lock_path = _kb.kanban_home() / "kanban" / ".notifier.lock"
+        _lock_handle, _lock_state = _acquire_singleton_lock(_lock_path)
+        if _lock_state == "contended":
+            logger.info(
+                "kanban notifier: another gateway already holds the notifier "
+                "lock (%s); this gateway will NOT poll for notifications.",
+                _lock_path,
+            )
+            return
+        if _lock_state == "held":
+            self._kanban_notifier_lock_handle = _lock_handle  # hold for process lifetime
+            logger.info("kanban notifier: holding singleton notifier lock (%s)", _lock_path)
+        else:
+            logger.warning(
+                "kanban notifier: advisory lock unavailable at %s; proceeding "
+                "on config control alone.", _lock_path,
+            )
+
         # "status" covers dashboard drag-drop and `_set_status_direct()`
         # writes — surface those transitions to subscribers too.
         TERMINAL_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out", "status", "archived", "unblocked")
@@ -230,6 +261,19 @@ def _collect():
                             )
                             continue
                         seen_db_paths.add(resolved_db_path)
+                        try:
+                            if _kb.count_notify_subs(board=slug) == 0:
+                                logger.debug(
+                                    "kanban notifier: board %s has no subscriptions; skipping open",
+                                    slug,
+                                )
+                                continue
+                        except Exception as exc:
+                            logger.debug(
+                                "kanban notifier: read-only subscription probe failed "
+                                "for board %s (%s); falling back to writable open",
+                                slug, exc,
+                            )
                         try:
                             conn = _kb.connect(board=slug)
                         except Exception as exc:
@@ -565,14 +609,22 @@ def _collect():
                             await asyncio.to_thread(
                                 self._kanban_unsub, sub, board_slug,
                             )
+            except asyncio.CancelledError:
+                logger.debug("kanban notifier: cancelled")
+                _release_singleton_lock(self._kanban_notifier_lock_handle)
+                self._kanban_notifier_lock_handle = None
+                raise
             except Exception as exc:
                 logger.warning("kanban notifier tick failed: %s", exc)
             # Sleep with cancellation checks.
             for _ in range(int(max(1, interval))):
                 if not self._running:
-                    return
+                    break
                 await asyncio.sleep(1)
 
+        _release_singleton_lock(self._kanban_notifier_lock_handle)
+        self._kanban_notifier_lock_handle = None
+
     def _kanban_advance(
         self, sub: dict, cursor: int, board: Optional[str] = None,
     ) -> None:

hermes_cli/kanban_db.py

@@ -8308,6 +8308,44 @@ def list_notify_subs(
     return [dict(r) for r in rows]
 
 
+def count_notify_subs(
+    db_path: Optional[Path] = None,
+    *,
+    board: Optional[str] = None,
+) -> int:
+    """Count ``kanban_notify_subs`` rows via a read-only connection.
+
+    Cheap probe for the gateway notifier's zero-subscription early exit:
+    unlike :func:`connect`, this never creates the DB file, never runs
+    schema init/migration, and never opens the database writable (no
+    write locks, no checkpoints — though a read-only open of a WAL
+    database may still create the ``-shm``/``-wal`` sidecars, it cannot
+    write table content). Rows in a not-yet-checkpointed WAL are
+    visible, so a freshly added subscription is never missed. A missing
+    DB, or a legacy DB that predates the subscriptions table, counts as
+    zero. Path resolution matches :func:`connect` (explicit ``db_path``,
+    else ``board`` via :func:`kanban_db_path`). Raises
+    :class:`sqlite3.Error` when the DB exists but cannot be read
+    (locked, corrupt); callers choose their own fallback.
+    """
+    path = db_path if db_path is not None else kanban_db_path(board=board)
+    if not path.exists():
+        return 0
+    conn = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)
+    try:
+        try:
+            row = conn.execute(
+                "SELECT COUNT(*) FROM kanban_notify_subs"
+            ).fetchone()
+        except sqlite3.OperationalError as exc:
+            if "no such table" in str(exc).lower():
+                return 0
+            raise
+        return int(row[0]) if row else 0
+    finally:
+        conn.close()
+
+
 def remove_notify_sub(
     conn: sqlite3.Connection,
     *,

tests/gateway/test_kanban_notifier_owner_gate.py

@@ -0,0 +1,182 @@
+"""Tests for the kanban notifier single-owner gate + zero-subscription skip.
+
+The notifier used to writable-open EVERY board DB from EVERY gateway whose
+config left ``kanban.dispatch_in_gateway`` at the default (true) — the exact
+N-gateway ``-shm``/``-wal`` contention its own top-of-function comment
+claimed to prevent. It now mirrors the dispatcher's machine-global singleton
+advisory lock with its own ``<kanban root>/kanban/.notifier.lock``: the
+lock-losing gateway polls zero boards and opens zero connections. Per-board
+work is further gated by a read-only subscription probe
+(``kanban_db.count_notify_subs``), so boards with zero subscriptions are
+never opened writable.
+"""
+
+import asyncio
+
+from unittest.mock import patch
+
+from gateway.config import Platform
+from gateway.kanban_watchers import _acquire_singleton_lock, _release_singleton_lock
+from gateway.run import GatewayRunner
+from hermes_cli import kanban_db as kb
+
+
+class RecordingAdapter:
+    def __init__(self):
+        self.sent = []
+
+    async def send(self, chat_id, text, metadata=None):
+        self.sent.append({"chat_id": chat_id, "text": text, "metadata": metadata or {}})
+
+
+def _make_runner(adapter):
+    runner = GatewayRunner.__new__(GatewayRunner)
+    runner._running = True
+    runner.adapters = {Platform.TELEGRAM: adapter}
+    runner._kanban_sub_fail_counts = {}
+    return runner
+
+
+async def _run_one_notifier_tick(monkeypatch, runner):
+    real_sleep = asyncio.sleep
+
+    async def fake_sleep(delay):
+        if delay == 5:
+            return None
+        runner._running = False
+        await real_sleep(0)
+
+    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
+    await runner._kanban_notifier_watcher(interval=1)
+
+
+def _create_completed_task(*, subscribe: bool) -> str:
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="owner gate", assignee="worker")
+        if subscribe:
+            kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat-1")
+        kb.complete_task(conn, tid, summary="done")
+        return tid
+    finally:
+        conn.close()
+
+
+def test_zero_sub_board_is_never_opened_writable(tmp_path, monkeypatch):
+    """A board with zero subscriptions must be skipped BEFORE `_kb.connect`."""
+    db_path = tmp_path / "zero-subs.db"
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path))
+    kb.init_db()
+    _create_completed_task(subscribe=False)
+
+    adapter = RecordingAdapter()
+    runner = _make_runner(adapter)
+
+    with patch.object(kb, "connect", wraps=kb.connect) as spy_connect:
+        asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
+
+    spy_connect.assert_not_called()
+    assert adapter.sent == []
+
+
+def test_subscribed_board_still_delivers_through_the_gate(tmp_path, monkeypatch):
+    """Regression: the lock + probe must not change delivery for the owner."""
+    db_path = tmp_path / "subscribed.db"
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path))
+    kb.init_db()
+    tid = _create_completed_task(subscribe=True)
+
+    adapter = RecordingAdapter()
+    runner = _make_runner(adapter)
+    asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
+
+    assert len(adapter.sent) == 1
+    assert tid in adapter.sent[0]["text"]
+
+
+def test_lock_losing_instance_polls_zero_boards(tmp_path, monkeypatch):
+    """While another holder owns `.notifier.lock`, the watcher must return at
+    the gate: zero board enumerations, zero probes, zero DB opens, zero
+    deliveries."""
+    db_path = tmp_path / "locked-out.db"
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path))
+    kb.init_db()
+    _create_completed_task(subscribe=True)
+
+    lock_path = kb.kanban_home() / "kanban" / ".notifier.lock"
+    handle, state = _acquire_singleton_lock(lock_path)
+    assert state == "held", "test precondition: we hold the notifier lock"
+    try:
+        adapter = RecordingAdapter()
+        runner = _make_runner(adapter)
+        with patch.object(kb, "connect", wraps=kb.connect) as spy_connect, \
+                patch.object(kb, "list_boards", wraps=kb.list_boards) as spy_boards, \
+                patch.object(kb, "count_notify_subs", wraps=kb.count_notify_subs) as spy_probe:
+            asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
+    finally:
+        _release_singleton_lock(handle)
+
+    spy_boards.assert_not_called()
+    spy_probe.assert_not_called()
+    spy_connect.assert_not_called()
+    assert adapter.sent == []
+
+
+def test_two_instances_exactly_one_polls(tmp_path, monkeypatch):
+    """Two concurrent notifier instances against one kanban root: the first
+    acquires `.notifier.lock` and delivers; the second returns at the lock —
+    no second poller, no double delivery."""
+    db_path = tmp_path / "two-instances.db"
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path))
+    kb.init_db()
+    _create_completed_task(subscribe=True)
+
+    adapter1 = RecordingAdapter()
+    adapter2 = RecordingAdapter()
+    runner1 = _make_runner(adapter1)
+    runner2 = _make_runner(adapter2)
+
+    real_sleep = asyncio.sleep
+
+    async def fake_sleep(delay):
+        if delay == 5:
+            return None
+        runner1._running = False
+        runner2._running = False
+        await real_sleep(0)
+
+    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
+
+    async def run_both():
+        await asyncio.gather(
+            runner1._kanban_notifier_watcher(interval=1),
+            runner2._kanban_notifier_watcher(interval=1),
+        )
+
+    asyncio.run(run_both())
+
+    deliveries = adapter1.sent + adapter2.sent
+    assert len(deliveries) == 1, f"exactly one delivery expected, got {deliveries!r}"
+    # gather() starts runner1 first, so it wins the lock deterministically;
+    # runner2 must be the locked-out instance.
+    assert adapter2.sent == []
+
+
+def test_notifier_lock_released_on_return(tmp_path, monkeypatch):
+    """A finished watcher must release `.notifier.lock` so a successor in the
+    same process (gateway restart-in-place, next test) can acquire it."""
+    db_path = tmp_path / "release.db"
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path))
+    kb.init_db()
+    _create_completed_task(subscribe=True)
+
+    adapter = RecordingAdapter()
+    asyncio.run(_run_one_notifier_tick(monkeypatch, _make_runner(adapter)))
+    assert len(adapter.sent) == 1
+
+    lock_path = kb.kanban_home() / "kanban" / ".notifier.lock"
+    handle, state = _acquire_singleton_lock(lock_path)
+    try:
+        assert state == "held", "watcher exit must release the notifier lock"
+    finally:
+        _release_singleton_lock(handle)