fix: retry Kanban wakes until adapter admission

gateway/kanban_watchers_notifier.py

@@ -8,6 +8,7 @@ per-subscription delivery (``_KanbanNotification``) live here.
 from __future__ import annotations
 
 import re
+from functools import partial
 from pathlib import Path
 import weakref
 from typing import Any, Callable, Optional
@@ -372,12 +373,9 @@ _EVENT_FORMATTERS: dict[str, Callable[[Any, "_KanbanNotification"], tuple]] = {
 class _KanbanNotification:
     """Deliver one subscription's claimed events, then settle the cursor.
 
-    Cursor advance ordering by adapter class:
-    * push + notify: the text send WAS the delivery → advance now; wake
-      injection stays best-effort.
-    * non-push or wake-only: the wake IS the delivery → it runs FIRST and the
-      cursor advances only after it succeeds; failure rewinds like a failed
-      send(). An unknown platform advances the cursor so it can't replay forever.
+    Both legs of notify+wake must succeed before settling. Sent pings have a
+    separate durable checkpoint so wake retries do not resend them. Admission
+    is at-least-once queueing, not an execution or final-response receipt.
     """
 
     def __init__(self, runner: Any, d: dict, *, platform_cls: Any, sub_fail_counts: dict) -> None:
@@ -569,8 +567,14 @@ class _KanbanNotification:
             if not self.send_passive:
                 # Wake-only: the wake path is the sole delivery and resolves the counter.
                 continue
+            if ev.id <= self.sub.get("last_ping_event_id", 0):
+                continue
             try:
                 await self._send_event(ev, msg)
+                await _to_thread_process_service(partial(
+                    self.runner._kanban_sub_op, self.board_slug, "record_notify_ping", self.sub,
+                    event_id=ev.id,
+                ))
                 self.clear_failures()
             except Exception as exc:
                 await self.delivery_failed(
@@ -602,13 +606,18 @@ class _KanbanNotification:
         # All text pings delivered (or skipped for non-push / wake-only).
         self.build_wake_text()
         wake_kinds, is_push = self.wake_kinds, self.is_push_adapter
+        from gateway.wake import WakeNotAccepted
 
-        # Non-push self-post, or wake-only push sub: the wake IS the delivery
-        # and must succeed BEFORE the cursor advances.
-        if wake_kinds and (not self.send_passive if is_push else bool(self.session_key)):
+        # A requested wake is required even when its passive ping already landed.
+        if wake_kinds:
             try:
                 await self.wake()
                 self.clear_failures()
+            except WakeNotAccepted:
+                # Startup / full queue is not a dead destination. Keep the durable
+                # subscription alive regardless of how long admission takes.
+                await self.rewind()
+                return
             except Exception as _wk_err:
                 await self._wake_failed(
                     "kanban notifier: wake-only delivery failed for %s (attempt %d/%d): %s" if is_push
@@ -621,14 +630,6 @@ class _KanbanNotification:
         await self.advance()
         if not is_push:
             self.clear_failures()
-        if is_push and self.send_passive and wake_kinds:
-            # notify+wake: text ping was the delivery and the cursor has
-            # advanced; the wake stays best-effort, but log at WARNING so a
-            # persistently failing wake is visible.
-            try:
-                await self.wake()
-            except Exception as _wk_err:
-                logger.warning("kanban notifier: wakeup injection failed for %s: %s", self.task_id, _wk_err, exc_info=True)
         # Unsubscribe only on archive; ``done`` is reversible.
         if self.task and self.task.status == "archived":
             await self.unsub()

gateway/platforms/base.py

@@ -3479,6 +3479,7 @@ class BasePlatformAdapter(ABC):
     async def handle_message(self, event: MessageEvent) -> None:
         """Process an incoming message; returns quickly by spawning a background
         task so new messages (and interrupts) can arrive while an agent runs."""
+        event._gateway_accepted = False
         if not self._message_handler:
             return
         if event.allow_gateway_control:
@@ -3501,7 +3502,7 @@ class BasePlatformAdapter(ABC):
             await self._handle_message_while_active(event, session_key)
             return
         # Guard installed synchronously BEFORE the task spawns so a second message can't race in.
-        self._start_session_processing(event, session_key)
+        event._gateway_accepted = self._start_session_processing(event, session_key)
 
     async def _handle_message_while_active(self, event: MessageEvent, session_key: str) -> None:
         """Route a message that arrived while ``session_key`` is busy: bypass
@@ -3554,10 +3555,15 @@ class BasePlatformAdapter(ABC):
                     return
             except Exception as e:
                 logger.error("[%s] Busy-session handler failed: %s", self.name, e, exc_info=True)
+        # Without a runner FIFO, do not merge a wake into an occupied human slot
+        # (or collapse distinct wakes into one turn). Its caller can retry admission.
+        if event.internal and session_key in self._pending_messages:
+            return
         # Photo bursts/albums: queue without interrupting; they run after the current task.
         if event.message_type == MessageType.PHOTO:
             logger.debug("[%s] Queuing photo follow-up for session %s without interrupt", self.name, session_key)
             merge_pending_message_event(self._pending_messages, session_key, event)
+            event._gateway_accepted = True
             return
         if self._is_queue_text_debounce_candidate(event):
             logger.debug("[%s] New text message while session %s is active — "
@@ -3569,6 +3575,7 @@ class BasePlatformAdapter(ABC):
                          "(no interrupt, will cascade after current turn)", self.name, session_key)
             merge_pending_message_event(self._pending_messages, session_key, event,
                                         merge_text=event.message_type == MessageType.TEXT)
+            event._gateway_accepted = True
 
     @staticmethod
     def _get_human_delay() -> float:

gateway/platforms/event.py

@@ -77,6 +77,9 @@ class MessageEvent:
     # so untrusted payload text stays conversational. Kept last for positional compat.
     allow_gateway_control: bool = True
 
+    # Process-local admission receipt, never routing metadata or execution acknowledgement.
+    _gateway_accepted: bool = field(default=False, init=False, repr=False, compare=False)
+
     def is_command(self) -> bool:
         """Check if this is a command message (e.g., /new, /reset)."""
         return self.allow_gateway_control and (self.text or "").lstrip().startswith("/")

gateway/run_busy.py

@@ -52,6 +52,7 @@ class GatewayBusySessionMixin:
             self._session_state(session_key).conversation.queued_events.append(queued_event)
         else:
             pending_slot[session_key] = queued_event
+        queued_event._gateway_accepted = True
 
     def _promote_queued_event(
         self, session_key: str, adapter: Any, pending_event: Optional["MessageEvent"]
@@ -310,6 +311,7 @@ class GatewayBusySessionMixin:
                 adapter._pending_messages, session_key, event,
                 merge_text=event.message_type == MessageType.TEXT,
             )
+            event._gateway_accepted = True
             return
 
         if self._queue_depth(session_key, adapter=adapter) >= self._BUSY_QUEUE_MAX_PENDING:

gateway/wake.py

@@ -35,6 +35,22 @@ def adapter_supports_push(adapter: Any) -> bool:
     return bool(getattr(adapter, "supports_async_delivery", True))
 
 
+class WakeNotAccepted(RuntimeError):
+    """No adapter admission: retry without treating a healthy chat as dead."""
+
+
+async def admit_internal_event(adapter: Any, event: Any) -> None:
+    """Require a concrete adapter admission, not merely a handler returning None.
+
+    The public handler return stays unchanged. This receipt means scheduled/queued,
+    not model execution, authorization of a later turn, or successful outbound delivery.
+    """
+    event._gateway_accepted = False
+    await adapter.handle_message(event)
+    if event._gateway_accepted is not True:
+        raise WakeNotAccepted("internal wake not accepted by adapter")
+
+
 async def deliver_wake(adapter: Any, *, text: str, session_id: str = "", source: Any = None) -> None:
     """Deliver a wake turn to the session behind ``adapter``. ``session_id`` is the RAW session id
     (``X-Hermes-Session-Id`` / state.db key) — required for non-push adapters. ``source`` is the
@@ -45,7 +61,7 @@ async def deliver_wake(adapter: Any, *, text: str, session_id: str = "", source:
             raise ValueError("deliver_wake: push-capable adapter requires a SessionSource")
         from gateway.platforms.event import MessageEvent, MessageType
         synth_event = MessageEvent(text=text, message_type=MessageType.TEXT, source=source, internal=True)
-        await adapter.handle_message(synth_event)
+        await admit_internal_event(adapter, synth_event)
         return
     if not session_id:
         raise ValueError("deliver_wake: non-push adapter (supports_async_delivery=False) "

hermes_cli/kanban_db.py

@@ -1030,6 +1030,7 @@ CREATE TABLE IF NOT EXISTS kanban_notify_subs (
     delivery_metadata TEXT,
     created_at    INTEGER NOT NULL,
     last_event_id INTEGER NOT NULL DEFAULT 0,
+    last_ping_event_id INTEGER NOT NULL DEFAULT 0,
     PRIMARY KEY (task_id, platform, chat_id, thread_id)
 );
 

hermes_cli/kanban_db_connect.py

@@ -816,6 +816,7 @@ _LATER_TASK_COLUMNS = (
 )
 
 _NOTIFY_SUB_COLUMNS = (
+    ("last_ping_event_id", "last_ping_event_id INTEGER NOT NULL DEFAULT 0"),
     ("notifier_profile", "notifier_profile TEXT"),
     ("delivery_mode", "delivery_mode TEXT NOT NULL DEFAULT 'notify'"),
     ("chat_type", "chat_type TEXT"),
@@ -1016,6 +1017,7 @@ _REBUILD_SPECS = {
         " notifier_profile TEXT, delivery_mode TEXT NOT NULL DEFAULT 'notify',"
         " delivery_metadata TEXT, created_at INTEGER NOT NULL,"
         " last_event_id INTEGER NOT NULL DEFAULT 0,"
+        " last_ping_event_id INTEGER NOT NULL DEFAULT 0,"
         " PRIMARY KEY (task_id, platform, chat_id, thread_id))",
         ("CREATE INDEX idx_notify_task ON kanban_notify_subs(task_id)",),
     ),

hermes_cli/kanban_db_notify.py

@@ -393,6 +393,19 @@ def advance_notify_cursor(
         )
 
 
+def record_notify_ping(
+    conn: sqlite3.Connection, *, task_id: str, platform: str, chat_id: str,
+    thread_id: Optional[str] = None, event_id: int,
+) -> None:
+    """Checkpoint a sent ping independently of the retryable wake cursor."""
+    with _kb.write_txn(conn):
+        conn.execute(
+            "UPDATE kanban_notify_subs SET last_ping_event_id = MAX(last_ping_event_id, ?) "
+            + _SUB_KEY_WHERE,
+            (int(event_id), *_sub_key(task_id, platform, chat_id, thread_id)),
+        )
+
+
 def rewind_notify_cursor(
     conn: sqlite3.Connection,
     *,

plugins/platforms/raft/adapter.py

@@ -485,6 +485,10 @@ class RaftAdapter(BasePlatformAdapter):
 
     async def handle_message(self, event: MessageEvent) -> None:
         """Accept Raft wake hints without interrupting an active Hermes turn."""
+        if event.internal:
+            # Durable gateway wakes need the base session fence and admission receipt.
+            await super().handle_message(event)
+            return
         if not self._message_handler:
             return
         session_key = build_session_key(

tests/gateway/test_kanban_changes_requested_notifier.py

@@ -20,6 +20,7 @@ class RecordingAdapter:
 
     async def handle_message(self, event):
         self.handled.append(event)
+        event._gateway_accepted = True
 
 
 async def _run_one_tick(monkeypatch, runner):

tests/gateway/test_kanban_notifier.py

@@ -24,6 +24,7 @@ class RecordingAdapter:
 
     async def handle_message(self, event):
         self.handled.append(event)
+        event._gateway_accepted = True
 
 
 class DisconnectedAdapters(dict):

tests/gateway/test_kanban_notifier_wake_only_ordering.py

@@ -31,6 +31,7 @@ class RecordingAdapter:
 
     async def handle_message(self, event):
         self.handled.append(event)
+        event._gateway_accepted = True
 
 
 class FailingWakeAdapter(RecordingAdapter):
@@ -157,8 +158,8 @@ def test_wake_only_failure_rewinds_and_redelivers(tmp_path, monkeypatch):
     assert list(runner2._kanban_sub_fail_counts.values()) == [2]
 
 
-def test_notify_wake_failure_stays_best_effort(tmp_path, monkeypatch):
-    """notify+wake: text ping IS the delivery; failed wake must NOT rewind."""
+def test_notify_wake_failure_retries_without_repeating_ping(tmp_path, monkeypatch):
+    """notify+wake requires both deliveries, retaining the sent-ping checkpoint."""
     monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "notify-wake.db"))
     kb.init_db()
     tid = _make_completed_task("notify+wake")
@@ -168,16 +169,14 @@ def test_notify_wake_failure_stays_best_effort(tmp_path, monkeypatch):
     asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
 
     assert len(adapter.sent) == 1, "text ping delivered"
-    assert len(adapter.handled) == 1, "wake attempted best-effort"
-    assert _unseen_terminal_events(tid) == [], (
-        "notify+wake: cursor advances on text delivery; a failed wake is "
-        "best-effort and must not rewind"
-    )
-    assert runner._kanban_sub_fail_counts == {}, (
-        "best-effort wake failure must not bump the send-failure counter"
-    )
-    # (The sub itself unsubscribes because the task reached 'done' —
-    # pre-existing task_terminal behavior, unrelated to the wake outcome.)
+    assert len(adapter.handled) == 1, "wake attempted after the ping"
+    assert len(_unseen_terminal_events(tid)) == 1
+    assert list(runner._kanban_sub_fail_counts.values()) == [1]
+    runner._running = True
+    asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
+    assert len(adapter.sent) == 1
+    assert len(adapter.handled) == 2
+    assert list(runner._kanban_sub_fail_counts.values()) == [2]
 
 
 def test_wake_only_failure_cap_drops_subscription(tmp_path, monkeypatch):

tests/gateway/test_kanban_routed_transport.py

@@ -21,6 +21,7 @@ class RecordingAdapter:
 
     async def handle_message(self, event):
         self.handled.append(event)
+        event._gateway_accepted = True
 
 
 def setup_runner(tmp_path, monkeypatch):

tests/gateway/test_kanban_wake_acceptance.py

@@ -0,0 +1,150 @@
+"""Push admission and durable notifier retries use real adapter/SQLite lifecycles."""
+import asyncio
+
+import pytest
+
+from evals.heartbeat_idle_wire import WireAdapter
+from gateway.config import Platform, PlatformConfig
+from gateway.kanban_watchers_notifier import _KanbanNotification, _notifier_collect
+from gateway.platforms.event import MessageEvent
+from gateway.run import GatewayRunner
+from gateway.session import SessionSource, build_session_key
+from gateway.wake import admit_internal_event, deliver_wake
+from hermes_cli import kanban_db as kb
+from hermes_cli import kanban_db_connect as kbc
+from hermes_cli import kanban_db_notify as kbn
+
+
+def setup_route(raft=False):
+    if raft:
+        from plugins.platforms.raft.adapter import RaftAdapter
+        adapter = RaftAdapter(PlatformConfig(enabled=True, typing_indicator=False,
+            extra={"bridge_token": "owned-test-token", "port": 0}))
+    else:
+        adapter = WireAdapter(PlatformConfig(enabled=True, typing_indicator=False), Platform.TELEGRAM)
+    adapter.wire = []
+    runner = object.__new__(GatewayRunner)
+    runner._running_agents = {}
+    runner.adapters = {adapter.platform: adapter}
+    runner._adapter_for_source = lambda source: adapter
+    runner._kanban_dispatcher_lock_handle = object()
+    source = SessionSource(platform=adapter.platform, chat_id="42", user_id="42", chat_type="dm")
+    return runner, adapter, source, build_session_key(source)
+
+
+async def drain(adapter):
+    while adapter._background_tasks:
+        await asyncio.gather(*list(adapter._background_tasks))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("raft", [False, True])
+async def test_push_receipt_requires_real_admission_without_displacing_user(raft, monkeypatch):
+    runner, adapter, source, key = setup_route(raft)
+    if raft:
+        monkeypatch.setattr(adapter, "_spawn_bridge", lambda port: None)
+    release, started = asyncio.Event(), asyncio.Event()
+    received = []
+
+    async def handler(event):
+        received.append(event.text)
+        started.set()
+        await release.wait()
+        # Real runner FIFO promotion at the fake model boundary.
+        if key not in adapter._pending_messages:
+            pending = runner._promote_queued_event(key, adapter, None)
+            if pending is not None:
+                adapter._pending_messages[key] = pending
+
+    adapter.set_message_handler(handler)
+    await adapter.connect()
+    try:
+        await deliver_wake(adapter, text="idle", source=source)
+        await asyncio.wait_for(started.wait(), 2)
+        assert await adapter.handle_message(MessageEvent(text="human", source=source)) is None
+        with pytest.raises(RuntimeError, match="not accepted"):
+            await deliver_wake(adapter, text="no-fifo", source=source)
+        assert adapter._pending_messages[key].text == "human"
+        adapter.set_busy_session_handler(runner._handle_active_session_busy_message)
+        runner._BUSY_QUEUE_MAX_PENDING = 1
+        with pytest.raises(RuntimeError, match="not accepted"):
+            await deliver_wake(adapter, text="at-cap", source=source)
+        assert adapter._pending_messages[key].text == "human"
+        runner._BUSY_QUEUE_MAX_PENDING = 3
+        await deliver_wake(adapter, text="busy", source=source)
+        assert runner._queue_depth(key, adapter=adapter) == 2
+        rejected = MessageEvent(text="wrong-key", source=source, internal=True,
+                                metadata={"gateway_session_key": "agent:wrong"})
+        with pytest.raises(RuntimeError, match="not accepted"):
+            await admit_internal_event(adapter, rejected)
+        assert rejected._gateway_accepted is False
+        release.set()
+        await drain(adapter)
+        assert received == ["idle", "human", "busy"]
+        adapter.set_message_handler(None)
+        with pytest.raises(RuntimeError, match="not accepted"):
+            await deliver_wake(adapter, text="no-handler", source=source)
+    finally:
+        release.set()
+        await drain(adapter)
+        await adapter.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_notifier_retries_unaccepted_wake_without_repeating_pings(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "board.db"))
+    runner, adapter, source, key = setup_route()
+    conn = kbc.connect()
+    tids = {}
+    try:
+        for mode in ("notify+wake", "wake", "notify"):
+            tid = kb.create_task(conn, title=mode, assignee="worker", session_id=key)
+            kbn.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="42",
+                               user_id="42", chat_type="dm", delivery_mode=mode)
+            kb.complete_task(conn, tid, summary="handoff")
+            tids[mode] = tid
+    finally:
+        conn.close()
+
+    failures = {}
+
+    async def tick():
+        deliveries = await asyncio.to_thread(_notifier_collect, runner, kb,
+            notifier_profile=None, gc_due=False, gc_retention_days=30)
+        for delivery in deliveries:
+            await _KanbanNotification(runner, delivery, platform_cls=Platform,
+                                      sub_fail_counts=failures).deliver()
+
+    def unseen(mode):
+        conn = kbc.connect()
+        try:
+            return kbn.unseen_events_for_sub(conn, task_id=tids[mode], platform="telegram",
+                                            chat_id="42", kinds=["completed"])[1]
+        finally:
+            conn.close()
+
+    await adapter.connect()
+    await tick()  # send works, but no message handler has been installed yet
+    assert len(adapter.wire) == 2
+    assert unseen("notify+wake") and unseen("wake")
+    assert not unseen("notify")
+    # New notifier instances and DB connections replay the durable claim, not the ping.
+    for _ in range(13):
+        await tick()
+    assert len(adapter.wire) == 2
+    assert unseen("notify+wake") and unseen("wake")
+    assert failures == {}
+    received = []
+
+    async def handler(event):
+        received.append(event.text)
+
+    adapter.set_message_handler(handler)
+    await tick()
+    await drain(adapter)
+    await tick()
+    assert len(received) == 2
+    assert all("handoff" in text for text in received)
+    assert len(adapter.wire) == 2
+    assert not any(unseen(mode) for mode in tids)
+    await adapter.disconnect()

tests/gateway/test_kanban_wake_scope.py

@@ -38,6 +38,7 @@ class UnscopedAdapter:
 
     async def handle_message(self, event):
         self.handled.append(event)
+        event._gateway_accepted = True
 
 
 def _slack_adapter(channel_team=None):
@@ -47,7 +48,7 @@ def _slack_adapter(channel_team=None):
     adapter._app.client = AsyncMock()
     adapter._running = True
     adapter.send = AsyncMock()
-    adapter.handle_message = AsyncMock()
+    adapter.handle_message = AsyncMock(side_effect=lambda event: setattr(event, "_gateway_accepted", True))
     if channel_team:
         adapter._channel_team.update(channel_team)
     return adapter

website/docs/developer-guide/gateway-internals.md

@@ -183,6 +183,16 @@ Adapters implement a common interface:
 - `send()` — outbound message delivery
 - inbound events are normalized into a `MessageEvent` and forwarded via `handle_message()`
 
+Internal push wakes use `gateway.wake.admit_internal_event`: the public
+`handle_message()` still returns `None`, but the event's process-local
+`_gateway_accepted` receipt is set only after scheduling or queue insertion.
+A missing handler, mismatched explicit session key, or queue-cap drop is not
+acceptance. Custom adapters overriding ingress should delegate internal events to
+`BasePlatformAdapter.handle_message()` (or explicitly record actual admission),
+not equate a consumed/dropped callback with acceptance. This receipt is separate
+from heartbeat execution accounting and does not bypass authorization, emergency
+stop, or later turn-preparation gates.
+
 ### Token Locks
 
 Adapters that connect with unique credentials call `acquire_scoped_lock()` in `connect()` and `release_scoped_lock()` in `disconnect()`. This prevents two profiles from using the same bot token simultaneously.

website/docs/user-guide/features/kanban.md

@@ -1111,6 +1111,8 @@ A subscription removes itself automatically once the task reaches `done` or `arc
 | `notify+wake` | yes | yes | You also want the destination agent to take a real turn — read the board context and reply in its own voice. Chat-originated auto-subscribes use this. |
 | `wake` | no | yes | You only want the agent to act on the event, with no separate ping. |
 
+For `notify+wake`, delivery completes only once the wake is admitted to the adapter's turn queue as well as the passive ping being sent. Missing handlers, rejected routes, and full queues are retried on later notifier ticks without expiring the subscription. Sent pings are checkpointed separately in SQLite, so a rejected wake does not repeat an already checkpointed ping. `notify` remains passive and never starts a turn. Admission is not a guarantee of model execution or a successful reply; normal turn gates still apply. This is not exactly-once delivery: a process crash between a send and its checkpoint can repeat the ping, and the existing claim-before-delivery cursor is not a crash-recoverable queue.
+
 A "wake" forges a synthetic inbound message to the destination gateway agent so it takes a normal turn (reads the comment + result, reasons, replies) instead of getting a one-line passive notification. It only fires when the notifier runs inside a live gateway process; otherwise a `notify+wake` subscription still delivers its passive message, while a `wake`-only subscription does nothing in that process.
 
 **Which events wake.** The ones that hand a decision back to the origin: `completed`, `blocked`, `gave_up`, `crashed`, `timed_out`, `review_requested` (a worker finished the implementation and handed off via `kanban_request_review`) and `block_loop_detected` (the task was routed to `triage` after repeated blocks). `status`, `archived` and `unblocked` are delivered but never wake — they are bookkeeping transitions, not decisions. When a `completed` or `review_requested` event carries a summary, that handoff rides the wake turn, so the woken agent sees what the worker actually did.