diff --git a/docs/kanban/multi-gateway.md b/docs/kanban/multi-gateway.md index 126e39823a..fb6eec981f 100644 --- a/docs/kanban/multi-gateway.md +++ b/docs/kanban/multi-gateway.md @@ -10,11 +10,11 @@ Only one gateway owns the kanban dispatcher. The owning gateway keeps `kanban.dispatch_in_gateway: true` (the default); every other gateway sets it to `false`. -**Why this matters:** a gateway with `dispatch_in_gateway: true` opens per-board -SQLite connections for both the dispatcher and the notifier watcher. Multiple -gateways doing this concurrently multiplies the open file descriptors on each -`kanban.db` and amplifies WAL `-shm` reader contention. Gating both paths on the -same flag means exactly one process touches the kanban DBs. +**Why this matters:** dispatching is single-owner so multiple gateways do not +race to spawn the same work. Notification delivery is profile-owned instead: +each gateway polls only subscriptions for profiles whose platform adapters it +hosts. The atomic event claim prevents duplicate delivery across watcher +processes. ## Configuration @@ -30,10 +30,11 @@ Or set the env var: `HERMES_KANBAN_DISPATCH_IN_GATEWAY=false` ## What each gateway does -| Gateway role | dispatch_in_gateway | Opens per-board DBs? | Runs dispatcher + notifier? | -|---|---|---|---| -| default (dispatch owner) | true (default) | yes | yes | -| writer, admin, coder, etc. | false | no | no | +| Gateway role | dispatch_in_gateway | Opens subscribed board DBs? | Dispatcher | Notifier | +|---|---|---|---|---| +| default (confirmed dispatch-lock owner) | true (default) | yes | yes | owned profiles + legacy unstamped subscriptions | +| writer, admin, coder, etc. | false | yes, when the profile has subscriptions | no | that gateway's owned profiles | Non-dispatch gateways still deliver messages for their own platform adapters -(Telegram, Discord, etc.) — they just don't poll kanban boards. +(Telegram, Discord, etc.). They do not dispatch tasks, and they skip boards +that have no subscriptions owned by their profiles. diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 2f5137a8a8..5fbbdea07b 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -112,6 +112,16 @@ def _release_singleton_lock(handle) -> None: class GatewayKanbanWatchersMixin: """Kanban watcher / notifier / dispatcher loops for GatewayRunner.""" + def _owns_kanban_dispatcher_lock(self) -> bool: + """Return whether this gateway currently owns the singleton lock.""" + return getattr(self, "_kanban_dispatcher_lock_handle", None) is not None + + def _release_kanban_dispatcher_lock(self) -> None: + """Clear notifier-visible ownership before releasing the OS lock.""" + handle = getattr(self, "_kanban_dispatcher_lock_handle", None) + self._kanban_dispatcher_lock_handle = None + _release_singleton_lock(handle) + async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None: """Poll ``kanban_notify_subs`` and deliver terminal events to users. @@ -127,34 +137,15 @@ class GatewayKanbanWatchersMixin: WAL lock. Failures in one tick don't stop subsequent ticks. **Multi-board:** iterates every board discovered on disk per - tick. Subscriptions live inside each board's own DB and cannot - cross boards, so delivery semantics are unchanged — this is - purely a fan-out of the single-DB poll. + tick. Each gateway polls only subscriptions owned by profiles whose + adapters it hosts. The dispatch-owning gateway also handles legacy + subscriptions without a profile stamp. """ - # 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. - try: - from hermes_cli.config import load_config as _load_config - except Exception: - logger.warning("kanban notifier: config loader unavailable; disabled") - return - env_override = os.environ.get("HERMES_KANBAN_DISPATCH_IN_GATEWAY", "").strip().lower() - if env_override in {"0", "false", "no", "off"}: - logger.info("kanban notifier: disabled via HERMES_KANBAN_DISPATCH_IN_GATEWAY env") - return - try: - cfg = _load_config() - except Exception as exc: - logger.warning("kanban notifier: cannot load config (%s); disabled", exc) - return - kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {} - if not kanban_cfg.get("dispatch_in_gateway", True): - logger.info( - "kanban notifier: disabled via config kanban.dispatch_in_gateway=false" - ) - return + # Dispatch and delivery have separate ownership. A deployment may run + # one dispatcher while each profile has its own gateway credentials; + # those adapter-owning gateways must still poll and deliver their own + # subscriptions. Legacy rows without a notifier_profile are visible + # only while this process holds the actual singleton dispatcher lock. from gateway.config import Platform as _Platform try: from hermes_cli import kanban_db as _kb @@ -204,6 +195,13 @@ class GatewayKanbanWatchersMixin: try: def _collect(): deliveries: list[dict] = [] + include_unowned = self._owns_kanban_dispatcher_lock() + notifier_profiles = {notifier_profile} + notifier_profiles.update( + str(profile).strip() + for profile in getattr(self, "_profile_adapters", {}) + if str(profile).strip() + ) active_platforms = { getattr(platform, "value", str(platform)).lower() for platform in self.adapters.keys() @@ -262,10 +260,14 @@ class GatewayKanbanWatchersMixin: # checkpoint traffic) is exactly the per-tick cost # this skip avoids. try: - if _kb.count_notify_subs(board=slug) == 0: + if _kb.count_notify_subs( + board=slug, + notifier_profiles=notifier_profiles, + include_unowned=include_unowned, + ) == 0: logger.debug( - "kanban notifier: board %s has no subscriptions; skipping open", - slug, + "kanban notifier: board %s has no subscriptions owned by %s; skipping open", + slug, sorted(notifier_profiles), ) continue except Exception as exc: @@ -292,7 +294,11 @@ class GatewayKanbanWatchersMixin: # a legacy DB. `_add_column_if_missing` now # tolerates that race, but we still skip the # redundant call to avoid the wasted work. - subs = _kb.list_notify_subs(conn) + subs = _kb.list_notify_subs( + conn, + notifier_profiles=notifier_profiles, + include_unowned=include_unowned, + ) if not subs: logger.debug("kanban notifier: board %s has no subscriptions", slug) for sub in subs: @@ -1472,8 +1478,7 @@ class GatewayKanbanWatchersMixin: last_warn_at = now except asyncio.CancelledError: logger.debug("kanban dispatcher: cancelled") - _release_singleton_lock(self._kanban_dispatcher_lock_handle) - self._kanban_dispatcher_lock_handle = None + self._release_kanban_dispatcher_lock() raise except Exception: logger.exception("kanban dispatcher: unexpected watcher error") @@ -1485,5 +1490,4 @@ class GatewayKanbanWatchersMixin: await asyncio.sleep(min(1.0, interval - slept)) slept += 1.0 - _release_singleton_lock(self._kanban_dispatcher_lock_handle) - self._kanban_dispatcher_lock_handle = None + self._release_kanban_dispatcher_lock() diff --git a/gateway/run.py b/gateway/run.py index 7dde08044d..2cff27be45 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -11002,9 +11002,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Start background session expiry watcher to finalize expired sessions self._spawn_supervised(self._session_expiry_watcher, "session_expiry_watcher") - # Start background kanban notifier — delivers `completed`, `blocked`, - # `spawn_auto_blocked`, and `crashed` events to gateway subscribers - # so human-in-the-loop workflows hear back without polling. + # Start background kanban notifier — each gateway delivers events for + # subscriptions owned by the profiles whose adapters it hosts, even + # when another gateway owns the single dispatcher. self._spawn_supervised(self._kanban_notifier_watcher, "kanban_notifier_watcher") # Start background kanban dispatcher — spawns workers for ready diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 5b688a31aa..72fa92cf20 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -9598,15 +9598,65 @@ def add_notify_sub( ) +def _notify_profile_filter( + notifier_profiles: Optional[Iterable[str]], + *, + include_unowned: bool, +) -> tuple[str, list[str]]: + """Build an optional SQL predicate for notification profile ownership.""" + if notifier_profiles is None: + return "", [] + + profiles = sorted( + { + str(profile).strip() + for profile in notifier_profiles + if str(profile).strip() + } + ) + clauses: list[str] = [] + params: list[str] = [] + if profiles: + clauses.append( + "notifier_profile IN (" + ",".join("?" for _ in profiles) + ")" + ) + params.extend(profiles) + if include_unowned: + clauses.append("notifier_profile IS NULL OR notifier_profile = ''") + if not clauses: + return "0", [] + return "(" + ") OR (".join(clauses) + ")", params + + def list_notify_subs( - conn: sqlite3.Connection, task_id: Optional[str] = None, + conn: sqlite3.Connection, + task_id: Optional[str] = None, + *, + notifier_profiles: Optional[Iterable[str]] = None, + include_unowned: bool = False, ) -> list[dict]: + """List subscriptions, optionally restricted to notifier profile owners. + + Passing no ``notifier_profiles`` preserves the historical all-subscriptions + result. Gateway notifier processes pass the profiles whose adapters they + own so they cannot claim another gateway's events. ``include_unowned`` is + used by the dispatch owner for legacy rows created before profile stamping. + """ + owner_where, owner_params = _notify_profile_filter( + notifier_profiles, include_unowned=include_unowned, + ) + where: list[str] = [] + params: list[Any] = [] if task_id is not None: - rows = conn.execute( - "SELECT * FROM kanban_notify_subs WHERE task_id = ?", (task_id,), - ).fetchall() - else: - rows = conn.execute("SELECT * FROM kanban_notify_subs").fetchall() + where.append("task_id = ?") + params.append(task_id) + if owner_where: + where.append(owner_where) + params.extend(owner_params) + sql = "SELECT * FROM kanban_notify_subs" + if where: + sql += " WHERE " + " AND ".join(f"({clause})" for clause in where) + rows = conn.execute(sql, params).fetchall() out: list[dict] = [] for row in rows: item = dict(row) @@ -9622,6 +9672,8 @@ def count_notify_subs( db_path: Optional[Path] = None, *, board: Optional[str] = None, + notifier_profiles: Optional[Iterable[str]] = None, + include_unowned: bool = False, ) -> int: """Count ``kanban_notify_subs`` rows via a read-only connection. @@ -9633,8 +9685,10 @@ def count_notify_subs( 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 + zero. When ``notifier_profiles`` is supplied, only subscriptions owned + by those profiles are counted; ``include_unowned`` also includes legacy + rows without an owner stamp. 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. """ @@ -9644,9 +9698,13 @@ def count_notify_subs( conn = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True) try: try: - row = conn.execute( - "SELECT COUNT(*) FROM kanban_notify_subs" - ).fetchone() + owner_where, owner_params = _notify_profile_filter( + notifier_profiles, include_unowned=include_unowned, + ) + sql = "SELECT COUNT(*) FROM kanban_notify_subs" + if owner_where: + sql += " WHERE " + owner_where + row = conn.execute(sql, owner_params).fetchone() except sqlite3.OperationalError as exc: if "no such table" in str(exc).lower(): return 0 diff --git a/tests/gateway/test_kanban_notifier.py b/tests/gateway/test_kanban_notifier.py index 32a729f483..7400efd726 100644 --- a/tests/gateway/test_kanban_notifier.py +++ b/tests/gateway/test_kanban_notifier.py @@ -4,6 +4,10 @@ from pathlib import Path 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 @@ -45,6 +49,9 @@ def _make_runner(adapter): runner._running = True runner.adapters = {Platform.TELEGRAM: adapter} runner._kanban_sub_fail_counts = {} + # Most tests model the default gateway after its dispatcher acquired the + # singleton lock. Tests for startup or non-owner gateways clear this. + runner._kanban_dispatcher_lock_handle = object() return runner @@ -160,6 +167,105 @@ def test_active_named_profile_subscription_is_delivered(tmp_path, monkeypatch): assert "blocked" in message +def test_non_dispatch_gateway_claims_only_its_profile_subscriptions( + tmp_path, monkeypatch, +): + """A profile gateway delivers its events while another gateway dispatches.""" + db_path = tmp_path / "cross-profile-notifier.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + conn = kb.connect() + try: + foreign_tid = kb.create_task( + conn, title="default-owned", assignee="worker", + ) + kb.add_notify_sub( + conn, + task_id=foreign_tid, + platform="telegram", + chat_id="default-chat", + notifier_profile="default", + ) + kb.complete_task(conn, foreign_tid, summary="default done") + + owned_tid = kb.create_task( + conn, title="writer-owned", assignee="worker", + ) + kb.add_notify_sub( + conn, + task_id=owned_tid, + platform="telegram", + chat_id="writer-chat", + notifier_profile="writer", + ) + kb.complete_task(conn, owned_tid, summary="writer done") + finally: + conn.close() + + adapter = RecordingAdapter() + runner = _make_runner(adapter) + runner._active_profile_name = lambda: "writer" + runner._kanban_dispatcher_lock_handle = None + + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + assert [delivery["chat_id"] for delivery in adapter.sent] == ["writer-chat"] + assert owned_tid in adapter.sent[0]["text"] + assert len(_unseen_terminal_events_for(foreign_tid, "default-chat")) == 1 + + +def test_legacy_subscription_requires_confirmed_dispatcher_lock_owner( + tmp_path, monkeypatch, +): + """Startup and lock-losing gateways cannot claim legacy notifications.""" + db_path = tmp_path / "legacy-lock-owner.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + conn = kb.connect() + try: + task_id = kb.create_task(conn, title="legacy", assignee="worker") + kb.add_notify_sub( + conn, + task_id=task_id, + platform="telegram", + chat_id="legacy-chat", + ) + kb.complete_task(conn, task_id, summary="legacy done") + finally: + conn.close() + + startup_adapter = RecordingAdapter() + startup_runner = _make_runner(startup_adapter) + startup_runner._kanban_dispatcher_lock_handle = None + asyncio.run(_run_one_notifier_tick(monkeypatch, startup_runner)) + assert startup_adapter.sent == [] + assert len(_unseen_terminal_events_for(task_id, "legacy-chat")) == 1 + + lock_path = tmp_path / ".dispatcher.lock" + winner_handle, winner_state = _acquire_singleton_lock(lock_path) + loser_handle, loser_state = _acquire_singleton_lock(lock_path) + try: + assert winner_state == "held" + assert loser_state == "contended" + + loser_adapter = RecordingAdapter() + loser_runner = _make_runner(loser_adapter) + loser_runner._kanban_dispatcher_lock_handle = loser_handle + asyncio.run(_run_one_notifier_tick(monkeypatch, loser_runner)) + assert loser_adapter.sent == [] + assert len(_unseen_terminal_events_for(task_id, "legacy-chat")) == 1 + + winner_adapter = RecordingAdapter() + winner_runner = _make_runner(winner_adapter) + winner_runner._kanban_dispatcher_lock_handle = winner_handle + asyncio.run(_run_one_notifier_tick(monkeypatch, winner_runner)) + assert [item["chat_id"] for item in winner_adapter.sent] == ["legacy-chat"] + assert task_id in winner_adapter.sent[0]["text"] + finally: + _release_singleton_lock(loser_handle) + _release_singleton_lock(winner_handle) + + class FailingAdapter: """Adapter whose send() always raises, simulating a transient send error.""" @@ -347,8 +453,8 @@ def test_kanban_notifier_isolates_per_subscription_failure(tmp_path, monkeypatch # unordered SELECT's scan order. original_list = kb.list_notify_subs - def bad_first(conn, task_id=None): - subs = original_list(conn, task_id) + def bad_first(conn, task_id=None, **kwargs): + subs = original_list(conn, task_id, **kwargs) return sorted(subs, key=lambda s: 0 if s["task_id"] == tid_bad else 1) monkeypatch.setattr(kb, "list_notify_subs", bad_first) diff --git a/tests/gateway/test_kanban_notifier_apiserver_wake.py b/tests/gateway/test_kanban_notifier_apiserver_wake.py index 93dbf0e0da..5374ecf44a 100644 --- a/tests/gateway/test_kanban_notifier_apiserver_wake.py +++ b/tests/gateway/test_kanban_notifier_apiserver_wake.py @@ -69,6 +69,7 @@ def _make_runner(adapters): runner._running = True runner.adapters = adapters runner._kanban_sub_fail_counts = {} + runner._kanban_dispatcher_lock_handle = object() return runner @@ -135,4 +136,3 @@ def test_apiserver_sub_wakes_real_session_via_self_post(tmp_path, monkeypatch): # once the wake succeeds. assert _unseen_terminal_events(tid, "api_server", "raw-sid-123") == [] - diff --git a/tests/gateway/test_kanban_notifier_watcher_dispatch_gate.py b/tests/gateway/test_kanban_notifier_watcher_dispatch_gate.py index 71c37c9ce4..b3276c9682 100644 --- a/tests/gateway/test_kanban_notifier_watcher_dispatch_gate.py +++ b/tests/gateway/test_kanban_notifier_watcher_dispatch_gate.py @@ -1,9 +1,4 @@ -"""Tests for the dispatch_in_gateway gate on _kanban_notifier_watcher. - -- Non-dispatch gateways (dispatch_in_gateway=false) exit before opening any DB. -- HERMES_KANBAN_DISPATCH_IN_GATEWAY env var disables without loading config. -- Dispatch-owning gateways (dispatch_in_gateway=true) proceed past the gate. -""" +"""Notifier polling stays active when another gateway owns dispatching.""" import asyncio from unittest.mock import MagicMock, patch @@ -20,12 +15,8 @@ def _make_runner(with_adapter=False): return runner -def _fake_config(dispatch_in_gateway): - return {"kanban": {"dispatch_in_gateway": dispatch_in_gateway}} - - -def test_notifier_watcher_runs_when_dispatch_enabled(): - """dispatch_in_gateway=true proceeds past the gate to the board fan-out.""" +def test_notifier_watcher_polls_without_dispatch_ownership(): + """A profile gateway still polls its profile-owned subscriptions.""" runner = _make_runner(with_adapter=True) past_gate = [] sleep_calls = [] @@ -42,13 +33,14 @@ def test_notifier_watcher_runs_when_dispatch_enabled(): import hermes_cli.kanban_db as _kb - with patch("hermes_cli.config.load_config", return_value=_fake_config(True)): - with patch.object( - _kb, "list_boards", - side_effect=lambda *a, **kw: past_gate.append(True) or [], - ): - with patch("asyncio.sleep", side_effect=fake_sleep): - with patch("asyncio.to_thread", side_effect=fake_to_thread): - asyncio.run(runner._kanban_notifier_watcher()) + with patch.object( + _kb, "list_boards", + side_effect=lambda *a, **kw: past_gate.append(True) or [], + ): + with patch("asyncio.sleep", side_effect=fake_sleep): + with patch("asyncio.to_thread", side_effect=fake_to_thread): + asyncio.run(runner._kanban_notifier_watcher()) - assert past_gate, "list_boards should be called when dispatch_in_gateway=true" + assert past_gate, ( + "gateways without the dispatch lock must still poll owned subscriptions" + ) diff --git a/tests/hermes_cli/test_kanban_count_notify_subs.py b/tests/hermes_cli/test_kanban_count_notify_subs.py index 5a83812cc0..c8c419133e 100644 --- a/tests/hermes_cli/test_kanban_count_notify_subs.py +++ b/tests/hermes_cli/test_kanban_count_notify_subs.py @@ -63,3 +63,29 @@ def test_legacy_db_without_subs_table_counts_zero_and_stays_unmigrated(tmp_path) ) +def test_count_notify_subs_filters_profile_owners(tmp_path): + db_path = tmp_path / "owners.db" + kb.init_db(db_path) + conn = kb.connect(db_path) + try: + for profile in ("default", "writer", None): + task_id = kb.create_task(conn, title=f"owned by {profile}") + kb.add_notify_sub( + conn, + task_id=task_id, + platform="telegram", + chat_id=f"chat-{profile}", + notifier_profile=profile, + ) + finally: + conn.close() + + assert kb.count_notify_subs( + db_path=db_path, notifier_profiles={"writer"}, + ) == 1 + assert kb.count_notify_subs( + db_path=db_path, + notifier_profiles={"default"}, + include_unowned=True, + ) == 2 + diff --git a/tests/hermes_cli/test_kanban_notify.py b/tests/hermes_cli/test_kanban_notify.py index 0e52295ad5..69ff262b4f 100644 --- a/tests/hermes_cli/test_kanban_notify.py +++ b/tests/hermes_cli/test_kanban_notify.py @@ -160,6 +160,7 @@ async def test_notifier_artifact_delivery_skips_missing_files(kanban_home, tmp_p runner = object.__new__(GatewayRunner) runner._running = True runner._kanban_sub_fail_counts = {} + runner._kanban_dispatcher_lock_handle = object() fake_adapter = MagicMock() fake_adapter.name = "telegram"