diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 3b5692667d96..50e44f980304 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -336,6 +336,14 @@ class GatewayKanbanWatchersMixin: continue title = (task.title if task else sub["task_id"])[:120] board_tag = f"[{board_slug}] " if board_slug else "" + # Per-subscription failure-counter key. Hoisted out of the + # event loop: the wake self-post path (in the loop's + # ``else`` clause) needs it even when every event in the + # claim was skipped before reaching the send site. + sub_key = ( + sub["task_id"], sub["platform"], + sub["chat_id"], sub.get("thread_id") or "", + ) for ev in d["events"]: kind = ev.kind # Identity prefix: attribute terminal pings to the @@ -408,10 +416,6 @@ class GatewayKanbanWatchersMixin: metadata: dict[str, Any] = {} if sub.get("thread_id"): metadata["thread_id"] = sub["thread_id"] - sub_key = ( - sub["task_id"], sub["platform"], - sub["chat_id"], sub.get("thread_id") or "", - ) # Adapters with no push channel (the API server — # ``supports_async_delivery = False``) can NEVER # satisfy a text-send: ``send()`` always reports @@ -434,7 +438,10 @@ class GatewayKanbanWatchersMixin: "on wake self-post instead", platform_str, sub["task_id"], ) - sub_fail_counts.pop(sub_key, None) + # Do NOT reset the failure counter here: on this + # path the wake self-post below IS the delivery, + # so the counter is resolved (reset or bumped) by + # the self-post outcome, not by skipping the send. continue try: _send_res = await adapter.send( @@ -511,12 +518,108 @@ class GatewayKanbanWatchersMixin: # dropping the subscription is the terminal action. break else: - # All events delivered; advance cursor. The cursor + # All text pings delivered (or intentionally skipped + # for non-push adapters, whose delivery is the wake + # self-post below). Whether the cursor may advance now + # depends on the adapter class: + # + # * push-capable: the text send WAS the delivery, so + # advance immediately (pre-existing behavior); the + # wake injection below stays best-effort. + # * non-push (api_server): the wake self-post IS the + # delivery. Advancing first would let a failed / + # retry-exhausted self-post (swallowed by the + # best-effort except) permanently lose the event. + # So the self-post runs FIRST and the cursor only + # advances after it succeeds — a failure rewinds the + # claim exactly like a failed send() above, so the + # 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} + from gateway.wake import adapter_supports_push as _adapter_push_ok + + _is_push_adapter = _adapter_push_ok(adapter) + _session_key = "" + _synth = "" + if _wake_kinds: + _session_key = getattr(task, "session_id", None) or "" + if _wake_kinds and _session_key: + _title = (task.title if task else sub["task_id"])[:120] + _assignee = task.assignee if task else "" + _parts = [] + if "completed" in _wake_kinds: _parts.append(t("gateway.kanban.wake.completed")) + if "gave_up" in _wake_kinds: _parts.append(t("gateway.kanban.wake.gave_up")) + if "crashed" in _wake_kinds: _parts.append(t("gateway.kanban.wake.crashed")) + if "timed_out" in _wake_kinds: _parts.append(t("gateway.kanban.wake.timed_out")) + if "blocked" in _wake_kinds: _parts.append(t("gateway.kanban.wake.blocked")) + _status = t("gateway.kanban.wake.status_joiner").join(_parts) or t("gateway.kanban.wake.status_default") + _synth = t( + "gateway.kanban.wake.message", + task_id=sub["task_id"], + status=_status, + title=_title, + assignee=_assignee, + board=board_slug, + ) + + if not _is_push_adapter and _wake_kinds and _session_key: + # Wake self-post IS the delivery on this path — + # it must succeed BEFORE the cursor advances. + from gateway.wake import deliver_wake + + try: + await deliver_wake( + adapter, + text=_synth, + session_id=_session_key, + ) + logger.info( + "kanban notifier: woke agent for %s on %s/%s profile=%s events=%s", + sub["task_id"], platform_str, sub["chat_id"], sub_profile or "default", _wake_kinds, + ) + sub_fail_counts.pop(sub_key, None) + except Exception as _wk_err: + fails = sub_fail_counts.get(sub_key, 0) + 1 + sub_fail_counts[sub_key] = fails + logger.warning( + "kanban notifier: wake self-post failed " + "for %s (attempt %d/%d): %s", + sub["task_id"], fails, + MAX_SEND_FAILURES, _wk_err, exc_info=True, + ) + if fails >= MAX_SEND_FAILURES: + logger.warning( + "kanban notifier: dropping subscription " + "%s on %s after %d consecutive wake failures", + sub["task_id"], platform_str, fails, + ) + await asyncio.to_thread(self._kanban_unsub, sub, board_slug) + sub_fail_counts.pop(sub_key, None) + else: + # Rewind the pre-send claim so the next + # tick retries the self-post — the event + # is NOT lost. + await asyncio.to_thread( + self._kanban_rewind, + sub, + d["cursor"], + d.get("old_cursor", 0), + board_slug, + ) + continue + + # Delivery complete (text ping for push adapters, wake + # self-post for non-push): advance cursor. The cursor # is the dedup mechanism — it prevents re-delivery # of the same event on subsequent ticks. await asyncio.to_thread( self._kanban_advance, sub, d["cursor"], board_slug, ) + if not _is_push_adapter: + # Nothing left to deliver on this path (the wake, + # if any, already succeeded above). + sub_fail_counts.pop(sub_key, None) # Unsubscribe only when the task has reached a truly # final status (done / archived). For blocked / # gave_up / crashed / timed_out the subscription is @@ -524,77 +627,50 @@ 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. - 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} - if _wake_kinds: + if _is_push_adapter and _wake_kinds and _session_key: try: - _session_key = getattr(task, "session_id", None) or "" - if _session_key: - _title = (task.title if task else sub["task_id"])[:120] - _assignee = task.assignee if task else "" - _parts = [] - if "completed" in _wake_kinds: _parts.append(t("gateway.kanban.wake.completed")) - if "gave_up" in _wake_kinds: _parts.append(t("gateway.kanban.wake.gave_up")) - if "crashed" in _wake_kinds: _parts.append(t("gateway.kanban.wake.crashed")) - if "timed_out" in _wake_kinds: _parts.append(t("gateway.kanban.wake.timed_out")) - if "blocked" in _wake_kinds: _parts.append(t("gateway.kanban.wake.blocked")) - _status = t("gateway.kanban.wake.status_joiner").join(_parts) or t("gateway.kanban.wake.status_default") - _synth = t( - "gateway.kanban.wake.message", - task_id=sub["task_id"], - status=_status, - title=_title, - assignee=_assignee, - board=board_slug, - ) - 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:") 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. - _source = SessionSource( - platform=plat, - chat_id=sub["chat_id"], - chat_type="group", - thread_id=sub.get("thread_id") or None, - user_id=sub.get("user_id"), - profile=sub_profile or None, - ) - # deliver_wake preserves the synthetic - # MessageEvent/handle_message path for - # push-capable adapters, and self-POSTs - # /v1/chat/completions with the task's RAW - # session id for the stateless API server — - # handle_message there would run the wake - # under a build_session_key()-derived key - # that never matches the raw - # X-Hermes-Session-Id session real turns - # run under (wrong-session wake bug). - await deliver_wake( - adapter, - text=_synth, - session_id=_session_key, - source=_source, - ) - logger.info( - "kanban notifier: woke agent for %s on %s/%s profile=%s events=%s", - sub["task_id"], platform_str, sub["chat_id"], sub_profile or "default", _wake_kinds, - ) + 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:") 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. + _source = SessionSource( + platform=plat, + chat_id=sub["chat_id"], + chat_type="group", + thread_id=sub.get("thread_id") or None, + user_id=sub.get("user_id"), + profile=sub_profile or None, + ) + # deliver_wake preserves the synthetic + # MessageEvent/handle_message path for + # push-capable adapters (the non-push / + # self-post branch is handled BEFORE the + # cursor advance above). + await deliver_wake( + adapter, + text=_synth, + session_id=_session_key, + source=_source, + ) + logger.info( + "kanban notifier: woke agent for %s on %s/%s profile=%s events=%s", + sub["task_id"], platform_str, sub["chat_id"], sub_profile or "default", _wake_kinds, + ) except Exception as _wk_err: # Best-effort: the notification itself already # delivered and the cursor has advanced, so a diff --git a/tests/gateway/test_kanban_notifier_apiserver_wake.py b/tests/gateway/test_kanban_notifier_apiserver_wake.py index 0fd7f3ed999b..4d05ee3ee4ce 100644 --- a/tests/gateway/test_kanban_notifier_apiserver_wake.py +++ b/tests/gateway/test_kanban_notifier_apiserver_wake.py @@ -152,3 +152,68 @@ def test_apiserver_sub_wakes_real_session_via_self_post(tmp_path, monkeypatch): # fallback is attempted for stateless api_server subs) — cursor advances # once the wake succeeds. assert _unseen_terminal_events(tid, "api_server", "raw-sid-123") == [] + + +def test_apiserver_failed_self_post_rewinds_cursor(tmp_path, monkeypatch): + """A failed/exhausted wake self-post must NOT advance the cursor: on the + api_server path the self-post IS the delivery, so advancing first would + permanently lose the event behind a best-effort except. The claim is + rewound and the event stays visible for the next tick's retry.""" + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "apiserver_fail.db")) + kb.init_db() + tid = _create_completed_subscription( + "api_server", "raw-sid-999", session_id="raw-sid-999", + ) + + async def failing_self_post(adapter, *, text, session_id): + raise RuntimeError("self-post exhausted retries") + + import gateway.wake as wake_mod + + monkeypatch.setattr(wake_mod, "_self_post_chat_completion", failing_self_post) + + adapter = ApiServerLikeAdapter() + runner = _make_runner({Platform.API_SERVER: adapter}) + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + # Event NOT lost: the cursor was rewound, so the completed event is still + # unseen and will be re-claimed (and the self-post retried) next tick. + assert [ev.kind for ev in _unseen_terminal_events(tid, "api_server", "raw-sid-999")] == [ + "completed" + ] + # And the failure was counted toward the drop threshold. + assert list(runner._kanban_sub_fail_counts.values()) == [1] + + +def test_apiserver_self_post_succeeds_after_earlier_failure(tmp_path, monkeypatch): + """The rewound event is retried on the next tick; a successful self-post + then advances the cursor and clears the failure counter.""" + monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "apiserver_retry.db")) + kb.init_db() + tid = _create_completed_subscription( + "api_server", "raw-sid-777", session_id="raw-sid-777", + ) + + calls = {"n": 0} + + async def flaky_self_post(adapter, *, text, session_id): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("transient outage") + + import gateway.wake as wake_mod + + monkeypatch.setattr(wake_mod, "_self_post_chat_completion", flaky_self_post) + + adapter = ApiServerLikeAdapter() + runner = _make_runner({Platform.API_SERVER: adapter}) + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + assert calls["n"] == 1 + assert len(_unseen_terminal_events(tid, "api_server", "raw-sid-777")) == 1 + + # Second tick: the re-claimed event's self-post succeeds → cursor advances. + runner._running = True + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + assert calls["n"] == 2 + assert _unseen_terminal_events(tid, "api_server", "raw-sid-777") == [] + assert runner._kanban_sub_fail_counts == {}