feat(gateway): expose Kanban wake metadata

gateway/kanban_watchers.py

@@ -532,6 +532,20 @@ def _collect():
                         )
                         if sub.get("thread_id") and not metadata.get("thread_id"):
                             metadata["thread_id"] = sub["thread_id"]
+                        # Preserve subscription routing metadata while adding
+                        # a separate, privacy-bounded identity contract for
+                        # this specific user-visible notification. Build a new
+                        # nested mapping for every event: callers can reconcile
+                        # retries from the claimed cursor without parsing the
+                        # localized text, and neither the stored subscription
+                        # dict nor arbitrary event/task data is exposed.
+                        metadata["hermes_kanban_notification"] = {
+                            "version": 1,
+                            "board": board_slug,
+                            "task_id": sub["task_id"],
+                            "event_kind": kind,
+                            "cursor": d["cursor"],
+                        }
                         # Adapters with no push channel (the API server —
                         # ``supports_async_delivery = False``) can NEVER
                         # satisfy a text-send: ``send()`` always reports
@@ -658,6 +672,7 @@ def _collect():
                         _is_push_adapter = _adapter_push_ok(adapter)
                         _session_key = ""
                         _synth = ""
+                        _wake_metadata: dict[str, Any] = {}
                         if _wake_kinds:
                             _session_key = getattr(task, "session_id", None) or ""
                         if _wake_kinds and _session_key:
@@ -678,6 +693,25 @@ def _collect():
                                 assignee=_assignee,
                                 board=board_slug,
                             )
+                            # Keep the localized text above for humans and
+                            # older adapters. Push-capable adapters also get a
+                            # deliberately small machine contract: only stable
+                            # routing/event identity from the claimed batch,
+                            # never task content, event payloads, or delivery
+                            # metadata from the subscription.
+                            _wake_metadata = {
+                                "hermes_kanban_wake": {
+                                    "version": 1,
+                                    "board": board_slug,
+                                    "task_id": sub["task_id"],
+                                    "event_kinds": [
+                                        kind
+                                        for kind in _WAKE_KINDS
+                                        if kind in _wake_kinds
+                                    ],
+                                    "cursor": d["cursor"],
+                                }
+                            }
 
                         if not _is_push_adapter and _wake_kinds and _session_key:
                             # Wake self-post IS the delivery on this path —
@@ -787,6 +821,7 @@ def _collect():
                                     text=_synth,
                                     session_id=_session_key,
                                     source=_source,
+                                    metadata=_wake_metadata,
                                 )
                                 logger.info(
                                     "kanban notifier: woke agent for %s on %s/%s profile=%s events=%s",

gateway/wake.py

@@ -26,8 +26,9 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import logging
-from typing import Any, Optional
+from typing import Any, Mapping, Optional
 
 logger = logging.getLogger(__name__)
 
@@ -59,13 +60,16 @@ async def deliver_wake(
     text: str,
     session_id: str = "",
     source: Any = None,
+    metadata: Optional[Mapping[str, Any]] = None,
 ) -> None:
     """Deliver a wake turn to the session behind ``adapter``.
 
     ``session_id`` is the RAW session id (the ``X-Hermes-Session-Id`` value /
     ``state.db`` key) — required for non-push adapters. ``source`` is the
     ``SessionSource`` used to build the synthetic event — required for
-    push-capable adapters.
+    push-capable adapters. ``metadata`` is copied onto that synthetic event so
+    adapters can identify machine-generated wake types without parsing
+    human-readable text. Stateless self-post wakes intentionally ignore it.
 
     Raises on failure (bad arguments, exhausted retries, HTTP error) so the
     caller can rewind/retry instead of treating the wake as delivered.
@@ -82,6 +86,7 @@ async def deliver_wake(
             message_type=MessageType.TEXT,
             source=source,
             internal=True,
+            metadata=copy.deepcopy(dict(metadata)) if metadata is not None else {},
         )
         await adapter.handle_message(synth_event)
         return

tests/gateway/test_kanban_notifier.py

@@ -2,7 +2,7 @@
 import sqlite3
 from pathlib import Path
 
-
+import pytest
 from gateway.config import Platform
 from gateway.kanban_watchers import (
     _acquire_singleton_lock,
@@ -66,6 +66,29 @@ def _create_completed_subscription(summary="done once"):
         conn.close()
 
 
+def _create_wake_subscription(event_kinds, *, payload=None):
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(
+            conn,
+            title="wake metadata",
+            assignee="worker",
+            session_id="origin-session",
+        )
+        kb.add_notify_sub(
+            conn,
+            task_id=tid,
+            platform="telegram",
+            chat_id="chat-1",
+        )
+        for kind in event_kinds:
+            kb._append_event(conn, tid, kind=kind, payload=payload)
+        cursor = kb.list_events(conn, tid)[-1].id
+        return tid, cursor
+    finally:
+        conn.close()
+
+
 def _unseen_terminal_events(tid):
     conn = kb.connect()
     try:
@@ -123,6 +146,13 @@ def test_kanban_notifier_replays_telegram_dm_topic_delivery_metadata(tmp_path, m
         "telegram_dm_topic_reply_fallback": True,
         "telegram_reply_to_message_id": "462",
         "thread_id": "20197",
+        "hermes_kanban_notification": {
+            "version": 1,
+            "board": "default",
+            "task_id": tid,
+            "event_kind": "completed",
+            "cursor": adapter.handled[0].metadata["hermes_kanban_wake"]["cursor"],
+        },
     }
     assert len(adapter.handled) == 1
     assert adapter.handled[0].source.chat_type == "dm"
@@ -395,6 +425,218 @@ def test_notifier_wakeup_uses_subscription_chat_type(tmp_path, monkeypatch):
     assert ":group:" not in wake_key
 
 
+@pytest.mark.parametrize(
+    "event_kind",
+    ["completed", "blocked", "gave_up", "crashed", "timed_out"],
+)
+def test_notifier_wake_exposes_terminal_event_identity(
+    tmp_path,
+    monkeypatch,
+    event_kind,
+):
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / f"{event_kind}.db"))
+    kb.init_db()
+    task_id, cursor = _create_wake_subscription([event_kind])
+
+    adapter = RecordingAdapter()
+    asyncio.run(_run_one_notifier_tick(monkeypatch, _make_runner(adapter)))
+
+    assert len(adapter.sent) == 1
+    assert adapter.sent[0]["metadata"] == {
+        "hermes_kanban_notification": {
+            "version": 1,
+            "board": "default",
+            "task_id": task_id,
+            "event_kind": event_kind,
+            "cursor": cursor,
+        }
+    }
+    assert len(adapter.handled) == 1
+    assert adapter.handled[0].metadata == {
+        "hermes_kanban_wake": {
+            "version": 1,
+            "board": "default",
+            "task_id": task_id,
+            "event_kinds": [event_kind],
+            "cursor": cursor,
+        }
+    }
+
+
+def test_notifier_wake_metadata_is_ordered_and_privacy_bounded(
+    tmp_path,
+    monkeypatch,
+):
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "bounded.db"))
+    kb.init_db()
+    task_id, cursor = _create_wake_subscription(
+        ["timed_out", "completed", "blocked", "crashed", "gave_up"],
+        payload={
+            "arbitrary": "must-not-leak",
+            "assignee": "private-worker",
+            "workspace": "/private/worktree",
+            "credential_hint": "must-not-leak",
+            "delivery_metadata": {"thread_id": "private-thread"},
+        },
+    )
+
+    adapter = RecordingAdapter()
+    asyncio.run(_run_one_notifier_tick(monkeypatch, _make_runner(adapter)))
+
+    assert len(adapter.handled) == 1
+    assert [
+        item["metadata"]["hermes_kanban_notification"]
+        for item in adapter.sent
+    ] == [
+        {
+            "version": 1,
+            "board": "default",
+            "task_id": task_id,
+            "event_kind": kind,
+            "cursor": cursor,
+        }
+        for kind in ["timed_out", "completed", "blocked", "crashed", "gave_up"]
+    ]
+    assert adapter.handled[0].metadata == {
+        "hermes_kanban_wake": {
+            "version": 1,
+            "board": "default",
+            "task_id": task_id,
+            "event_kinds": [
+                "completed",
+                "gave_up",
+                "crashed",
+                "timed_out",
+                "blocked",
+            ],
+            "cursor": cursor,
+        }
+    }
+
+    # The claim cursor remains the exact-once boundary: a later poll has no
+    # terminal event to deliver or wake again.
+    runner = _make_runner(adapter)
+    asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
+    assert len(adapter.sent) == 5
+    assert len(adapter.handled) == 1
+
+
+class FailOnceRecordingAdapter(RecordingAdapter):
+    def __init__(self):
+        super().__init__()
+        self.attempted_metadata = []
+
+    async def send(self, chat_id, text, metadata=None):
+        import copy
+
+        self.attempted_metadata.append(copy.deepcopy(metadata or {}))
+        if len(self.attempted_metadata) == 1:
+            raise RuntimeError("simulated transient failure")
+        await super().send(chat_id, text, metadata=metadata)
+
+
+def test_notifier_retry_preserves_notification_identity_without_duplicates(
+    tmp_path,
+    monkeypatch,
+):
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "retry-identity.db"))
+    kb.init_db()
+    task_id, cursor = _create_wake_subscription(
+        ["blocked"],
+        payload={"reason": "visible reason", "secret": "must-not-leak"},
+    )
+
+    adapter = FailOnceRecordingAdapter()
+    asyncio.run(_run_one_notifier_tick(monkeypatch, _make_runner(adapter)))
+    assert adapter.sent == []
+    assert adapter.handled == []
+
+    asyncio.run(_run_one_notifier_tick(monkeypatch, _make_runner(adapter)))
+    expected = {
+        "hermes_kanban_notification": {
+            "version": 1,
+            "board": "default",
+            "task_id": task_id,
+            "event_kind": "blocked",
+            "cursor": cursor,
+        }
+    }
+    assert adapter.attempted_metadata == [expected, expected]
+    assert [item["metadata"] for item in adapter.sent] == [expected]
+    assert len(adapter.handled) == 1
+
+    # A successful cursor advance is the exact-once boundary for both planes.
+    asyncio.run(_run_one_notifier_tick(monkeypatch, _make_runner(adapter)))
+    assert adapter.attempted_metadata == [expected, expected]
+    assert len(adapter.sent) == 1
+    assert len(adapter.handled) == 1
+
+
+def test_notifier_does_not_mutate_stored_subscription_metadata(
+    tmp_path,
+    monkeypatch,
+):
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "metadata-copy.db"))
+    kb.init_db()
+    original = {
+        "chat_type": "dm",
+        "thread_id": "thread-7",
+        "delivery_secret": "routing-value",
+    }
+    conn = kb.connect()
+    try:
+        task_id = kb.create_task(conn, title="copy metadata", assignee="worker")
+        kb.add_notify_sub(
+            conn,
+            task_id=task_id,
+            platform="telegram",
+            chat_id="chat-1",
+            thread_id="thread-7",
+            delivery_metadata=original,
+        )
+        kb._append_event(
+            conn,
+            task_id,
+            kind="crashed",
+            payload={
+                "summary": "must-not-leak",
+                "result": "must-not-leak",
+                "reason": "must-not-leak",
+                "assignee": "private-profile",
+                "workspace": "/private/worktree",
+                "api_token": "must-not-leak",
+            },
+        )
+        cursor = kb.list_events(conn, task_id)[-1].id
+    finally:
+        conn.close()
+
+    adapter = RecordingAdapter()
+    asyncio.run(_run_one_notifier_tick(monkeypatch, _make_runner(adapter)))
+
+    assert original == {
+        "chat_type": "dm",
+        "thread_id": "thread-7",
+        "delivery_secret": "routing-value",
+    }
+    assert adapter.sent[0]["metadata"] == {
+        **original,
+        "hermes_kanban_notification": {
+            "version": 1,
+            "board": "default",
+            "task_id": task_id,
+            "event_kind": "crashed",
+            "cursor": cursor,
+        },
+    }
+    conn = kb.connect()
+    try:
+        stored = kb.list_notify_subs(conn, task_id)[0]["delivery_metadata"]
+    finally:
+        conn.close()
+    assert stored == original
+
+
 def _unseen_terminal_events_for(tid, chat_id):
     conn = kb.connect()
     try: