From 0f7606a13ba4d1a83a9a018124d1ecd36881c12a Mon Sep 17 00:00:00 2001 From: Rasmus Hjulskov Date: Thu, 13 Aug 2026 16:55:40 +0200 Subject: [PATCH 1/2] feat(gateway): expose Kanban wake metadata --- gateway/kanban_watchers.py | 21 +++++ gateway/wake.py | 9 +- tests/gateway/test_kanban_notifier.py | 98 +++++++++++++++++++++- tests/gateway/test_wake_delivery.py | 38 ++++++++- website/docs/user-guide/features/kanban.md | 9 ++ 5 files changed, 171 insertions(+), 4 deletions(-) diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 1a34feb12b9f..948308bfa95a 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -658,6 +658,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 +679,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 +807,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", diff --git a/gateway/wake.py b/gateway/wake.py index cee8d45e50c3..3dad2d8d4607 100644 --- a/gateway/wake.py +++ b/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 diff --git a/tests/gateway/test_kanban_notifier.py b/tests/gateway/test_kanban_notifier.py index 7400efd7263d..14c69f545e56 100644 --- a/tests/gateway/test_kanban_notifier.py +++ b/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: @@ -395,6 +418,79 @@ 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.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 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 + + def _unseen_terminal_events_for(tid, chat_id): conn = kb.connect() try: diff --git a/tests/gateway/test_wake_delivery.py b/tests/gateway/test_wake_delivery.py index 3d24e34f7070..f2712363d50d 100644 --- a/tests/gateway/test_wake_delivery.py +++ b/tests/gateway/test_wake_delivery.py @@ -53,6 +53,37 @@ def test_adapter_supports_push_default_true(): assert adapter_supports_push(ApiServerLikeAdapter()) is False +def test_deliver_wake_push_defaults_are_unchanged(): + adapter = PushAdapter() + + asyncio.run(deliver_wake(adapter, text="wake", source=_source())) + + assert len(adapter.handled) == 1 + event = adapter.handled[0] + assert event.text == "wake" + assert event.internal is True + assert event.metadata == {} + + +def test_deliver_wake_push_copies_metadata(): + adapter = PushAdapter() + metadata = {"custom_wake": {"version": 1, "items": ["one"]}} + + asyncio.run( + deliver_wake( + adapter, + text="wake", + source=_source(), + metadata=metadata, + ) + ) + metadata["custom_wake"]["items"].append("two") + + assert adapter.handled[0].metadata == { + "custom_wake": {"version": 1, "items": ["one"]} + } + + async def _serve(handler): """Spin an in-process aiohttp server on an ephemeral loopback port.""" from aiohttp import web @@ -85,7 +116,12 @@ async def run(): runner, port = await _serve(handler) try: adapter = ApiServerLikeAdapter(host="0.0.0.0", port=port, key="sekrit") - await deliver_wake(adapter, text="task done — wake", session_id="raw-sid-42") + await deliver_wake( + adapter, + text="task done — wake", + session_id="raw-sid-42", + metadata={"custom_wake": {"version": 1}}, + ) finally: await runner.cleanup() diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index 7ee2942d2b2f..8b42c5f41f8f 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/website/docs/user-guide/features/kanban.md @@ -1034,6 +1034,15 @@ Duplicate delivery across gateways is prevented by the atomic per-event claim in the board DB. No relays, credential sharing, or extra dispatchers are needed — each profile gateway simply delivers through its own adapters. +Push-capable platform adapters receive Kanban terminal wakes as internal +`MessageEvent` objects. In addition to the localized wake text, adapters can +identify these events through `event.metadata["hermes_kanban_wake"]`, a +versioned envelope containing only `board`, `task_id`, deterministically ordered +`event_kinds`, and the claimed event `cursor`. Delivery metadata, event payloads, +task content, and profile details are intentionally excluded. Stateless API +self-post wakes keep their existing request shape and do not receive this +adapter-only metadata. + ## Runs — one row per attempt A task is a logical unit of work; a **run** is one attempt to execute it. When the dispatcher claims a ready task it creates a row in `task_runs` and points `tasks.current_run_id` at it. When that attempt ends — completed, blocked, crashed, timed out, spawn-failed, reclaimed — the run row closes with an `outcome` and the task's pointer clears. A task that's been attempted three times has three `task_runs` rows. From 8ca79ec96929087a7d014eed8abdfef7d44a816d Mon Sep 17 00:00:00 2001 From: Rasmus Hjulskov Date: Thu, 13 Aug 2026 17:09:18 +0200 Subject: [PATCH 2/2] fix(gateway): identify direct Kanban notifications --- gateway/kanban_watchers.py | 14 ++ tests/gateway/test_kanban_notifier.py | 146 +++++++++++++++++++++ website/docs/user-guide/features/kanban.md | 27 ++-- 3 files changed, 179 insertions(+), 8 deletions(-) diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 948308bfa95a..463cf5d7089f 100644 --- a/gateway/kanban_watchers.py +++ b/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 diff --git a/tests/gateway/test_kanban_notifier.py b/tests/gateway/test_kanban_notifier.py index 14c69f545e56..a97fcf6bad3a 100644 --- a/tests/gateway/test_kanban_notifier.py +++ b/tests/gateway/test_kanban_notifier.py @@ -146,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" @@ -434,6 +441,16 @@ def test_notifier_wake_exposes_terminal_event_identity( 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": { @@ -467,6 +484,19 @@ def test_notifier_wake_metadata_is_ordered_and_privacy_bounded( 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, @@ -491,6 +521,122 @@ def test_notifier_wake_metadata_is_ordered_and_privacy_bounded( 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: diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index 8b42c5f41f8f..bc3f903ca6cc 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/website/docs/user-guide/features/kanban.md @@ -1034,14 +1034,25 @@ Duplicate delivery across gateways is prevented by the atomic per-event claim in the board DB. No relays, credential sharing, or extra dispatchers are needed — each profile gateway simply delivers through its own adapters. -Push-capable platform adapters receive Kanban terminal wakes as internal -`MessageEvent` objects. In addition to the localized wake text, adapters can -identify these events through `event.metadata["hermes_kanban_wake"]`, a -versioned envelope containing only `board`, `task_id`, deterministically ordered -`event_kinds`, and the claimed event `cursor`. Delivery metadata, event payloads, -task content, and profile details are intentionally excluded. Stateless API -self-post wakes keep their existing request shape and do not receive this -adapter-only metadata. +Push-capable platform adapters receive structured identity on both Kanban +terminal delivery planes. Each direct, user-visible `send()` merges the +subscription's existing chat/thread routing fields with a separate +`metadata["hermes_kanban_notification"]` envelope. Version 1 contains only +`board`, `task_id`, that notification's `event_kind`, and the claimed batch +`cursor`. The later internal `MessageEvent` wake carries +`event.metadata["hermes_kanban_wake"]`; version 1 contains only `board`, +`task_id`, deterministically ordered aggregate `event_kinds`, and the same +claimed `cursor`. + +These names are intentionally distinct: a direct notification represents one +visible event, while a synthetic wake can aggregate several claimed events. +Both envelopes are privacy allowlists. Task content, summaries/results/reasons, +arbitrary event payloads, assignee/profile names, paths, credentials, secrets, +and subscription delivery metadata are excluded from the envelopes. Existing +routing metadata remains alongside (not inside) the notification envelope and +is not mutated. Localized visible text and ordinary adapter sends remain +unchanged. Stateless API self-post wakes keep their existing request shape and +receive neither adapter-only envelope. ## Runs — one row per attempt