diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 17b65e1110f5b..ce5b9be467ce9 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -419,6 +419,9 @@ def _collect(): sub["task_id"], sub["platform"], sub["chat_id"], sub.get("thread_id") or "", ) + mode = sub.get("delivery_mode") or "notify" + wake_agent = mode in ("notify+wake", "wake") + send_passive = mode != "wake" for ev in d["events"]: kind = ev.kind # Identity prefix: attribute terminal pings to the @@ -513,6 +516,7 @@ def _collect(): if isinstance(delivery_metadata, dict) else {} ) + if sub.get("thread_id") and not metadata.get("thread_id"): metadata["thread_id"] = sub["thread_id"] # Adapters with no push channel (the API server — @@ -530,7 +534,7 @@ def _collect(): # creator is woken via the self-post below instead. from gateway.wake import adapter_supports_push - if not adapter_supports_push(adapter): + if not adapter_supports_push(adapter) and wake_agent: logger.debug( "kanban notifier: adapter %s has no push " "channel; skipping text ping for %s, relying " @@ -542,6 +546,12 @@ def _collect(): # so the counter is resolved (reset or bumped) by # the self-post outcome, not by skipping the send. continue + if not send_passive: + # Wake-only subscriptions intentionally skip the + # visible platform message. The retained wake path + # below is the sole delivery. + sub_fail_counts.pop(sub_key, None) + continue try: _send_res = await adapter.send( sub["chat_id"], msg, metadata=metadata, @@ -635,7 +645,11 @@ def _collect(): # next tick retries. task_terminal = task and task.status in {"done", "archived"} _WAKE_KINDS = ("completed", "gave_up", "crashed", "timed_out", "blocked") - _wake_kinds = {ev.kind for ev in d["events"] if ev.kind in _WAKE_KINDS} + _wake_kinds = ( + {ev.kind for ev in d["events"] if ev.kind in _WAKE_KINDS} + if wake_agent + else set() + ) from gateway.wake import adapter_supports_push as _adapter_push_ok _is_push_adapter = _adapter_push_ok(adapter) @@ -643,7 +657,7 @@ def _collect(): _synth = "" if _wake_kinds: _session_key = getattr(task, "session_id", None) or "" - if _wake_kinds and _session_key: + if _wake_kinds: _title = (task.title if task else sub["task_id"])[:120] _assignee = task.assignee if task else "" _parts = [] @@ -726,7 +740,7 @@ def _collect(): # dispatcher respawns the task and it cycles into the # same state. See the longer comment on TERMINAL_KINDS # above for the failure mode this prevents. - if _is_push_adapter and _wake_kinds and _session_key: + if _is_push_adapter and _wake_kinds: try: from gateway.session import SessionSource from gateway.wake import deliver_wake @@ -758,6 +772,7 @@ def _collect(): chat_type=_chat_type, thread_id=sub.get("thread_id") or None, user_id=sub.get("user_id"), + user_id_alt=sub.get("user_id_alt"), profile=sub_profile or None, ) # deliver_wake preserves the synthetic diff --git a/gateway/run.py b/gateway/run.py index c3aae82cf53af..85c181723954a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -21891,6 +21891,7 @@ def _set_session_env(self, context: SessionContext) -> list: chat_name=context.source.chat_name or "", thread_id=str(context.source.thread_id) if context.source.thread_id else "", user_id=str(context.source.user_id) if context.source.user_id else "", + user_id_alt=str(context.source.user_id_alt) if context.source.user_id_alt else "", user_name=str(context.source.user_name) if context.source.user_name else "", session_key=context.session_key, message_id=str(context.source.message_id) if context.source.message_id else "", diff --git a/gateway/session_context.py b/gateway/session_context.py index 0cf35ef229d1b..22b3eb4349bb1 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -78,6 +78,7 @@ def session_context_engaged() -> bool: _SESSION_CHAT_NAME: ContextVar = ContextVar("HERMES_SESSION_CHAT_NAME", default=_UNSET) _SESSION_THREAD_ID: ContextVar = ContextVar("HERMES_SESSION_THREAD_ID", default=_UNSET) _SESSION_USER_ID: ContextVar = ContextVar("HERMES_SESSION_USER_ID", default=_UNSET) +_SESSION_USER_ID_ALT: ContextVar = ContextVar("HERMES_SESSION_USER_ID_ALT", default=_UNSET) _SESSION_USER_NAME: ContextVar = ContextVar("HERMES_SESSION_USER_NAME", default=_UNSET) _SESSION_KEY: ContextVar = ContextVar("HERMES_SESSION_KEY", default=_UNSET) _SESSION_ID: ContextVar = ContextVar("HERMES_SESSION_ID", default=_UNSET) @@ -135,6 +136,7 @@ def session_context_engaged() -> bool: "HERMES_SESSION_CHAT_NAME": _SESSION_CHAT_NAME, "HERMES_SESSION_THREAD_ID": _SESSION_THREAD_ID, "HERMES_SESSION_USER_ID": _SESSION_USER_ID, + "HERMES_SESSION_USER_ID_ALT": _SESSION_USER_ID_ALT, "HERMES_SESSION_USER_NAME": _SESSION_USER_NAME, "HERMES_SESSION_KEY": _SESSION_KEY, "HERMES_SESSION_ID": _SESSION_ID, @@ -211,6 +213,7 @@ def set_session_vars( chat_name: str = "", thread_id: str = "", user_id: str = "", + user_id_alt: str = "", user_name: str = "", session_key: str = "", session_id: str = "", @@ -253,6 +256,7 @@ def set_session_vars( _SESSION_CHAT_NAME.set(chat_name), _SESSION_THREAD_ID.set(thread_id), _SESSION_USER_ID.set(user_id), + _SESSION_USER_ID_ALT.set(user_id_alt), _SESSION_USER_NAME.set(user_name), _SESSION_KEY.set(session_key), _SESSION_ID.set(session_id), @@ -290,6 +294,7 @@ def clear_session_vars(tokens: list) -> None: _SESSION_CHAT_NAME, _SESSION_THREAD_ID, _SESSION_USER_ID, + _SESSION_USER_ID_ALT, _SESSION_USER_NAME, _SESSION_KEY, _SESSION_ID, diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index fe86f0ad7dd92..828a6c671194c 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -499,6 +499,11 @@ async def _handle_kanban_command(self, event: MessageEvent) -> str: chat_type = str(getattr(source, "chat_type", "") or "") or None thread_id = str(getattr(source, "thread_id", "") or "") user_id = str(getattr(source, "user_id", "") or "") or None + # Persist the platform-specific stable alt id (Signal UUID, + # Feishu union_id) too: build_session_key keys the participant + # on ``user_id_alt or user_id``, so a replayed wake only rebuilds + # the same session key when the alt id survives the round-trip. + user_id_alt = str(getattr(source, "user_id_alt", "") or "") or None delivery_metadata = self._thread_metadata_for_source( source, self._reply_anchor_for_event(event) ) or None @@ -517,7 +522,11 @@ def _sub(): chat_type=chat_type, thread_id=thread_id or None, user_id=user_id, + user_id_alt=user_id_alt, notifier_profile=getattr(self, "_kanban_notifier_profile", None) or self._active_profile_name(), + # Subscribing from chat: deliver the passive + # message and wake the destination agent. + delivery_mode="notify+wake", delivery_metadata=delivery_metadata, ) finally: diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index a08cb8f9b4076..6659a59ebadd8 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -770,13 +770,33 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu p_nsub.add_argument("task_id") p_nsub.add_argument("--platform", required=True) p_nsub.add_argument("--chat-id", required=True) - p_nsub.add_argument("--chat-type", default="", help="dm / group / channel (used by wake routing)") p_nsub.add_argument("--thread-id", default=None) p_nsub.add_argument("--user-id", default=None) + p_nsub.add_argument("--user-id-alt", default=None) + p_nsub.add_argument( + "--chat-type", + choices=("dm", "group", "channel", "thread"), + default=None, + help="Originating source chat_type, recorded so the active-wake " + "delivery modes resolve the operator's real session. Omit to " + "leave an existing sub unchanged (new subs default to 'dm').", + ) p_nsub.add_argument( "--notifier-profile", default=None, help="Profile gateway that owns/delivers this subscription (default: active profile)", ) + p_nsub.add_argument( + "--delivery-mode", + # Single source of truth shared with the DB/watcher enum. + choices=kb._NOTIFY_DELIVERY_MODES, + default=None, + help="How the kanban-notifier reacts to terminal events for this " + "subscription: 'notify' (passive message only; default), " + "'notify+wake' (message AND wake the destination gateway agent so " + "it reads the full board context and replies in its own voice), or " + "'wake' (wake the agent only, no passive message). Omit to leave an " + "existing subscription's mode unchanged (new subs default to 'notify').", + ) p_nlist = sub.add_parser( "notify-list", @@ -2760,7 +2780,9 @@ def _cmd_notify_subscribe(args: argparse.Namespace) -> int: platform=args.platform, chat_id=args.chat_id, chat_type=args.chat_type, thread_id=args.thread_id, user_id=args.user_id, + user_id_alt=getattr(args, "user_id_alt", None), notifier_profile=args.notifier_profile or _profile_author(), + delivery_mode=getattr(args, "delivery_mode", None), ) print(f"Subscribed {args.platform}:{args.chat_id}" + (f":{args.thread_id}" if args.thread_id else "") @@ -2780,8 +2802,13 @@ def _cmd_notify_list(args: argparse.Namespace) -> int: for s in subs: thr = f":{s['thread_id']}" if s.get("thread_id") else "" owner = f" owner={s['notifier_profile']}" if s.get("notifier_profile") else "" + dmode = s.get("delivery_mode") or "notify" + mode = "" if dmode == "notify" else f" mode={dmode}" + ctype = s.get("chat_type") or "dm" + ct = "" if ctype == "dm" else f" chat_type={ctype}" + uid_alt = f" user_id_alt={s['user_id_alt']}" if s.get("user_id_alt") else "" print(f" {s['task_id']:10s} {s['platform']}:{s['chat_id']}{thr}" - f" (since event {s['last_event_id']}){owner}") + f" (since event {s['last_event_id']}){owner}{ct}{uid_alt}{mode}") return 0 diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index c4bb7caf9482b..57e3489dcf4fd 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -1354,10 +1354,12 @@ class Event: task_id TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL, - chat_type TEXT, thread_id TEXT NOT NULL DEFAULT '', user_id TEXT, + user_id_alt TEXT, + chat_type TEXT, 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, @@ -2514,9 +2516,32 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: _add_column_if_missing( conn, "kanban_notify_subs", "notifier_profile", "notifier_profile TEXT" ) + if "delivery_mode" not in notify_cols: + _add_column_if_missing( + conn, + "kanban_notify_subs", + "delivery_mode", + "delivery_mode TEXT NOT NULL DEFAULT 'notify'", + ) if "chat_type" not in notify_cols: _add_column_if_missing( - conn, "kanban_notify_subs", "chat_type", "chat_type TEXT" + conn, + "kanban_notify_subs", + "chat_type", + "chat_type TEXT", + ) + if "user_id_alt" not in notify_cols: + # Records the originating source's platform-specific stable alt ID + # (Signal UUID, Feishu union_id, ...) alongside ``user_id`` so an + # active-wake replay reconstructs the SAME ``build_session_key`` as + # the original event. ``build_session_key`` prefers ``user_id_alt`` + # over ``user_id`` when both are present (gateway/session.py); a + # wake that only replayed ``user_id`` would key to a different, + # context-less session whenever the two diverge. Legacy rows + # default to NULL, which is inert: ``user_id_alt or user_id`` falls + # back to the already-persisted ``user_id``. + _add_column_if_missing( + conn, "kanban_notify_subs", "user_id_alt", "user_id_alt TEXT" ) if "delivery_metadata" not in notify_cols: _add_column_if_missing( @@ -2644,8 +2669,10 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: "kanban_notify_subs": ( "CREATE TABLE kanban_notify_subs (" " task_id TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL," - " chat_type TEXT, thread_id TEXT NOT NULL DEFAULT '', user_id TEXT," - " notifier_profile TEXT, delivery_metadata TEXT, created_at INTEGER NOT NULL," + " thread_id TEXT NOT NULL DEFAULT '', user_id TEXT, user_id_alt TEXT," + " chat_type TEXT," + " 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," " PRIMARY KEY (task_id, platform, chat_id, thread_id))", ("CREATE INDEX idx_notify_task ON kanban_notify_subs(task_id)",), @@ -3251,6 +3278,47 @@ def create_task( "INSERT OR IGNORE INTO task_links (parent_id, child_id) VALUES (?, ?)", (pid, task_id), ) + if parents: + # ACK-edge inheritance: a child inherits the parent/root + # task's terminal-notification return path (chat_type and + # delivery_mode included), so the originating channel still + # hears about a child that BLOCKs, not just the final fan-in. + # The child task_id is brand-new, so INSERT OR IGNORE copies + # each sub. + placeholders = ",".join("?" * len(parents)) + parent_subs = conn.execute( + "SELECT * FROM kanban_notify_subs " + f"WHERE task_id IN ({placeholders}) " + "ORDER BY created_at ASC", + parents, + ).fetchall() + for psub in parent_subs: + # Inherit chat_type and delivery_mode so a woken child + # notification keys to the parent's channel. + psub_mode = psub["delivery_mode"] or "notify" + psub_chat_type = psub["chat_type"] or "dm" + conn.execute( + """ + INSERT OR IGNORE INTO kanban_notify_subs + (task_id, platform, chat_id, thread_id, user_id, user_id_alt, + chat_type, notifier_profile, delivery_mode, + delivery_metadata, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + task_id, + psub["platform"], + psub["chat_id"], + psub["thread_id"] or "", + psub["user_id"], + psub["user_id_alt"], + psub_chat_type, + psub["notifier_profile"], + psub_mode, + psub["delivery_metadata"], + now, + ), + ) _append_event( conn, task_id, @@ -9680,6 +9748,14 @@ def task_age(task: Task) -> dict: # Notification subscriptions (used by the gateway kanban-notifier) # --------------------------------------------------------------------------- +# How the gateway kanban-notifier reacts to a terminal event for a +# subscription: +# "notify" -> passive ``adapter.send`` only (default) +# "notify+wake" -> passive send AND wake the destination gateway agent +# "wake" -> wake the agent only; no passive message is sent +_NOTIFY_DELIVERY_MODES = ("notify", "notify+wake", "wake") + + def _encode_notify_delivery_metadata( metadata: Optional[Mapping[str, Any]], ) -> Optional[str]: @@ -9721,15 +9797,34 @@ def add_notify_sub( task_id: str, platform: str, chat_id: str, - chat_type: Optional[str] = None, thread_id: Optional[str] = None, user_id: Optional[str] = None, + user_id_alt: Optional[str] = None, + chat_type: Optional[str] = None, notifier_profile: Optional[str] = None, + delivery_mode: Optional[str] = None, 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). + ``user_id_alt`` records the originating source's platform-specific stable + alt ID (Signal UUID, Feishu union_id, ...) alongside ``user_id``. Active-wake + replay must reproduce it so the woken turn's ``build_session_key`` matches + the original event's — ``build_session_key`` prefers ``user_id_alt`` over + ``user_id`` (gateway/session.py), so replaying only ``user_id`` would key a + wake into a different session whenever the two diverge for this source. + + ``chat_type`` records the originating source's chat type; the active-wake + delivery modes replay it so the woken turn resolves the operator's real + channel. ``None`` keeps an existing row's value. + + ``delivery_mode`` (see ``_NOTIFY_DELIVERY_MODES``) selects how the + kanban-notifier reacts to a terminal event for this subscription. ``None`` + leaves an existing row's mode untouched (and inserts the ``"notify"`` + default for a fresh row); an explicit value is last-write-wins, so an + operator can intentionally re-subscribe to change the mode (e.g. + ``notify`` -> ``wake``). An unknown value falls back to ``"notify"``. 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 @@ -9739,44 +9834,58 @@ def add_notify_sub( AFTER they subscribe; the gateway/tool auto-subscribe paths run at task creation, where the snapshot is 0 anyway. """ + insert_mode = delivery_mode if delivery_mode in _NOTIFY_DELIVERY_MODES else "notify" + insert_chat_type = chat_type or "dm" now = int(time.time()) metadata_json = _encode_notify_delivery_metadata(delivery_metadata) with write_txn(conn): conn.execute( """ INSERT OR IGNORE INTO kanban_notify_subs - (task_id, platform, chat_id, chat_type, thread_id, user_id, - notifier_profile, delivery_metadata, created_at, last_event_id) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, + (task_id, platform, chat_id, thread_id, user_id, user_id_alt, + chat_type, notifier_profile, delivery_mode, delivery_metadata, + created_at, last_event_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, COALESCE((SELECT MAX(id) FROM task_events WHERE task_id = ?), 0)) """, ( task_id, platform, chat_id, - chat_type, thread_id or "", user_id, + user_id_alt, + insert_chat_type, notifier_profile, + insert_mode, metadata_json, now, task_id, ), ) if chat_type: - # Self-heal rows created before chat_type was persisted. + # Explicit chat_type is last-write-wins on re-subscribe. conn.execute( """ UPDATE kanban_notify_subs SET chat_type = ? WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ? - AND (chat_type IS NULL OR chat_type = '') """, (chat_type, task_id, platform, chat_id, thread_id or ""), ) + if user_id_alt: + # Self-heal legacy rows created before alternate IDs were tracked. + conn.execute( + """ + UPDATE kanban_notify_subs + SET user_id_alt = ? + WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ? + AND (user_id_alt IS NULL OR user_id_alt = '') + """, + (user_id_alt, task_id, platform, chat_id, thread_id or ""), + ) if notifier_profile: - # Self-heal legacy rows that predate notifier ownership by - # backfilling only when the existing value is unset. + # Self-heal legacy rows that predate notifier ownership. conn.execute( """ UPDATE kanban_notify_subs @@ -9786,10 +9895,18 @@ def add_notify_sub( """, (notifier_profile, task_id, platform, chat_id, thread_id or ""), ) + if delivery_mode in _NOTIFY_DELIVERY_MODES: + # Explicit delivery_mode is last-write-wins on re-subscribe. + conn.execute( + """ + UPDATE kanban_notify_subs + SET delivery_mode = ? + WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ? + """, + (delivery_mode, task_id, platform, chat_id, thread_id or ""), + ) if metadata_json: - # A duplicate subscribe from the same chat/thread should refresh - # the routing anchor. Telegram DM-topic notifications need the - # latest reply anchor to stay inside the visible topic lane. + # Refresh the routing anchor for duplicate subscriptions. conn.execute( """ UPDATE kanban_notify_subs diff --git a/tests/gateway/test_kanban_notifier.py b/tests/gateway/test_kanban_notifier.py index 7400efd7263d9..33273bbfbd650 100644 --- a/tests/gateway/test_kanban_notifier.py +++ b/tests/gateway/test_kanban_notifier.py @@ -100,6 +100,7 @@ def test_kanban_notifier_replays_telegram_dm_topic_delivery_metadata(tmp_path, m platform="telegram", chat_id="chat-1", thread_id="20197", + delivery_mode="notify+wake", delivery_metadata={ "chat_type": "dm", "direct_messages_topic_id": "20197", @@ -372,6 +373,7 @@ def test_notifier_wakeup_uses_subscription_chat_type(tmp_path, monkeypatch): platform="telegram", chat_id="chat-dm", chat_type="dm", + delivery_mode="notify+wake", ) kb.complete_task(conn, tid, summary="done") finally: diff --git a/tests/hermes_cli/test_kanban_notify.py b/tests/hermes_cli/test_kanban_notify.py index 69ff262b4f48c..8158fcc3dafb9 100644 --- a/tests/hermes_cli/test_kanban_notify.py +++ b/tests/hermes_cli/test_kanban_notify.py @@ -35,6 +35,529 @@ def _assert_inherited_notify_sub(subs: list[dict]) -> None: assert subs[0]["notifier_profile"] == "default" +def test_notify_sub_delivery_mode_persists_and_last_write_wins(kanban_home): + """delivery_mode persists; an explicit re-subscribe is last-write-wins, a + ``None`` re-subscribe leaves the existing mode untouched, an unknown value + is ignored, and none of this clobbers the notifier_profile owner.""" + import hermes_cli.kanban_db as kb + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="mode sub task", assignee="worker1") + # Fresh sub without a mode -> defaults to "notify". + kb.add_notify_sub( + conn, task_id=tid, platform="telegram", chat_id="chat1", + notifier_profile="owner-a", + ) + subs = kb.list_notify_subs(conn, tid) + assert len(subs) == 1 + assert subs[0]["delivery_mode"] == "notify" + assert subs[0]["notifier_profile"] == "owner-a" + + # Explicit re-subscribe changes the mode (last-write-wins) and must NOT + # overwrite the existing owner (owner self-heals only when unset). + kb.add_notify_sub( + conn, task_id=tid, platform="telegram", chat_id="chat1", + notifier_profile="owner-b", delivery_mode="wake", + ) + subs = kb.list_notify_subs(conn, tid) + assert len(subs) == 1 + assert subs[0]["delivery_mode"] == "wake" + assert subs[0]["notifier_profile"] == "owner-a" + + # A None re-subscribe leaves the existing mode untouched. + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat1") + subs = kb.list_notify_subs(conn, tid) + assert subs[0]["delivery_mode"] == "wake" + + # An unknown mode is ignored (treated like None: no clobber). + kb.add_notify_sub( + conn, task_id=tid, platform="telegram", chat_id="chat1", + delivery_mode="bogus", + ) + subs = kb.list_notify_subs(conn, tid) + assert subs[0]["delivery_mode"] == "wake" + finally: + conn.close() + + +def test_child_task_inherits_parent_delivery_mode(kanban_home): + """Graph children inherit the parent's ACK edge AND its delivery_mode.""" + import hermes_cli.kanban_db as kb + + conn = kb.connect() + try: + parent = kb.create_task(conn, title="root", assignee=None) + kb.add_notify_sub( + conn, task_id=parent, platform="telegram", chat_id="chat1", + thread_id="42", user_id="u1", user_id_alt="alt-u1", notifier_profile="default", + delivery_mode="notify+wake", + ) + child = kb.create_task( + conn, title="review child", assignee="ccreviewer", parents=[parent], + ) + subs = kb.list_notify_subs(conn, child) + finally: + conn.close() + + assert len(subs) == 1 + assert subs[0]["platform"] == "telegram" + assert subs[0]["chat_id"] == "chat1" + assert subs[0]["thread_id"] == "42" + assert subs[0]["user_id"] == "u1" + assert subs[0]["user_id_alt"] == "alt-u1" + assert subs[0]["notifier_profile"] == "default" + assert subs[0]["delivery_mode"] == "notify+wake" + + +def test_notify_sub_chat_type_persists_and_last_write_wins(kanban_home): + """chat_type persists, defaults to 'dm', an explicit re-subscribe is + last-write-wins, and a None re-subscribe leaves it untouched. The + active-wake path replays this field so the woken turn keys to the + operator's real channel.""" + import hermes_cli.kanban_db as kb + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="chat_type sub", assignee="worker1") + # Fresh sub without chat_type -> defaults to "dm". + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat1") + subs = kb.list_notify_subs(conn, tid) + assert subs[0]["chat_type"] == "dm" + + # Explicit re-subscribe corrects the recorded chat_type (last-write-wins). + kb.add_notify_sub( + conn, task_id=tid, platform="telegram", chat_id="chat1", + chat_type="group", + ) + subs = kb.list_notify_subs(conn, tid) + assert subs[0]["chat_type"] == "group" + + # A None re-subscribe (here changing only the mode) must NOT clobber + # the recorded chat_type. + kb.add_notify_sub( + conn, task_id=tid, platform="telegram", chat_id="chat1", + delivery_mode="wake", + ) + subs = kb.list_notify_subs(conn, tid) + assert subs[0]["chat_type"] == "group" + assert subs[0]["delivery_mode"] == "wake" + finally: + conn.close() + + +def test_notify_sub_user_id_alt_persists_and_backfills_legacy_rows(kanban_home): + """user_id_alt is persisted with the notify subscription routing tuple and + can backfill a pre-existing row created before the alt id was known.""" + import hermes_cli.kanban_db as kb + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="alt sub", assignee="worker1") + kb.add_notify_sub( + conn, task_id=tid, platform="telegram", chat_id="chat1", + user_id="open-id", + ) + subs = kb.list_notify_subs(conn, tid) + assert subs[0]["user_id"] == "open-id" + assert subs[0]["user_id_alt"] is None + + kb.add_notify_sub( + conn, task_id=tid, platform="telegram", chat_id="chat1", + user_id="open-id", user_id_alt="union-id", + ) + subs = kb.list_notify_subs(conn, tid) + finally: + conn.close() + + assert subs[0]["user_id"] == "open-id" + assert subs[0]["user_id_alt"] == "union-id" + + +def test_child_task_inherits_parent_chat_type(kanban_home): + """Graph children inherit the parent's chat_type alongside its ACK edge and + delivery_mode, so a woken child notification keys to the same session as + the parent's originating channel.""" + import hermes_cli.kanban_db as kb + + conn = kb.connect() + try: + parent = kb.create_task(conn, title="root", assignee=None) + kb.add_notify_sub( + conn, task_id=parent, platform="telegram", chat_id="chat1", + user_id="u1", user_id_alt="alt-u1", chat_type="group", + delivery_mode="notify+wake", + ) + child = kb.create_task( + conn, title="impl child", assignee="coder", parents=[parent], + ) + subs = kb.list_notify_subs(conn, child) + finally: + conn.close() + + assert len(subs) == 1 + assert subs[0]["chat_type"] == "group" + assert subs[0]["delivery_mode"] == "notify+wake" + assert subs[0]["user_id"] == "u1" + assert subs[0]["user_id_alt"] == "alt-u1" + + +@pytest.mark.asyncio +async def test_notifier_notify_plus_wake_sends_and_wakes(kanban_home): + """notify+wake delivers the passive message AND wakes the agent; a plain + notify sub only sends. The agent is woken only for the notify+wake sub.""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + passive_tid = kb.create_task(conn, title="passive task", assignee="worker1") + active_tid = kb.create_task(conn, title="active task", assignee="worker1") + kb.add_notify_sub(conn, task_id=passive_tid, platform="telegram", chat_id="chat1") + kb.add_notify_sub( + conn, task_id=active_tid, platform="telegram", chat_id="chat1", + delivery_mode="notify+wake", + ) + kb.block_task(conn, passive_tid, reason="passive block") + kb.block_task(conn, active_tid, reason="active block") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._owns_kanban_dispatcher_lock = lambda: True + runner._running = True + runner._kanban_sub_fail_counts = {} + + fake_adapter = MagicMock() + sent_msgs: list[str] = [] + + async def _send(chat_id, msg, metadata=None): + sent_msgs.append(msg) + + fake_adapter.send = AsyncMock(side_effect=_send) + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + tick_count = 0 + + async def _fast_sleep(_): + nonlocal tick_count + await _orig_sleep(0) + tick_count += 1 + if tick_count >= 3: + runner._running = False + + wake_mock = AsyncMock() + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep), \ + patch("gateway.wake.deliver_wake", new=wake_mock): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + # Both subs still get a passive send (notify AND notify+wake send). + assert len(sent_msgs) == 2 + assert any("passive block" in m for m in sent_msgs) + assert any("active block" in m for m in sent_msgs) + # Only the notify+wake sub woke the agent, exactly once. + wake_mock.assert_awaited_once() + assert active_tid in wake_mock.await_args.kwargs["text"] + + +@pytest.mark.asyncio +async def test_notifier_plain_notify_never_wakes_even_with_session_id(kanban_home): + """Plain/default notify must remain passive even when the task carries a + creator session_id. This guards against the older unconditional wake path + that forged adapter.handle_message events after every terminal delivery.""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task( + conn, + title="legacy passive task", + assignee="worker1", + session_id="origin-session-id", + ) + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat1") + kb.block_task(conn, tid, reason="plain notify block") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._owns_kanban_dispatcher_lock = lambda: True + runner._running = True + runner._kanban_sub_fail_counts = {} + fake_adapter = MagicMock() + fake_adapter.send = AsyncMock() + fake_adapter.handle_message = AsyncMock() + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + tick_count = 0 + + async def _fast_sleep(_): + nonlocal tick_count + await _orig_sleep(0) + tick_count += 1 + if tick_count >= 3: + runner._running = False + + wake_mock = AsyncMock() + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep), \ + patch("gateway.wake.deliver_wake", new=wake_mock): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send.assert_awaited_once() + wake_mock.assert_not_awaited() + fake_adapter.handle_message.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_notifier_notify_wake_does_not_wake_on_status_event(kanban_home): + """notify+wake wakes on terminal outcomes, not on dashboard status churn.""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="status-only task", assignee="worker1") + kb.add_notify_sub( + conn, task_id=tid, platform="telegram", chat_id="chat1", + delivery_mode="notify+wake", + ) + kb._append_event(conn, tid, kind="status", payload={"status": "review"}) + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._owns_kanban_dispatcher_lock = lambda: True + runner._running = True + runner._kanban_sub_fail_counts = {} + fake_adapter = MagicMock() + fake_adapter.send = AsyncMock() + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + tick_count = 0 + + async def _fast_sleep(_): + nonlocal tick_count + await _orig_sleep(0) + tick_count += 1 + if tick_count >= 3: + runner._running = False + + wake_mock = AsyncMock() + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep), \ + patch("gateway.wake.deliver_wake", new=wake_mock): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + fake_adapter.send.assert_awaited_once() + wake_mock.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_notifier_wake_forwards_persisted_chat_type_and_user_id(kanban_home): + """The active-wake call must carry the subscription's persisted chat_type and + user_id so ``deliver_wake`` resolves the operator's real (e.g. group) + session instead of a hardcoded one.""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="group wake", assignee="worker1") + kb.add_notify_sub( + conn, task_id=tid, platform="telegram", chat_id="grp1", + user_id="op-42", chat_type="group", delivery_mode="wake", + notifier_profile="owner-profile", + ) + kb.block_task(conn, tid, reason="group block") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._owns_kanban_dispatcher_lock = lambda: True + runner._running = True + runner._kanban_sub_fail_counts = {} + fake_adapter = MagicMock() + fake_adapter.send = AsyncMock() + runner.adapters = {Platform.TELEGRAM: fake_adapter} + runner._profile_adapters = {"owner-profile": {Platform.TELEGRAM: fake_adapter}} + runner._authorization_adapter = lambda platform, profile=None: fake_adapter + + _orig_sleep = asyncio.sleep + tick_count = 0 + + async def _fast_sleep(_): + nonlocal tick_count + await _orig_sleep(0) + tick_count += 1 + if tick_count >= 3: + runner._running = False + + wake_mock = AsyncMock() + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep), \ + patch("gateway.wake.deliver_wake", new=wake_mock): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + wake_mock.assert_awaited_once() + source = wake_mock.await_args.kwargs["source"] + assert source.chat_type == "group" + assert source.user_id == "op-42" + assert source.profile == "owner-profile" + + +@pytest.mark.asyncio +async def test_notifier_wake_only_skips_send_and_advances_cursor(kanban_home): + """wake-only: NO passive send, the agent is woken exactly once, and the + cursor advances so repeated ticks do not re-wake.""" + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + try: + tid = kb.create_task(conn, title="wake only task", assignee="worker1") + kb.add_notify_sub( + conn, task_id=tid, platform="telegram", chat_id="chat1", + delivery_mode="wake", + ) + kb.block_task(conn, tid, reason="wake only block") + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._owns_kanban_dispatcher_lock = lambda: True + runner._running = True + runner._kanban_sub_fail_counts = {} + + fake_adapter = MagicMock() + fake_adapter.send = AsyncMock() + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + tick_count = 0 + + async def _fast_sleep(_): + nonlocal tick_count + await _orig_sleep(0) + tick_count += 1 + if tick_count >= 3: + runner._running = False + + wake_mock = AsyncMock() + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep), \ + patch("gateway.wake.deliver_wake", new=wake_mock): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + # wake-only never uses the passive transport... + fake_adapter.send.assert_not_awaited() + # ...and wakes the agent exactly once across several ticks (proves the + # cursor advanced; otherwise it would re-wake on every poll). + wake_mock.assert_awaited_once() + assert tid in wake_mock.await_args.kwargs["text"] + + # The subscription survives (blocked is non-terminal) but its cursor moved + # past the blocked event. + conn = kb.connect() + try: + subs = kb.list_notify_subs(conn, tid) + finally: + conn.close() + assert len(subs) == 1 + assert int(subs[0]["last_event_id"]) > 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize('kind', ["gave_up", "crashed", "timed_out"]) +async def test_notifier_unsubs_after_abnormal_events(kind, kanban_home): + """ + Event kinds gave_up / crashed / timed_out send a notification but DO + NOT delete the subscription. The dispatcher may respawn the task and + fire the same event kind again (e.g. a worker that crashes, gets + reclaimed, and crashes a second time); the user must hear about the + second event too. Subscriptions are removed only when the task hits + a truly final status (done / archived) — see the comment on + TERMINAL_KINDS in gateway/run.py and PR #21398. + """ + import hermes_cli.kanban_db as kb + from gateway.run import GatewayRunner + from gateway.config import Platform + + conn = kb.connect() + + try: + tid = kb.create_task(conn, title=f"test {kind} task", assignee="worker1") + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat1") + kb._append_event(conn, tid, kind=kind) + finally: + conn.close() + + runner = object.__new__(GatewayRunner) + runner._owns_kanban_dispatcher_lock = lambda: True + runner._running = True + runner._kanban_sub_fail_counts = {} + + fake_adapter = MagicMock() + + async def _send_and_stop(chat_id, msg, metadata=None): + runner._running = False + + fake_adapter.send = AsyncMock(side_effect=_send_and_stop) + runner.adapters = {Platform.TELEGRAM: fake_adapter} + + _orig_sleep = asyncio.sleep + + async def _fast_sleep(_): + await _orig_sleep(0) + + with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep): + await asyncio.wait_for( + runner._kanban_notifier_watcher(interval=1), + timeout=10.0, + ) + + # The user is notified about the abnormal event... + fake_adapter.send.assert_called_once() + assert kind.replace('_', ' ') in fake_adapter.send.call_args[0][1] + + # ...but the subscription survives so a respawn-then-same-event cycle + # reaches the user too. The cursor (last_event_id) advanced inside + # the same write txn as the claim, so the same event won't re-fire. + conn = kb.connect() + try: + subs = kb.list_notify_subs(conn, tid) + finally: + conn.close() + assert len(subs) == 1, ( + f"Subscription should survive {kind!r} so the next cycle of the " + f"same event reaches the user; got {subs!r}" + ) + assert int(subs[0]["last_event_id"]) >= 1, ( + "Cursor should have advanced past the delivered event " + "(claim_unseen_events_for_sub advances atomically inside the " + "same write txn as the read)." + ) + @@ -79,6 +602,7 @@ async def test_gateway_create_autosubscribes_on_explicit_board(kanban_home): kb.create_board("projx") runner = object.__new__(GatewayRunner) + runner._owns_kanban_dispatcher_lock = lambda: True source = SimpleNamespace( platform=Platform.TELEGRAM, chat_id="chat1", @@ -123,6 +647,97 @@ async def test_gateway_create_autosubscribes_on_explicit_board(kanban_home): conn.close() +@pytest.mark.parametrize( + "chat_type,thread_id,thread_sessions_per_user", + [ + # Isolated group/channel: no thread, so the participant id is appended + # to the key when group_sessions_per_user is on. + ("channel", None, False), + # Thread with per-user isolation on: the participant id is still + # appended, so the alt id must survive the round-trip here too. + ("group", "th-42", True), + ], +) +@pytest.mark.asyncio +async def test_gateway_autosubscribe_roundtrips_user_id_alt_for_session_key( + kanban_home, chat_type, thread_id, thread_sessions_per_user, +): + """The gateway `/kanban create` auto-subscribe must persist ``user_id_alt`` + so a replayed wake rebuilds the *same* session key as the original event. + + ``build_session_key`` keys the participant on ``user_id_alt or user_id`` + (Signal UUID / Feishu union_id carry the canonical participant in the alt + slot). If the subscription row drops ``user_id_alt``, the replayed source + falls back to ``user_id`` and lands in a different session — the woken turn + answers into a parallel conversation. Drive the real handler and compare the + key built from the original source with the key rebuilt from the persisted + row. + """ + from gateway.run import GatewayRunner + from gateway.config import Platform + from gateway.session import SessionSource, build_session_key + + runner = object.__new__(GatewayRunner) + runner._owns_kanban_dispatcher_lock = lambda: True + # user_id != user_id_alt is the whole point: the alt id is the canonical + # participant, so dropping it silently corrupts the session key. + source = SimpleNamespace( + platform=Platform.TELEGRAM, + chat_id="chat1", + chat_type=chat_type, + thread_id=thread_id, + user_id="open-id", + user_id_alt="union-id", + ) + event = SimpleNamespace( + text='/kanban create "hello" --assignee alice', + source=source, + ) + + out = await GatewayRunner._handle_kanban_command(runner, event) + assert "subscribed" in out.lower() + + conn = kb.connect() + try: + subs = kb.list_notify_subs(conn) + finally: + conn.close() + assert len(subs) == 1 + row = subs[0] + assert row["user_id"] == "open-id" + assert row["user_id_alt"] == "union-id" + + original = SessionSource( + platform=Platform.TELEGRAM, + chat_id="chat1", + chat_type=chat_type, + thread_id=thread_id, + user_id="open-id", + user_id_alt="union-id", + ) + # Reconstruct exactly as the wake path does: from the persisted row. + replayed = SessionSource( + platform=Platform(row["platform"]), + chat_id=row["chat_id"], + chat_type=row["chat_type"], + thread_id=row["thread_id"] or None, + user_id=row["user_id"], + user_id_alt=row["user_id_alt"], + ) + + original_key = build_session_key( + original, thread_sessions_per_user=thread_sessions_per_user + ) + replayed_key = build_session_key( + replayed, thread_sessions_per_user=thread_sessions_per_user + ) + assert original_key == replayed_key + # Regression guard: the canonical alt id — not the raw user_id — is what + # keys the participant. Proves the alt id actually reached the key. + assert "union-id" in replayed_key + assert "open-id" not in replayed_key + + @pytest.mark.asyncio async def test_notifier_artifact_delivery_skips_missing_files(kanban_home, tmp_path, monkeypatch): """Missing artifact paths are silently skipped — they may have been @@ -158,6 +773,7 @@ async def test_notifier_artifact_delivery_skips_missing_files(kanban_home, tmp_p os.environ.pop("HERMES_KANBAN_TASK", None) runner = object.__new__(GatewayRunner) + runner._owns_kanban_dispatcher_lock = lambda: True runner._running = True runner._kanban_sub_fail_counts = {} runner._kanban_dispatcher_lock_handle = object() diff --git a/tests/tools/test_kanban_tools.py b/tests/tools/test_kanban_tools.py index 476d14dd325cf..315c19ca2ec50 100644 --- a/tests/tools/test_kanban_tools.py +++ b/tests/tools/test_kanban_tools.py @@ -810,6 +810,89 @@ def _sub_index(subs): return out +def test_create_subscribes_gateway_session(monkeypatch, worker_env): + """A gateway session (platform + chat_id set) gets auto-subscribed + to its own kanban_create result, and the response surfaces the + ``subscribed`` flag so the orchestrator can react.""" + from tools import kanban_tools as kt + monkeypatch.setenv("HERMES_SESSION_PLATFORM", "telegram") + monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "chat-42") + monkeypatch.setenv("HERMES_SESSION_THREAD_ID", "thread-7") + monkeypatch.setenv("HERMES_SESSION_USER_ID", "user-9") + monkeypatch.setenv("HERMES_SESSION_USER_ID_ALT", "alt-user-9") + monkeypatch.setenv("HERMES_SESSION_CHAT_TYPE", "forum") + + out = kt._handle_create({ + "title": "auto-sub gateway", + "assignee": "peer", + }) + d = json.loads(out) + assert d["ok"] is True + new_tid = d["task_id"] + assert d["subscribed"] is True, d + + subs = _sub_index(_list_subs_for_task(new_tid)) + assert len(subs) == 1 + s = subs[0] + assert s["platform"] == "telegram" + assert s["chat_id"] == "chat-42" + assert s["thread_id"] == "thread-7" + assert s["user_id"] == "user-9" + assert s["user_id_alt"] == "alt-user-9" + assert s["chat_type"] == "forum" + assert s["delivery_mode"] == "notify+wake" + + +def test_create_subscribes_tui_session_via_session_key(monkeypatch, worker_env): + """TUI / desktop sessions don't have a platform/chat_id (single + local channel), but the parent process exports HERMES_SESSION_KEY. + We should still auto-subscribe, with platform='tui' and + chat_id=.""" + from tools import kanban_tools as kt + monkeypatch.delenv("HERMES_SESSION_PLATFORM", raising=False) + monkeypatch.delenv("HERMES_SESSION_CHAT_ID", raising=False) + monkeypatch.delenv("HERMES_SESSION_THREAD_ID", raising=False) + monkeypatch.delenv("HERMES_SESSION_USER_ID", raising=False) + monkeypatch.setenv("HERMES_SESSION_KEY", "tui-session-abc") + monkeypatch.delenv("HERMES_SESSION_ID", raising=False) + + out = kt._handle_create({ + "title": "auto-sub tui", + "assignee": "peer", + }) + d = json.loads(out) + assert d["ok"] is True + new_tid = d["task_id"] + assert d["subscribed"] is True, d + + subs = _sub_index(_list_subs_for_task(new_tid)) + assert len(subs) == 1 + assert subs[0]["platform"] == "tui" + assert subs[0]["chat_id"] == "tui-session-abc" + assert subs[0]["chat_type"] == "dm" + assert subs[0]["delivery_mode"] == "notify" + + +def test_create_does_not_subscribe_in_cli_session(monkeypatch, worker_env): + """CLI / cron / test sessions have no persistent delivery channel. + _maybe_auto_subscribe returns False and no row is written.""" + from tools import kanban_tools as kt + monkeypatch.delenv("HERMES_SESSION_PLATFORM", raising=False) + monkeypatch.delenv("HERMES_SESSION_CHAT_ID", raising=False) + monkeypatch.delenv("HERMES_SESSION_KEY", raising=False) + monkeypatch.delenv("HERMES_SESSION_ID", raising=False) + + out = kt._handle_create({ + "title": "no sub cli", + "assignee": "peer", + }) + d = json.loads(out) + assert d["ok"] is True + assert d["subscribed"] is False, d + + assert _list_subs_for_task(d["task_id"]) == [] + + def test_create_respects_auto_subscribe_on_create_false(monkeypatch, worker_env, tmp_path): """The config gate kanban.auto_subscribe_on_create=false must suppress auto-subscription even when the session has a delivery diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index fdf846d7353a4..3a9e91c3f5f8a 100644 --- a/tools/kanban_tools.py +++ b/tools/kanban_tools.py @@ -1425,9 +1425,12 @@ def _maybe_auto_subscribe(conn: Any, task_id: str) -> bool: return False # CLI / cron / test — no persistent channel platform = "tui" chat_id = session_key + is_gateway_session = platform != "tui" + chat_type = get_session_env("HERMES_SESSION_CHAT_TYPE", "") or None + delivery_mode = "notify+wake" if is_gateway_session else None thread_id = get_session_env("HERMES_SESSION_THREAD_ID", "") or None user_id = get_session_env("HERMES_SESSION_USER_ID", "") or None - chat_type = get_session_env("HERMES_SESSION_CHAT_TYPE", "") or None + user_id_alt = get_session_env("HERMES_SESSION_USER_ID_ALT", "") or None message_id = get_session_env("HERMES_SESSION_MESSAGE_ID", "") or "" notifier_profile = ( get_session_env("HERMES_SESSION_PROFILE", "") @@ -1460,9 +1463,10 @@ def _maybe_auto_subscribe(conn: Any, task_id: str) -> bool: _kb.add_notify_sub( conn, task_id=task_id, platform=platform, chat_id=chat_id, + thread_id=thread_id, user_id=user_id, user_id_alt=user_id_alt, chat_type=chat_type, - thread_id=thread_id, user_id=user_id, notifier_profile=notifier_profile, + delivery_mode=delivery_mode, delivery_metadata=delivery_metadata or None, ) return True diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index 8e5859a3bed44..d7969723dbc76 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/website/docs/user-guide/features/kanban.md @@ -737,6 +737,7 @@ hermes kanban stats [--json] # per-status + per-assign hermes kanban log [--tail BYTES] # worker log from ~/.hermes/kanban/logs/ hermes kanban notify-subscribe # gateway bridge hook (used by /kanban in the gateway) --platform --chat-id [--thread-id ] [--user-id ] + [--chat-type dm|group|channel|thread] [--delivery-mode notify|notify+wake|wake] hermes kanban notify-list [] [--json] hermes kanban notify-unsubscribe --platform --chat-id [--thread-id ] @@ -853,6 +854,8 @@ bot> ✓ t_9fc1a3 completed by transcriber Subscriptions auto-remove themselves once the task reaches `done` or `archived`. If you script a create with `--json` (machine output) the auto-subscribe is skipped — the assumption is that scripted callers want to manage subscriptions explicitly via `/kanban notify-subscribe`. +A chat-originated auto-subscribe is created in `notify+wake` mode: on a terminal event the destination agent both receives the passive message **and** takes a real turn, so it can read the board context and reply in its own voice. See [Delivery modes](#delivery-modes) below. + ### Output truncation in messaging Gateway platforms have practical message-length caps. If `/kanban list`, `/kanban show`, or `/kanban tail` produce more than ~3800 characters of output, the response is truncated with a `… (truncated; use \`hermes kanban …\` in your terminal for full output)` footer. The CLI surface has no such cap. @@ -900,7 +903,8 @@ You can manage subscriptions explicitly from the CLI — useful when a script / ```bash hermes kanban notify-subscribe t_abcd \ - --platform telegram --chat-id 12345678 --thread-id 7 + --platform telegram --chat-id 12345678 --thread-id 7 \ + --chat-type group --delivery-mode notify+wake hermes kanban notify-list hermes kanban notify-unsubscribe t_abcd \ --platform telegram --chat-id 12345678 --thread-id 7 @@ -908,6 +912,20 @@ hermes kanban notify-unsubscribe t_abcd \ A subscription removes itself automatically once the task reaches `done` or `archived`; no cleanup needed. +### Delivery modes + +`--delivery-mode` controls **how** the notifier reacts to a terminal event. Every subscription is in one of three modes (`notify` is the default and the original behavior): + +| Mode | Passive message | Wakes the agent | Use it when | +|------|-----------------|-----------------|-------------| +| `notify` | yes | no | You just want a heads-up message in the chat (default). | +| `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. | + +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. + +`--chat-type` (`dm` | `group` | `channel` | `thread`) records the originating chat's type so a woken turn resolves the operator's **real** session: `build_session_key` keys groups, channels, and threads differently from DMs, so an inaccurate `chat_type` would route the wake into a separate, context-less session. The `/kanban` auto-subscribe and slash-command paths capture this automatically — you only set it by hand when subscribing a chat from a script or cron. Omit it to leave an existing subscription unchanged (new subscriptions default to `dm`). + ### Multi-profile setups: delivery is profile-owned In a one-gateway-per-profile deployment (one dispatcher, separate gateway