"""Kanban notifier behavior on stateless (api_server) subscriptions. 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 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 from gateway.config import Platform from gateway.platforms.base import SendResult from gateway.run import GatewayRunner from hermes_cli import kanban_db as kb class SoftFailAdapter: """Push-capable adapter whose send() returns SendResult(success=False) WITHOUT raising — previously treated as delivered (event lost).""" def __init__(self): self.attempts = 0 async def send(self, chat_id, text, metadata=None): self.attempts += 1 return SendResult(success=False, error="soft failure") class ApiServerLikeAdapter: supports_async_delivery = False def __init__(self): self._host = "127.0.0.1" self._port = 8642 self._api_key = "k" self._model_name = "hermes" self.handle_message_calls = [] self.send_calls = 0 async def send(self, chat_id, text, metadata=None): self.send_calls += 1 return SendResult( success=False, error="API server uses HTTP request/response, not send()", ) async def handle_message(self, event): self.handle_message_calls.append(event) async def _run_one_notifier_tick(monkeypatch, runner): real_sleep = asyncio.sleep async def fake_sleep(delay): if delay == 5: return None runner._running = False await real_sleep(0) monkeypatch.setattr(asyncio, "sleep", fake_sleep) await runner._kanban_notifier_watcher(interval=1) def _make_runner(adapters): runner = GatewayRunner.__new__(GatewayRunner) runner._running = True runner.adapters = adapters runner._kanban_sub_fail_counts = {} runner._kanban_dispatcher_lock_handle = object() return runner def _create_completed_subscription(platform, chat_id, session_id=None): conn = kb.connect() try: tid = kb.create_task( conn, title="notify once", assignee="worker", session_id=session_id, ) kb.add_notify_sub(conn, task_id=tid, platform=platform, chat_id=chat_id) kb.complete_task(conn, tid, summary="done once") return tid finally: conn.close() def _unseen_terminal_events(tid, platform, chat_id): conn = kb.connect() try: _, events = kb.unseen_events_for_sub( conn, task_id=tid, platform=platform, chat_id=chat_id, kinds=["completed", "blocked", "gave_up", "crashed", "timed_out"], ) return events finally: conn.close() 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", "origin-session", session_id="worker-session", ) 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) adapter = ApiServerLikeAdapter() runner = _make_runner({Platform.API_SERVER: adapter}) asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) assert adapter.handle_message_calls == [], ( "api_server wake must not go through handle_message (wrong-session bug)" ) assert len(posts) == 1 assert posts[0]["session_id"] == "origin-session" assert all(post["session_id"] != "worker-session" for post in posts) wake_text = posts[0]["text"] assert tid in wake_text # Graph-safe wake turn (#70752): the synthetic turn must carry the # worker's completion handoff and the don't-recreate guidance so a # woken orchestrator doesn't re-decompose existing work. assert "done once" in wake_text, "creator wake must carry the worker handoff" assert "not a request to decompose" in wake_text.lower() assert "do not recreate" in wake_text.lower() # 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", "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") == []