diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 2563091f4fc4..14d2e43a6644 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -236,6 +236,26 @@ class GatewayKanbanWatchersMixin: ) continue seen_db_paths.add(resolved_db_path) + # Zero-subscription early exit: probe the board with a + # cheap read-only connection BEFORE the writable + # `connect()`. A board with no subscriptions has + # nothing to notify, and the writable open (schema + # init/migration on first open, WAL/-shm sidecars, + # checkpoint traffic) is exactly the per-tick cost + # this skip avoids. + 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: diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 3877a7f10b42..bd3c428cf002 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -9511,6 +9511,44 @@ def list_notify_subs( return out +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, *, diff --git a/tests/gateway/test_kanban_notifier_zero_sub_gate.py b/tests/gateway/test_kanban_notifier_zero_sub_gate.py new file mode 100644 index 000000000000..d8e64f2de96e --- /dev/null +++ b/tests/gateway/test_kanban_notifier_zero_sub_gate.py @@ -0,0 +1,120 @@ +"""Tests for the kanban notifier zero-subscription early exit. + +The notifier used to writable-open EVERY board DB on every tick even when a +board had zero subscriptions — paying schema init/migration on first open, +WAL/-shm sidecar creation, and checkpoint traffic for boards with nothing to +notify. Per-board work is now gated by a read-only subscription probe +(``kanban_db.count_notify_subs``), so boards with zero subscriptions are +never opened writable. + +(The companion machine-global ``.notifier.lock`` singleton gate from PR +#63001 was deliberately NOT salvaged: a lock-winning default-profile gateway +cannot deliver a secondary profile's subscriptions in standalone-profile +deployments — profile routing fails closed in +``gateway/authz_mixin.py::_authorization_adapter`` — so the lock could +suppress delivery entirely. The read-only probe captures the per-tick cost +win without that risk.) +""" + +import asyncio + +from unittest.mock import patch + +from gateway.config import Platform +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 zero-sub probe must not change delivery for live subs.""" + 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_probe_failure_falls_back_to_writable_open(tmp_path, monkeypatch): + """If the read-only probe raises (locked/corrupt DB), the notifier must + fall back to the writable open — a broken probe must never silently + disable notifications for a board with live subscriptions.""" + db_path = tmp_path / "probe-broken.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + tid = _create_completed_task(subscribe=True) + + def _boom(*args, **kwargs): + raise RuntimeError("probe exploded") + + monkeypatch.setattr(kb, "count_notify_subs", _boom) + + 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"] diff --git a/tests/hermes_cli/test_kanban_count_notify_subs.py b/tests/hermes_cli/test_kanban_count_notify_subs.py new file mode 100644 index 000000000000..f1361e29c5ee --- /dev/null +++ b/tests/hermes_cli/test_kanban_count_notify_subs.py @@ -0,0 +1,98 @@ +"""Tests for ``kanban_db.count_notify_subs`` — the read-only subscription probe. + +The gateway notifier uses it to skip boards with zero subscriptions BEFORE +any writable ``connect()``: the probe must never create the DB file, never +run schema init/migration, and never write — that first-open cost on every +tick is exactly what the zero-sub early exit avoids. It must also never +UNDER-count: rows sitting in a not-yet-checkpointed WAL still count, or the +notifier would skip a board that has a live subscription. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setenv("HERMES_KANBAN_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + return home + + +def test_missing_db_counts_zero_and_creates_nothing(kanban_home): + db_path = kb.kanban_db_path(board="default") + assert not db_path.exists() + assert kb.count_notify_subs(board="default") == 0 + assert not db_path.exists(), "read-only probe must not create the DB" + + +def test_counts_rows_via_board_resolution(kanban_home): + conn = kb.connect(board="default") + try: + tid = kb.create_task(conn, title="t", assignee="w") + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="c1") + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="c2") + finally: + conn.close() + assert kb.count_notify_subs(board="default") == 2 + + +def test_probe_is_read_only_and_sees_uncheckpointed_wal_rows(kanban_home): + """A sub committed by a still-open writer (rows only in the WAL, not yet + checkpointed into the main DB file) must be counted — under-counting + would make the notifier skip a board that has a live subscription. And + the probe itself must be read-only: the writer's connection stays the + only writer.""" + conn = kb.connect(board="default") + try: + tid = kb.create_task(conn, title="t", assignee="w") + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="c1") + # Writer still open: the row lives in the -wal, not the main file. + assert kb.count_notify_subs(board="default") == 1 + finally: + conn.close() + + +def test_legacy_db_without_subs_table_counts_zero_and_stays_unmigrated(tmp_path): + legacy = tmp_path / "legacy.db" + conn = sqlite3.connect(legacy) + try: + conn.execute("CREATE TABLE something_else (id INTEGER)") + conn.commit() + finally: + conn.close() + assert kb.count_notify_subs(db_path=legacy) == 0 + # The probe must not have run schema init on the foreign/legacy DB. + conn = sqlite3.connect(legacy) + try: + tables = { + r[0] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + } + finally: + conn.close() + assert "kanban_notify_subs" not in tables, ( + "read-only probe must never create schema" + ) + + +def test_explicit_db_path_overrides_board(kanban_home, tmp_path): + pinned = tmp_path / "pinned.db" + conn = kb.connect(db_path=pinned) + try: + tid = kb.create_task(conn, title="t", assignee="w") + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="c1") + finally: + conn.close() + assert kb.count_notify_subs(pinned) == 1 + assert kb.count_notify_subs(board="default") == 0