From a0bc7d5572cfee1cb8dc240c6292f835db58866c Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sun, 26 Jul 2026 13:36:41 -0700 Subject: [PATCH] fix(kanban): snap notify-sub cursor to current MAX(task_events.id) at creation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes the boot-storm half of issue #29905: kanban_notify_subs.last_event_id defaulted to 0, so a subscription created on an already-active task replayed the task's ENTIRE terminal-event backlog on the next notifier tick. With many stale subs (27 observed in the report) a gateway boot after downtime burst 100+ notifications in one go. add_notify_sub now snaps the cursor to COALESCE(MAX(task_events.id), 0) for the task inside the same INSERT, so new subscriptions start caught up and only receive events that occur AFTER subscribing. The gateway slash-command and kanban-tool auto-subscribe paths run at task creation, where the snapshot is just the 'created' event — behavior there is unchanged. Stale fixtures that asserted the literal 0 creation cursor now assert 'cursor unchanged/unclaimed' instead, which is what they actually meant. --- hermes_cli/kanban_db.py | 18 +++++-- .../test_kanban_core_functionality.py | 51 ++++++++++++++++++- tests/hermes_cli/test_kanban_notify.py | 7 ++- .../tui_gateway/test_kanban_notify_poller.py | 10 +++- 4 files changed, 79 insertions(+), 7 deletions(-) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index bd3c428cf002..355a7933a31d 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -9431,7 +9431,17 @@ def add_notify_sub( delivery_metadata: Optional[Mapping[str, Any]] = None, ) -> None: """Register a gateway source that wants terminal-state notifications - for ``task_id``. Idempotent on (task, platform, chat, thread).""" + for ``task_id``. Idempotent on (task, platform, chat, thread). + + New subscriptions start "caught up": ``last_event_id`` snaps to the + task's current ``MAX(task_events.id)`` at creation instead of the + schema default 0. A cursor of 0 on an already-active task made the + gateway notifier replay every historical terminal event on its next + tick — and with many stale subs, a single boot-time burst of 100+ + messages (issue #29905). Subscribers only want events that occur + AFTER they subscribe; the gateway/tool auto-subscribe paths run at + task creation, where the snapshot is 0 anyway. + """ now = int(time.time()) metadata_json = _encode_notify_delivery_metadata(delivery_metadata) with write_txn(conn): @@ -9439,8 +9449,9 @@ def add_notify_sub( """ INSERT OR IGNORE INTO kanban_notify_subs (task_id, platform, chat_id, chat_type, thread_id, user_id, - notifier_profile, delivery_metadata, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + notifier_profile, delivery_metadata, created_at, last_event_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, + COALESCE((SELECT MAX(id) FROM task_events WHERE task_id = ?), 0)) """, ( task_id, @@ -9452,6 +9463,7 @@ def add_notify_sub( notifier_profile, metadata_json, now, + task_id, ), ) if chat_type: diff --git a/tests/hermes_cli/test_kanban_core_functionality.py b/tests/hermes_cli/test_kanban_core_functionality.py index 5bb66e0d4eca..e75a90b28911 100644 --- a/tests/hermes_cli/test_kanban_core_functionality.py +++ b/tests/hermes_cli/test_kanban_core_functionality.py @@ -602,6 +602,9 @@ def test_notify_claim_is_single_owner_and_rewindable(kanban_home): try: tid = kb.create_task(conn1, title="x", assignee="w") kb.add_notify_sub(conn1, task_id=tid, platform="telegram", chat_id="123") + # New subs start caught up at the task's current MAX(task_events.id) + # (the `created` event) — issue #29905. + initial_cursor = int(kb.list_notify_subs(conn1, tid)[0]["last_event_id"]) kb.complete_task(conn1, tid, result="ok") old_cursor, claimed_cursor, events = kb.claim_unseen_events_for_sub( @@ -611,7 +614,7 @@ def test_notify_claim_is_single_owner_and_rewindable(kanban_home): chat_id="123", kinds=["completed", "blocked"], ) - assert old_cursor == 0 + assert old_cursor == initial_cursor assert claimed_cursor > old_cursor assert [ev.kind for ev in events] == ["completed"] @@ -4801,3 +4804,49 @@ def test_dispatch_once_stale_disabled_when_timeout_zero(kanban_home, monkeypatch ) assert res.stale == [], "stale_timeout_seconds=0 should disable detection" assert kb.get_task(conn, t).status == "running" + + +def test_notify_sub_starts_caught_up_on_active_task(kanban_home): + """A new subscription must NOT replay historical terminal events. + + Regression for issue #29905: `kanban_notify_subs.last_event_id` defaulted + to 0, so subscribing to a task that already had terminal events in + `task_events` replayed the entire backlog on the next notifier tick — 27 + stale subs produced a 100+ message burst at gateway boot. The cursor now + snaps to the task's MAX(task_events.id) at creation: only events that + occur AFTER subscribing are delivered. + """ + conn = kb.connect() + try: + tid = kb.create_task(conn, title="old task", assignee="w") + # Historical terminal activity BEFORE anyone subscribes. + kb.complete_task(conn, tid, result="done long ago") + + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="123") + sub = kb.list_notify_subs(conn, tid)[0] + assert int(sub["last_event_id"]) > 0, ( + "cursor must snap to MAX(task_events.id) at subscription time" + ) + _, events = kb.unseen_events_for_sub( + conn, task_id=tid, platform="telegram", chat_id="123", + kinds=["completed", "blocked", "gave_up", "crashed", "timed_out"], + ) + assert events == [], "historical events must not replay to a new sub" + finally: + conn.close() + + +def test_notify_sub_on_fresh_task_still_gets_future_events(kanban_home): + """The caught-up snap must not lose events that happen AFTER subscribing.""" + conn = kb.connect() + try: + tid = kb.create_task(conn, title="fresh", assignee="w") + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="123") + kb.complete_task(conn, tid, result="ok") + _, events = kb.unseen_events_for_sub( + conn, task_id=tid, platform="telegram", chat_id="123", + kinds=["completed"], + ) + assert [ev.kind for ev in events] == ["completed"] + finally: + conn.close() diff --git a/tests/hermes_cli/test_kanban_notify.py b/tests/hermes_cli/test_kanban_notify.py index 37acc5ae8ff0..63b327ceef24 100644 --- a/tests/hermes_cli/test_kanban_notify.py +++ b/tests/hermes_cli/test_kanban_notify.py @@ -451,6 +451,9 @@ async def test_notifier_skips_subscription_owned_by_other_profile(kanban_home): notifier_profile="default", ) kb.complete_task(conn, tid, result="done") + # New subs start caught up at the creation-time MAX(task_events.id) + # (issue #29905); claiming the completion would advance past this. + pre_claim_cursor = int(kb.list_notify_subs(conn, tid)[0]["last_event_id"]) finally: conn.close() @@ -486,7 +489,9 @@ async def test_notifier_skips_subscription_owned_by_other_profile(kanban_home): finally: conn.close() assert len(subs) == 1 - assert int(subs[0]["last_event_id"]) == 0, "wrong profile must not claim the event" + assert int(subs[0]["last_event_id"]) == pre_claim_cursor, ( + "wrong profile must not claim the event" + ) @pytest.mark.asyncio diff --git a/tests/tui_gateway/test_kanban_notify_poller.py b/tests/tui_gateway/test_kanban_notify_poller.py index 228c8b38a1e5..fd4374bb0bef 100644 --- a/tests/tui_gateway/test_kanban_notify_poller.py +++ b/tests/tui_gateway/test_kanban_notify_poller.py @@ -88,17 +88,23 @@ class TestCollectKanbanNotifications: def test_ignores_other_sessions_and_platforms(self): tid_other_session = _create_subscribed_task(chat_id="some-other-session") tid_gateway = _create_subscribed_task(platform="telegram", chat_id="chat-1") + # New subs start caught up at creation time (issue #29905); record the + # pre-completion cursors so we can assert they were never claimed. + pre_cursors = { + tid: _sub_rows(tid)[0]["last_event_id"] + for tid in (tid_other_session, tid_gateway) + } _complete(tid_other_session) _complete(tid_gateway) texts = _collect_kanban_notifications(_session()) assert texts == [] - # Foreign subscriptions untouched: cursors still 0, rows still there. + # Foreign subscriptions untouched: cursors unclaimed, rows still there. for tid in (tid_other_session, tid_gateway): rows = _sub_rows(tid) assert len(rows) == 1 - assert rows[0]["last_event_id"] == 0 + assert rows[0]["last_event_id"] == pre_cursors[tid] def test_no_session_key_is_a_noop(self): tid = _create_subscribed_task()