fix(kanban): snap notify-sub cursor to current MAX(task_events.id) at creation

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

@@ -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:

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()

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

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()