mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-28 18:19:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(gateway): preserve kanban notifier chat type

Browse source
This commit is contained in:
lijun 2026-07-05 10:12:28 +08:00 • committed by Teknium
parent aa636c6fca
commit a417c6e08d
7 changed files with 176 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

33
gateway/kanban_watchers.py
View file

@ -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:<chat_id>") 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:<chat_id>") 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,

3
gateway/run.py
View file

@ -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 "",

5
gateway/session_context.py
View file

@ -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,

2
gateway/slash_commands.py
View file

@ -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(),

34
hermes_cli/kanban_db.py
View file

@ -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.

115
tests/gateway/test_kanban_notifier.py
View file

@ -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:

10
tools/kanban_tools.py
View file

@ -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,
)