From a417c6e08d496737d1ac143d93f52362b5cd2eb5 Mon Sep 17 00:00:00 2001 From: lijun Date: Sun, 5 Jul 2026 10:12:28 +0800 Subject: [PATCH] fix(gateway): preserve kanban notifier chat type --- gateway/kanban_watchers.py | 33 ++++---- gateway/run.py | 3 + gateway/session_context.py | 5 ++ gateway/slash_commands.py | 2 + hermes_cli/kanban_db.py | 34 +++++++- tests/gateway/test_kanban_notifier.py | 115 ++++++++++++++++++++++++++ tools/kanban_tools.py | 10 ++- 7 files changed, 176 insertions(+), 26 deletions(-) diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 50e44f980304..4f53c0e1f8fe 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -631,27 +631,24 @@ class GatewayKanbanWatchersMixin: try: from gateway.session import SessionSource from gateway.wake import deliver_wake - # KNOWN LIMITATION (tracked follow-up): the - # subscription row does not persist the - # creator's chat_type, and it is not carried - # on the session-context bridge, so we cannot - # faithfully reconstruct the creator's real - # session key here. build_session_key() keys - # DMs (":dm:") on a wholly different - # shape from group/thread, so any hardcoded - # value mis-routes some creators. "group" is - # the least-surprising default for the - # dashboard/group flows this wake primarily - # serves; DM-originated creators are handled - # by the follow-up that stamps + persists - # chat_type end-to-end. handle_message() - # get_or_create_session's the target, so a - # mismatch degrades to "wake lands in a fresh - # group session" — never an exception. + # Rebuild the creator's real session scope from + # the chat_type persisted on the subscription + # row (#56580). build_session_key() keys DMs + # (":dm:") on a wholly different shape + # from group/thread, so the old hardcoded + # "group" mis-routed DM/thread creators into a + # fresh session. Legacy rows written before the + # column existed store NULL/'' — fall back to + # "group" for them (the historical default that + # suits the dashboard/group flows). + # handle_message() get_or_create_session's the + # target, so a mismatch only ever degrades to a + # fresh session, never an exception. + _chat_type = str(sub.get("chat_type") or "").strip() or "group" _source = SessionSource( platform=plat, chat_id=sub["chat_id"], - chat_type="group", + chat_type=_chat_type, thread_id=sub.get("thread_id") or None, user_id=sub.get("user_id"), profile=sub_profile or None, diff --git a/gateway/run.py b/gateway/run.py index 9e91d9083f2b..02677455d583 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -17446,6 +17446,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew return set_session_vars( platform=context.source.platform.value, chat_id=context.source.chat_id, + chat_type=( + str(context.source.chat_type) if context.source.chat_type else "" + ), 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 "", diff --git a/gateway/session_context.py b/gateway/session_context.py index a792726e47e1..cf227fac1211 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -73,6 +73,7 @@ def session_context_engaged() -> bool: _SESSION_PLATFORM: ContextVar = ContextVar("HERMES_SESSION_PLATFORM", default=_UNSET) _SESSION_SOURCE: ContextVar = ContextVar("HERMES_SESSION_SOURCE", default=_UNSET) _SESSION_CHAT_ID: ContextVar = ContextVar("HERMES_SESSION_CHAT_ID", default=_UNSET) +_SESSION_CHAT_TYPE: ContextVar = ContextVar("HERMES_SESSION_CHAT_TYPE", default=_UNSET) _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) @@ -123,6 +124,7 @@ _VAR_MAP = { "HERMES_SESSION_PLATFORM": _SESSION_PLATFORM, "HERMES_SESSION_SOURCE": _SESSION_SOURCE, "HERMES_SESSION_CHAT_ID": _SESSION_CHAT_ID, + "HERMES_SESSION_CHAT_TYPE": _SESSION_CHAT_TYPE, "HERMES_SESSION_CHAT_NAME": _SESSION_CHAT_NAME, "HERMES_SESSION_THREAD_ID": _SESSION_THREAD_ID, "HERMES_SESSION_USER_ID": _SESSION_USER_ID, @@ -157,6 +159,7 @@ def set_session_vars( platform: str = "", source: str = "", chat_id: str = "", + chat_type: str = "", chat_name: str = "", thread_id: str = "", user_id: str = "", @@ -193,6 +196,7 @@ def set_session_vars( _SESSION_PLATFORM.set(platform), _SESSION_SOURCE.set(source), _SESSION_CHAT_ID.set(chat_id), + _SESSION_CHAT_TYPE.set(chat_type), _SESSION_CHAT_NAME.set(chat_name), _SESSION_THREAD_ID.set(thread_id), _SESSION_USER_ID.set(user_id), @@ -228,6 +232,7 @@ def clear_session_vars(tokens: list) -> None: _SESSION_PLATFORM, _SESSION_SOURCE, _SESSION_CHAT_ID, + _SESSION_CHAT_TYPE, _SESSION_CHAT_NAME, _SESSION_THREAD_ID, _SESSION_USER_ID, diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 58c365ddb3e7..65af38ab7be2 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -478,6 +478,7 @@ class GatewaySlashCommandsMixin: platform.value if hasattr(platform, "value") else str(platform or "") ).lower() chat_id = str(getattr(source, "chat_id", "") or "") + 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 if platform_str and chat_id: @@ -488,6 +489,7 @@ class GatewaySlashCommandsMixin: _kb.add_notify_sub( conn, task_id=task_id, platform=platform_str, chat_id=chat_id, + chat_type=chat_type, thread_id=thread_id or None, user_id=user_id, notifier_profile=getattr(self, "_kanban_notifier_profile", None) or self._active_profile_name(), diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 75f922d0565c..f8735f55d0f6 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -1300,6 +1300,7 @@ CREATE TABLE IF NOT EXISTS 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, @@ -2445,6 +2446,10 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: _add_column_if_missing( conn, "kanban_notify_subs", "notifier_profile", "notifier_profile TEXT" ) + if "chat_type" not in notify_cols: + _add_column_if_missing( + conn, "kanban_notify_subs", "chat_type", "chat_type TEXT" + ) # One-shot backfill: any task that is 'running' before runs existed # had its claim_lock / claim_expires / worker_pid on the task row. @@ -2567,7 +2572,7 @@ _REBUILD_SPECS = { "kanban_notify_subs": ( "CREATE TABLE kanban_notify_subs (" " task_id TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL," - " thread_id TEXT NOT NULL DEFAULT '', user_id TEXT," + " chat_type TEXT, thread_id TEXT NOT NULL DEFAULT '', user_id TEXT," " notifier_profile TEXT, created_at INTEGER NOT NULL," " last_event_id INTEGER NOT NULL DEFAULT 0," " PRIMARY KEY (task_id, platform, chat_id, thread_id))", @@ -9335,6 +9340,7 @@ 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, notifier_profile: Optional[str] = None, @@ -9346,11 +9352,31 @@ def add_notify_sub( conn.execute( """ INSERT OR IGNORE INTO kanban_notify_subs - (task_id, platform, chat_id, thread_id, user_id, notifier_profile, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?) + (task_id, platform, chat_id, chat_type, thread_id, user_id, notifier_profile, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, - (task_id, platform, chat_id, thread_id or "", user_id, notifier_profile, now), + ( + task_id, + platform, + chat_id, + chat_type, + thread_id or "", + user_id, + notifier_profile, + now, + ), ) + if chat_type: + # Self-heal rows created before chat_type was persisted. + 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 notifier_profile: # Self-heal legacy rows that predate notifier ownership by # backfilling only when the existing value is unset. diff --git a/tests/gateway/test_kanban_notifier.py b/tests/gateway/test_kanban_notifier.py index 22c70e1c3989..b0764e24a9c9 100644 --- a/tests/gateway/test_kanban_notifier.py +++ b/tests/gateway/test_kanban_notifier.py @@ -1,4 +1,5 @@ import asyncio +import sqlite3 from pathlib import Path @@ -10,10 +11,14 @@ from hermes_cli import kanban_db as kb class RecordingAdapter: def __init__(self): self.sent = [] + self.handled = [] async def send(self, chat_id, text, metadata=None): self.sent.append({"chat_id": chat_id, "text": text, "metadata": metadata or {}}) + async def handle_message(self, event): + self.handled.append(event) + class DisconnectedAdapters(dict): """Expose a platform during collection, then simulate disconnect on get().""" @@ -294,6 +299,116 @@ def test_notifier_owning_profile_adapter_no_default_fallback(tmp_path, monkeypat assert [ev.kind for ev in _unseen_terminal_events_for(tid, "chat-beta")] == ["completed"] +def test_notifier_wakeup_uses_subscription_chat_type(tmp_path, monkeypatch): + db_path = tmp_path / "chat-type-wakeup.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + + conn = kb.connect() + try: + tid = kb.create_task( + conn, + title="dm requester", + assignee="worker", + session_id="origin-session", + ) + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat-dm", + chat_type="dm", + ) + kb.complete_task(conn, tid, summary="done") + finally: + conn.close() + + adapter = RecordingAdapter() + asyncio.run(_run_one_notifier_tick(monkeypatch, _make_runner(adapter))) + + assert len(adapter.sent) == 1 + assert len(adapter.handled) == 1 + assert adapter.handled[0].source.chat_type == "dm" + + +def test_auto_subscribe_persists_session_chat_type(tmp_path, monkeypatch): + db_path = tmp_path / "auto-sub-chat-type.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + + from gateway.session_context import clear_session_vars, set_session_vars + from tools import kanban_tools + + monkeypatch.setattr( + kanban_tools, + "load_config", + lambda: {"kanban": {"auto_subscribe_on_create": True}}, + ) + + tokens = set_session_vars( + platform="telegram", + chat_id="chat-dm", + chat_type="dm", + ) + conn = kb.connect() + try: + tid = kb.create_task(conn, title="auto sub", assignee="worker") + + assert kanban_tools._maybe_auto_subscribe(conn, tid) is True + [sub] = kb.list_notify_subs(conn, task_id=tid) + assert sub["chat_type"] == "dm" + finally: + conn.close() + clear_session_vars(tokens) + + +def test_notify_sub_migration_adds_chat_type_to_legacy_table(tmp_path, monkeypatch): + db_path = tmp_path / "legacy-notify-sub.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + + legacy = sqlite3.connect(db_path) + try: + legacy.execute( + """ + CREATE TABLE kanban_notify_subs ( + task_id TEXT NOT NULL, + platform TEXT NOT NULL, + chat_id TEXT NOT NULL, + thread_id TEXT NOT NULL DEFAULT '', + user_id TEXT, + notifier_profile TEXT, + created_at INTEGER NOT NULL, + last_event_id INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (task_id, platform, chat_id, thread_id) + ) + """ + ) + legacy.commit() + finally: + legacy.close() + + kb.init_db() + conn = kb.connect() + try: + cols = { + row["name"] for row in conn.execute("PRAGMA table_info(kanban_notify_subs)") + } + assert "chat_type" in cols + + tid = kb.create_task(conn, title="legacy sub", assignee="worker") + kb.add_notify_sub( + conn, + task_id=tid, + platform="telegram", + chat_id="chat-dm", + chat_type="dm", + ) + [sub] = kb.list_notify_subs(conn, task_id=tid) + assert sub["chat_type"] == "dm" + finally: + conn.close() + + def _unseen_terminal_events_for(tid, chat_id): conn = kb.connect() try: diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index 46991b4a477b..acf3442150c9 100644 --- a/tools/kanban_tools.py +++ b/tools/kanban_tools.py @@ -1272,10 +1272,10 @@ def _maybe_auto_subscribe(conn: Any, task_id: str) -> bool: Subscription paths: - - **Gateway** (telegram/discord/slack/etc): ``HERMES_SESSION_PLATFORM`` - and ``HERMES_SESSION_CHAT_ID`` are set in ContextVars by the - messaging gateway before agent dispatch. The notification poller - already keys off these, so we just register a row. + - **Gateway** (telegram/discord/slack/etc): ``HERMES_SESSION_PLATFORM``, + ``HERMES_SESSION_CHAT_ID``, and ``HERMES_SESSION_CHAT_TYPE`` are set in + ContextVars by the messaging gateway before agent dispatch. The + notification poller already keys off these, so we just register a row. - **TUI** (herm desktop / herm TUI): the platform/chat_id ContextVars are intentionally cleared (TUI is a single-channel local UI, not @@ -1333,6 +1333,7 @@ def _maybe_auto_subscribe(conn: Any, task_id: str) -> bool: chat_id = session_key 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 notifier_profile = ( get_session_env("HERMES_SESSION_PROFILE", "") or os.environ.get("HERMES_PROFILE") @@ -1343,6 +1344,7 @@ 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, + chat_type=chat_type, thread_id=thread_id, user_id=user_id, notifier_profile=notifier_profile, )