mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-24 16:54:43 +00:00
fix(kanban): advance notify cursor only after a successful wake self-post
On non-push adapters (api_server) the wake self-post IS the delivery, but the cursor advanced before the self-post ran and a failed/exhausted post was swallowed by the best-effort except — permanently losing the event. Reorder the else-branch: for non-push adapters run the self-post FIRST and only advance the cursor once it succeeds. A failure rewinds the pre-send claim (same guarantee as the existing SendResult(success=False) path) so the next tick retries, with the same MAX_SEND_FAILURES drop threshold. Push-capable adapters keep the pre-existing advance-then-best-effort-wake behavior. Follow-up to #64998 (sweeper review F1).
This commit is contained in:
Teknium
parent
246eacea7b
commit
2cc0ff44b6
2 changed files with 217 additions and 76 deletions
|
|
@ -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:<chat_id>") 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:<chat_id>") 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
|
||||
|
|
|
|||
|
|
@ -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 == {}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue