From 0b600c859aff172f2b62d2da7d9b08cedcb8ee61 Mon Sep 17 00:00:00 2001 From: xaviersudre <285329547+xaviersudre@users.noreply.github.com> Date: Thu, 30 Jul 2026 14:10:33 -0700 Subject: [PATCH] fix(kanban): wake API subscriptions in destination session --- gateway/kanban_watchers.py | 17 +++- .../test_kanban_notifier_apiserver_wake.py | 98 ++++++++++++++++--- 2 files changed, 102 insertions(+), 13 deletions(-) diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index c3924d774c..24864f7b11 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -711,7 +711,22 @@ class GatewayKanbanWatchersMixin: _session_key = "" _synth = "" if _wake_kinds: - _session_key = getattr(task, "session_id", None) or "" + if _is_push_adapter: + _session_key = getattr(task, "session_id", None) or "" + else: + # Non-push (api_server) wakes go to the + # subscription's delivery destination — + # sub["chat_id"] IS the raw session id the + # subscriber registered with. task.session_id + # is worker/creator provenance and may point + # at a WORKER session for child tasks with + # inherited subscriptions; falling back to it + # only when chat_id is empty (legacy rows). + _session_key = ( + sub["chat_id"] + or getattr(task, "session_id", None) + or "" + ) if _wake_kinds: _title = (task.title if task else sub["task_id"])[:120] _assignee = task.assignee if task else "" diff --git a/tests/gateway/test_kanban_notifier_apiserver_wake.py b/tests/gateway/test_kanban_notifier_apiserver_wake.py index 5374ecf44a..a257360d02 100644 --- a/tests/gateway/test_kanban_notifier_apiserver_wake.py +++ b/tests/gateway/test_kanban_notifier_apiserver_wake.py @@ -3,10 +3,9 @@ Covers the wrong-session-wake / silent-loss fixes: * a SendResult(success=False) return (the API server's send() stub) rewinds the cursor instead of advancing past a never-delivered event; -* api_server subscriptions wake the creator's REAL session via the - /v1/chat/completions self-post (raw task.session_id), never via - handle_message (which would run under a build_session_key()-derived key - that never matches the raw X-Hermes-Session-Id session real turns use). +* api_server subscriptions wake their ``chat_id`` delivery destinations via + the /v1/chat/completions self-post, never task ``session_id`` provenance or + handle_message (which would derive a different session key). """ import asyncio @@ -101,15 +100,13 @@ def _unseen_terminal_events(tid, platform, chat_id): conn.close() -def test_apiserver_sub_wakes_real_session_via_self_post(tmp_path, monkeypatch): - """An api_server subscription wakes the creator's REAL session by - self-posting with the task's raw session_id — never handle_message (which - would run the wake under a build_session_key()-derived key that can't - match the raw X-Hermes-Session-Id session).""" +def test_apiserver_sub_wakes_subscription_destination_via_self_post(tmp_path, monkeypatch): + """An api_server subscription wakes its chat_id destination, not the + task's worker-session provenance or a build_session_key()-derived session.""" monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "apiserver.db")) kb.init_db() tid = _create_completed_subscription( - "api_server", "raw-sid-123", session_id="raw-sid-123", + "api_server", "origin-session", session_id="worker-session", ) posts = [] @@ -129,10 +126,87 @@ def test_apiserver_sub_wakes_real_session_via_self_post(tmp_path, monkeypatch): "api_server wake must not go through handle_message (wrong-session bug)" ) assert len(posts) == 1 - assert posts[0]["session_id"] == "raw-sid-123" + assert posts[0]["session_id"] == "origin-session" + assert all(post["session_id"] != "worker-session" for post in posts) assert tid in posts[0]["text"] # The wake self-post IS the delivery on this path (no separate text-ping # fallback is attempted for stateless api_server subs) — cursor advances # once the wake succeeds. - assert _unseen_terminal_events(tid, "api_server", "raw-sid-123") == [] + assert _unseen_terminal_events(tid, "api_server", "origin-session") == [] + + +def test_apiserver_subscriptions_have_independent_wake_destinations( + tmp_path, monkeypatch, +): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "apiserver-multi.db")) + kb.init_db() + conn = kb.connect() + try: + tid = kb.create_task( + conn, + title="notify both", + assignee="worker", + session_id="worker-session", + ) + for chat_id in ("origin-a", "origin-b"): + kb.add_notify_sub( + conn, + task_id=tid, + platform="api_server", + chat_id=chat_id, + ) + kb.complete_task(conn, tid, summary="done once") + finally: + conn.close() + + posts = [] + + async def fake_self_post(adapter, *, text, session_id): + posts.append({"text": text, "session_id": session_id}) + + import gateway.wake as wake_mod + + monkeypatch.setattr(wake_mod, "_self_post_chat_completion", fake_self_post) + runner = _make_runner({Platform.API_SERVER: ApiServerLikeAdapter()}) + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + assert sorted(post["session_id"] for post in posts) == ["origin-a", "origin-b"] + assert all(post["session_id"] != "worker-session" for post in posts) + assert _unseen_terminal_events(tid, "api_server", "origin-a") == [] + assert _unseen_terminal_events(tid, "api_server", "origin-b") == [] + + +def test_apiserver_wake_failure_rewinds_then_retries_destination( + tmp_path, monkeypatch, +): + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "apiserver-retry.db")) + kb.init_db() + tid = _create_completed_subscription( + "api_server", "origin-session", session_id="worker-session", + ) + attempted_sessions = [] + + async def fail_once_then_succeed(adapter, *, text, session_id): + attempted_sessions.append(session_id) + if len(attempted_sessions) == 1: + raise RuntimeError("simulated wake failure") + + import gateway.wake as wake_mod + + monkeypatch.setattr( + wake_mod, + "_self_post_chat_completion", + fail_once_then_succeed, + ) + runner = _make_runner({Platform.API_SERVER: ApiServerLikeAdapter()}) + + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + assert _unseen_terminal_events(tid, "api_server", "origin-session") + + runner._running = True + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + assert attempted_sessions == ["origin-session", "origin-session"] + assert "worker-session" not in attempted_sessions + assert _unseen_terminal_events(tid, "api_server", "origin-session") == []