diff --git a/gateway/kanban_watchers_notifier.py b/gateway/kanban_watchers_notifier.py index dba036b86a..68debcaf2a 100644 --- a/gateway/kanban_watchers_notifier.py +++ b/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() diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 37972956be..9db82e46fe 100644 --- a/gateway/platforms/base.py +++ b/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: diff --git a/gateway/platforms/event.py b/gateway/platforms/event.py index 0de062922b..29fbf983bf 100644 --- a/gateway/platforms/event.py +++ b/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("/") diff --git a/gateway/run_busy.py b/gateway/run_busy.py index 448af230ad..b1ff931516 100644 --- a/gateway/run_busy.py +++ b/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: diff --git a/gateway/wake.py b/gateway/wake.py index 08b05396bc..4b70daf5a7 100644 --- a/gateway/wake.py +++ b/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) " diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 210c6eed3c..df1cec02b2 100644 --- a/hermes_cli/kanban_db.py +++ b/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) ); diff --git a/hermes_cli/kanban_db_connect.py b/hermes_cli/kanban_db_connect.py index dbd5e4becf..6db4994cb7 100644 --- a/hermes_cli/kanban_db_connect.py +++ b/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)",), ), diff --git a/hermes_cli/kanban_db_notify.py b/hermes_cli/kanban_db_notify.py index 12fb49d61c..c612631cc0 100644 --- a/hermes_cli/kanban_db_notify.py +++ b/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, *, diff --git a/plugins/platforms/raft/adapter.py b/plugins/platforms/raft/adapter.py index 7a7ed4c395..a3af407d83 100644 --- a/plugins/platforms/raft/adapter.py +++ b/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( diff --git a/tests/gateway/test_kanban_changes_requested_notifier.py b/tests/gateway/test_kanban_changes_requested_notifier.py index 374b26ad3a..17980011a7 100644 --- a/tests/gateway/test_kanban_changes_requested_notifier.py +++ b/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): diff --git a/tests/gateway/test_kanban_notifier.py b/tests/gateway/test_kanban_notifier.py index 01a78de33e..c8f59e437f 100644 --- a/tests/gateway/test_kanban_notifier.py +++ b/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): diff --git a/tests/gateway/test_kanban_notifier_wake_only_ordering.py b/tests/gateway/test_kanban_notifier_wake_only_ordering.py index 6f78ac31dd..ce5f3c5489 100644 --- a/tests/gateway/test_kanban_notifier_wake_only_ordering.py +++ b/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): diff --git a/tests/gateway/test_kanban_routed_transport.py b/tests/gateway/test_kanban_routed_transport.py index e00a9d8b5b..5768d32af8 100644 --- a/tests/gateway/test_kanban_routed_transport.py +++ b/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): diff --git a/tests/gateway/test_kanban_wake_acceptance.py b/tests/gateway/test_kanban_wake_acceptance.py new file mode 100644 index 0000000000..c7cb61ac20 --- /dev/null +++ b/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() diff --git a/tests/gateway/test_kanban_wake_scope.py b/tests/gateway/test_kanban_wake_scope.py index 1d8f26707e..fa5a6653bb 100644 --- a/tests/gateway/test_kanban_wake_scope.py +++ b/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 diff --git a/website/docs/developer-guide/gateway-internals.md b/website/docs/developer-guide/gateway-internals.md index e321569e65..5b92103db3 100644 --- a/website/docs/developer-guide/gateway-internals.md +++ b/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. diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index 413d355e58..01012f9ff8 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/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.