fix(kanban): wake API subscriptions in destination session

This commit is contained in:
xaviersudre
2026-07-30 14:10:33 -07:00
committed by Teknium
parent 720f0443a0
commit 0b600c859a
2 changed files with 102 additions and 13 deletions
+16 -1
View File
@@ -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 ""
@@ -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") == []