diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 5bcf70c8d218..ecb2fae40ad7 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -55,6 +55,56 @@ def _resolve_auto_decompose_settings( return enabled, per_tick +def _resolve_default_notify_targets( + load_config: Callable[[], Any], +) -> "list[dict]": + """Resolve the live board-wide ``kanban.default_notify`` target list. + + Each entry fans every ticket's terminal events out to a chat with zero + manual ``notify-subscribe``. Shape in ``config.yaml``:: + + kanban: + default_notify: + - platform: signal + chat_id: "group:..." + thread_id: "" # optional + + Read fresh on every notifier tick (like ``_resolve_auto_decompose_settings``) + so adding/removing a target takes effect on the next tick instead of + requiring a gateway restart. Fails **safe**: any read/parse error returns + ``[]`` (no auto-subscribe) rather than raising out of the notifier loop. + Entries missing ``platform`` or ``chat_id`` are dropped. ``thread_id`` + defaults to ``""`` to match the subscription primary key. + """ + try: + cfg = load_config() + except Exception: + return [] + kcfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {} + if not isinstance(kcfg, dict): + # A malformed live edit (e.g. ``kanban: false`` or a YAML list) must + # not raise out of the notifier loop — that would skip ALL per-task + # delivery every tick. Fail safe to "no default targets". + return [] + raw = kcfg.get("default_notify") or [] + if not isinstance(raw, (list, tuple)): + return [] + targets: list[dict] = [] + for entry in raw: + if not isinstance(entry, dict): + continue + platform = str(entry.get("platform") or "").strip().lower() + chat_id = str(entry.get("chat_id") or "").strip() + if not platform or not chat_id: + continue + targets.append({ + "platform": platform, + "chat_id": chat_id, + "thread_id": str(entry.get("thread_id") or "").strip(), + }) + return targets + + def _acquire_singleton_lock(lock_path) -> "tuple[Optional[object], str]": """Take an exclusive, non-blocking advisory lock for the sole dispatcher. @@ -192,6 +242,11 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None: while self._running: try: + # Resolve board-wide auto-subscribe targets fresh each tick so + # adding/removing a `kanban.default_notify` entry takes effect + # on the next tick (no gateway restart). Fails safe to []. + default_notify_targets = _resolve_default_notify_targets(_load_config) + def _collect(): deliveries: list[dict] = [] active_platforms = { @@ -244,6 +299,45 @@ def _collect(): # a legacy DB. `_add_column_if_missing` now # tolerates that race, but we still skip the # redundant call to avoid the wasted work. + # + # Board-wide auto-subscribe: before reading the + # subscription table, ensure every config-listed + # `kanban.default_notify` target is subscribed to + # every active (non-final) task on this board. This + # is what makes terminal events fan out to e.g. the + # "Kanban Workers" Signal group with zero manual + # `notify-subscribe`. `add_notify_sub` is + # INSERT-OR-IGNORE on the (task, platform, chat, + # thread) PK, so this is idempotent and never + # disturbs an existing per-task subscription (the + # cursor / fail-count state is keyed the same way). + # We only auto-subscribe targets whose platform + # adapter is currently connected, mirroring the + # delivery gate below. + if default_notify_targets: + # One bulk INSERT-OR-IGNORE per connected target + # instead of a per-task write txn each tick: the + # SELECT picks every active (non-final) task in a + # single transaction. Idempotent on the PK, so an + # existing per-task subscription (cursor / + # fail-count) is never disturbed. + for tgt in default_notify_targets: + if tgt["platform"] not in active_platforms: + continue + try: + _kb.add_default_notify_subs( + conn, + platform=tgt["platform"], + chat_id=tgt["chat_id"], + thread_id=tgt["thread_id"], + notifier_profile=notifier_profile, + ) + except Exception as exc: + logger.debug( + "kanban notifier: default-notify subscribe failed for %s on %s: %s", + tgt["chat_id"], tgt["platform"], exc, + ) + subs = _kb.list_notify_subs(conn) if not subs: logger.debug("kanban notifier: board %s has no subscriptions", slug) diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 2ba908dc49f1..e1e503e93852 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -2444,6 +2444,17 @@ def _ensure_hermes_home_managed(home: Path): # worker process (if still running host-locally) is terminated # before the reclaim. 0 disables stale detection entirely. "dispatch_stale_timeout_seconds": 14400, + # Board-wide auto-subscribe. Each entry fans EVERY ticket's terminal + # events (completed / blocked / gave_up / crashed / timed_out, plus + # completion artifacts) out to a chat with zero manual + # ``hermes kanban notify-subscribe``. The gateway notifier applies + # these to every active task each tick (idempotent INSERT-OR-IGNORE, + # so per-task subscriptions are never disturbed). Read fresh each + # tick — editing this list takes effect on the next tick, no restart. + # Entries are dicts: {platform, chat_id, thread_id?}. ``chat_id`` is + # the adapter's send-format id (e.g. Signal group: "group:"). + # Empty list = feature off (default). + "default_notify": [], }, # execute_code settings — controls the tool used for programmatic tool calls. diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index c3107e37d757..d2c545c23274 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -7878,6 +7878,41 @@ def list_notify_subs( return [dict(r) for r in rows] +def add_default_notify_subs( + conn: sqlite3.Connection, + *, + platform: str, + chat_id: str, + thread_id: Optional[str] = None, + notifier_profile: Optional[str] = None, + final_statuses: Iterable[str] = ("done", "archived"), +) -> None: + """Subscribe one board-wide target to every active (non-final) task in a + single write transaction. + + Equivalent to calling :func:`add_notify_sub` once per active task, but + issues one ``INSERT OR IGNORE ... SELECT`` instead of N IMMEDIATE write + transactions per notifier tick — the per-task loop opened a write txn for + every active task every 5s even when the rows already existed. Idempotent + on the (task, platform, chat, thread) PK, so existing per-task or default + subscriptions (and their cursors) are never disturbed. + """ + finals = tuple(final_statuses) + placeholders = ",".join("?" for _ in finals) + now = int(time.time()) + with write_txn(conn): + conn.execute( + f""" + INSERT OR IGNORE INTO kanban_notify_subs + (task_id, platform, chat_id, thread_id, notifier_profile, created_at) + SELECT id, ?, ?, ?, ?, ? + FROM tasks + WHERE status NOT IN ({placeholders}) + """, + (platform, chat_id, thread_id or "", notifier_profile, now, *finals), + ) + + def remove_notify_sub( conn: sqlite3.Connection, *, diff --git a/tests/gateway/test_kanban_default_notify_live.py b/tests/gateway/test_kanban_default_notify_live.py new file mode 100644 index 000000000000..8712a0f4c734 --- /dev/null +++ b/tests/gateway/test_kanban_default_notify_live.py @@ -0,0 +1,318 @@ +"""Tests for live board-wide auto-subscribe target resolution. + +``kanban.default_notify`` fans every ticket's terminal events out to the +listed chats (e.g. a Signal group) with zero manual ``notify-subscribe``. +``_resolve_default_notify_targets`` is called every notifier tick (like +``_resolve_auto_decompose_settings``) so editing the list takes effect on the +next tick without a gateway restart, and it must fail SAFE (empty list = no +auto-subscribe) on any malformed/erroring config so the notifier loop never +dies. + +The second half is an INTEGRATION test that drives the real notifier mixin +(``GatewayRunner._kanban_notifier_watcher``) against a temp board with +``kanban.default_notify`` set and NO manual subscription, proving a completed +ticket is delivered to the configured target end-to-end through the real +``_collect`` default-subscribe block + the real delivery path. +""" + +from __future__ import annotations + +import asyncio + +import gateway.kanban_watchers as kw +from gateway.config import Platform +from gateway.kanban_watchers import _resolve_default_notify_targets +from gateway.run import GatewayRunner +from hermes_cli import kanban_db as kb + + +def test_empty_when_key_absent(): + assert _resolve_default_notify_targets(lambda: {"kanban": {}}) == [] + + +def test_empty_when_default_notify_empty(): + assert _resolve_default_notify_targets( + lambda: {"kanban": {"default_notify": []}} + ) == [] + + +def test_single_target_normalized(): + targets = _resolve_default_notify_targets( + lambda: { + "kanban": { + "default_notify": [ + {"platform": "Signal", "chat_id": "group:abc="}, + ] + } + } + ) + assert targets == [ + {"platform": "signal", "chat_id": "group:abc=", "thread_id": ""} + ] + + +def test_thread_id_preserved(): + targets = _resolve_default_notify_targets( + lambda: { + "kanban": { + "default_notify": [ + {"platform": "telegram", "chat_id": "123", "thread_id": "7"}, + ] + } + } + ) + assert targets == [ + {"platform": "telegram", "chat_id": "123", "thread_id": "7"} + ] + + +def test_entries_missing_platform_or_chat_id_dropped(): + targets = _resolve_default_notify_targets( + lambda: { + "kanban": { + "default_notify": [ + {"platform": "signal"}, # no chat_id + {"chat_id": "group:x="}, # no platform + {"platform": "", "chat_id": "y="}, # blank platform + {"platform": "signal", "chat_id": "group:ok="}, # valid + ] + } + } + ) + assert targets == [ + {"platform": "signal", "chat_id": "group:ok=", "thread_id": ""} + ] + + +def test_non_dict_entries_skipped(): + targets = _resolve_default_notify_targets( + lambda: { + "kanban": { + "default_notify": [ + "not-a-dict", + None, + {"platform": "signal", "chat_id": "group:ok="}, + ] + } + } + ) + assert targets == [ + {"platform": "signal", "chat_id": "group:ok=", "thread_id": ""} + ] + + +def test_non_list_default_notify_fails_safe(): + assert _resolve_default_notify_targets( + lambda: {"kanban": {"default_notify": "group:x="}} + ) == [] + + +def test_non_dict_config_fails_safe(): + assert _resolve_default_notify_targets(lambda: None) == [] + assert _resolve_default_notify_targets(lambda: ["not", "a", "dict"]) == [] + + +def test_malformed_kanban_section_fails_safe(): + # A live edit that leaves ``kanban`` as a non-dict (e.g. ``kanban: false`` + # or a YAML list) must not raise — it would break ALL per-task delivery. + assert _resolve_default_notify_targets(lambda: {"kanban": False}) == [] + assert _resolve_default_notify_targets(lambda: {"kanban": ["a", "b"]}) == [] + assert _resolve_default_notify_targets(lambda: {"kanban": "nope"}) == [] + + +def test_config_read_error_fails_safe_empty(): + def _boom(): + raise RuntimeError("config read failed") + + assert _resolve_default_notify_targets(_boom) == [] + + +def test_live_edit_takes_effect_between_calls(): + state = {"kanban": {"default_notify": []}} + assert _resolve_default_notify_targets(lambda: state) == [] + state["kanban"]["default_notify"] = [ + {"platform": "signal", "chat_id": "group:new="} + ] + assert _resolve_default_notify_targets(lambda: state) == [ + {"platform": "signal", "chat_id": "group:new=", "thread_id": ""} + ] + + +# -------------------------------------------------------------------------- +# Integration: real notifier mixin, real board, default_notify drives delivery +# with NO manual notify-subscribe. +# -------------------------------------------------------------------------- + + +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 {}}) + + +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 _make_runner(adapter, platform=Platform.SIGNAL): + runner = GatewayRunner.__new__(GatewayRunner) + runner._running = True + runner.adapters = {platform: adapter} + runner._kanban_sub_fail_counts = {} + return runner + + +def test_default_notify_delivers_without_manual_subscribe(tmp_path, monkeypatch): + """A ticket fans out to the config-listed chat with zero manual + ``notify-subscribe``. Drives the real ``_collect`` default-subscribe block + + the real delivery path across the realistic two-tick lifecycle: + tick #1 subscribes the active (``ready``) task; the task then completes; + tick #2 delivers the completion to the auto-added target.""" + db_path = tmp_path / "default-notify.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + + target_chat = "group:y94kMF95wnSq3UYCBM6Ihei1ViUakcoGlmZkcTafwjk=" + + fake_cfg = { + "kanban": { + "dispatch_in_gateway": True, + "default_notify": [ + {"platform": "signal", "chat_id": target_chat}, + ], + } + } + monkeypatch.setattr( + "hermes_cli.config.load_config", lambda *a, **k: fake_cfg, raising=False + ) + + # Create an active task (status `ready`) — NO manual notify subscription. + conn = kb.connect() + try: + tid = kb.create_task(conn, title="auto fanout", assignee="researcher") + assert kb.list_notify_subs(conn, tid) == [] + finally: + conn.close() + + adapter = RecordingAdapter() + + # Tick #1: the default-notify block subscribes the active task. Nothing to + # deliver yet (no terminal event). + runner = _make_runner(adapter) + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + assert adapter.sent == [] + conn = kb.connect() + try: + subs = kb.list_notify_subs(conn, tid) + assert len(subs) == 1 and subs[0]["chat_id"] == target_chat, ( + "active task must be auto-subscribed to the default-notify target" + ) + # Worker finishes. + kb.complete_task(conn, tid, summary="echo ok done") + finally: + conn.close() + + # Tick #2: the completed event is delivered to the auto-added target. + runner = _make_runner(adapter) + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + assert len(adapter.sent) == 1, ( + f"expected one auto-fanout delivery, got {adapter.sent}" + ) + assert adapter.sent[0]["chat_id"] == target_chat + assert tid in adapter.sent[0]["text"] + assert "done" in adapter.sent[0]["text"].lower() + + +def test_default_notify_does_not_disturb_existing_per_task_sub(tmp_path, monkeypatch): + """The per-task subscribe path is untouched: a pre-existing subscription to + a different chat keeps delivering, and the default target is added + alongside it (no clobber, no regression).""" + db_path = tmp_path / "coexist.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + + default_chat = "group:y94kMF95wnSq3UYCBM6Ihei1ViUakcoGlmZkcTafwjk=" + manual_chat = "group:manual-existing=" + + fake_cfg = { + "kanban": { + "dispatch_in_gateway": True, + "default_notify": [{"platform": "signal", "chat_id": default_chat}], + } + } + monkeypatch.setattr( + "hermes_cli.config.load_config", lambda *a, **k: fake_cfg, raising=False + ) + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="coexist", assignee="researcher") + # Pre-existing manual subscription to a DIFFERENT chat. + kb.add_notify_sub(conn, task_id=tid, platform="signal", chat_id=manual_chat) + finally: + conn.close() + + adapter = RecordingAdapter() + + # Tick #1: default-notify adds its target alongside the manual sub while + # the task is still active. + runner = _make_runner(adapter) + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + conn = kb.connect() + try: + chats = sorted(s["chat_id"] for s in kb.list_notify_subs(conn, tid)) + assert chats == sorted([default_chat, manual_chat]), ( + f"both subs should coexist after tick #1; got {chats}" + ) + kb.complete_task(conn, tid, summary="ok") + finally: + conn.close() + + # Tick #2: both targets receive the completion. + runner = _make_runner(adapter) + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + chats = sorted(d["chat_id"] for d in adapter.sent) + assert chats == sorted([default_chat, manual_chat]), ( + f"both the manual and default targets should receive the completion; got {chats}" + ) + + +def test_add_default_notify_subs_bulk_active_only(tmp_path, monkeypatch): + """The bulk helper subscribes every active task and skips final ones in a + single transaction, and is idempotent on re-run (no duplicate rows).""" + db_path = tmp_path / "bulk.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + + chat = "group:bulk-target=" + conn = kb.connect() + try: + active1 = kb.create_task(conn, title="a1", assignee="researcher") + active2 = kb.create_task(conn, title="a2", assignee="researcher") + done = kb.create_task(conn, title="d", assignee="researcher") + kb.complete_task(conn, done, summary="ok") + + kb.add_default_notify_subs(conn, platform="signal", chat_id=chat) + subscribed = {s["task_id"] for s in kb.list_notify_subs(conn)} + assert subscribed == {active1, active2}, ( + f"only active tasks should be subscribed; got {subscribed}" + ) + + # Idempotent: a second call adds no duplicate rows. + kb.add_default_notify_subs(conn, platform="signal", chat_id=chat) + assert len(kb.list_notify_subs(conn)) == 2 + finally: + conn.close()