feat(kanban): explicit notify/wake delivery modes with faithful wake session routing
Salvage of #37865 by @verybigdog. Adds delivery_mode (notify / notify+wake / wake) on kanban notify subscriptions, persists chat_type + user_id_alt so a woken turn reconstructs the creator's real session key, inherits the return path to child tasks, and keeps wake out of the model-exposed send_message schema. Original commits were authored under a local placeholder identity (hermes-agent@users.noreply.local); re-attributed to the contributor's public email.
This commit is contained in:
@@ -426,6 +426,9 @@ class GatewayKanbanWatchersMixin:
|
||||
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
|
||||
@@ -530,6 +533,7 @@ class GatewayKanbanWatchersMixin:
|
||||
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 —
|
||||
@@ -547,7 +551,7 @@ class GatewayKanbanWatchersMixin:
|
||||
# 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 "
|
||||
@@ -559,6 +563,12 @@ class GatewayKanbanWatchersMixin:
|
||||
# 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,
|
||||
@@ -652,7 +662,11 @@ class GatewayKanbanWatchersMixin:
|
||||
# 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)
|
||||
@@ -660,7 +674,7 @@ class GatewayKanbanWatchersMixin:
|
||||
_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 = []
|
||||
@@ -743,7 +757,7 @@ class GatewayKanbanWatchersMixin:
|
||||
# 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
|
||||
@@ -775,6 +789,7 @@ class GatewayKanbanWatchersMixin:
|
||||
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
|
||||
|
||||
@@ -22644,6 +22644,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
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 "",
|
||||
scope_id=str(getattr(context.source, "scope_id", "") or ""),
|
||||
session_key=context.session_key,
|
||||
|
||||
@@ -78,6 +78,7 @@ _SESSION_CHAT_TYPE: ContextVar = ContextVar("HERMES_SESSION_CHAT_TYPE", default=
|
||||
_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)
|
||||
# Platform-neutral scope discriminator (Discord guild / Slack workspace /
|
||||
# Matrix server) of the originating chat. Captured at session-bind time so
|
||||
@@ -142,6 +143,7 @@ _VAR_MAP = {
|
||||
"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_SCOPE_ID": _SESSION_SCOPE_ID,
|
||||
"HERMES_SESSION_KEY": _SESSION_KEY,
|
||||
@@ -219,6 +221,7 @@ def set_session_vars(
|
||||
chat_name: str = "",
|
||||
thread_id: str = "",
|
||||
user_id: str = "",
|
||||
user_id_alt: str = "",
|
||||
user_name: str = "",
|
||||
scope_id: str = "",
|
||||
session_key: str = "",
|
||||
@@ -262,6 +265,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_SCOPE_ID.set(scope_id),
|
||||
_SESSION_KEY.set(session_key),
|
||||
@@ -300,6 +304,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_SCOPE_ID,
|
||||
_SESSION_KEY,
|
||||
|
||||
@@ -524,6 +524,11 @@ class GatewaySlashCommandsMixin:
|
||||
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
|
||||
@@ -542,7 +547,11 @@ class GatewaySlashCommandsMixin:
|
||||
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:
|
||||
|
||||
+29
-2
@@ -814,13 +814,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",
|
||||
@@ -2923,7 +2943,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 "")
|
||||
@@ -2943,8 +2965,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
|
||||
|
||||
|
||||
|
||||
+133
-16
@@ -1502,10 +1502,12 @@ 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,
|
||||
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,
|
||||
@@ -2680,9 +2682,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(
|
||||
@@ -2810,8 +2835,10 @@ _REBUILD_SPECS = {
|
||||
"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)",),
|
||||
@@ -3454,6 +3481,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,
|
||||
@@ -10863,6 +10931,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]:
|
||||
@@ -10904,15 +10980,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
|
||||
@@ -10922,44 +11017,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
|
||||
@@ -10969,10 +11078,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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)."
|
||||
)
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -137,6 +660,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",
|
||||
@@ -181,6 +705,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
|
||||
@@ -216,6 +831,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()
|
||||
|
||||
@@ -838,6 +838,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=<key>."""
|
||||
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
|
||||
|
||||
@@ -1556,9 +1556,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", "")
|
||||
@@ -1591,9 +1594,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
|
||||
|
||||
@@ -769,6 +769,7 @@ hermes kanban stats [--json] # per-status + per-assign
|
||||
hermes kanban log <id> [--tail BYTES] # worker log from ~/.hermes/kanban/logs/
|
||||
hermes kanban notify-subscribe <id> # gateway bridge hook (used by /kanban in the gateway)
|
||||
--platform <name> --chat-id <id> [--thread-id <id>] [--user-id <id>]
|
||||
[--chat-type dm|group|channel|thread] [--delivery-mode notify|notify+wake|wake]
|
||||
hermes kanban notify-list [<id>] [--json]
|
||||
hermes kanban notify-unsubscribe <id>
|
||||
--platform <name> --chat-id <id> [--thread-id <id>]
|
||||
@@ -885,6 +886,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.
|
||||
@@ -1002,7 +1005,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
|
||||
@@ -1010,6 +1014,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
|
||||
|
||||
Reference in New Issue
Block a user