mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-28 18:19:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(gateway): isolate per-subscription failures in kanban notifier

Browse source 
The kanban notifier _collect() loop iterates subscriptions without
per-subscription error handling. When claim_unseen_events_for_sub raises
for one subscription (e.g. DB corruption, lock contention), the entire
tick aborts — silently blocking delivery for ALL other subscriptions.

Wrap the per-subscription logic in try/except so one bad subscription
logs a warning and continues to the next, instead of jamming the
entire notifier.

Closes #59269
This commit is contained in:
AlexFucuson9 2026-07-06 08:10:26 +07:00 • committed by Teknium
parent 83dc0b9b85
commit f6ccfa6bd3
2 changed files with 87 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

79
gateway/kanban_watchers.py
View file

@ -252,45 +252,54 @@ class GatewayKanbanWatchersMixin:
if not subs:
logger.debug("kanban notifier: board %s has no subscriptions", slug)
for sub in subs:
owner_profile = sub.get("notifier_profile") or None
if owner_profile and owner_profile != notifier_profile:
_owner_adapters = getattr(self, "_profile_adapters", {}).get(owner_profile)
if not _owner_adapters:
try:
owner_profile = sub.get("notifier_profile") or None
if owner_profile and owner_profile != notifier_profile:
_owner_adapters = getattr(self, "_profile_adapters", {}).get(owner_profile)
if not _owner_adapters:
logger.debug(
"kanban notifier: subscription for %s owned by profile %s; current profile %s has no adapter for it, skipping",
sub.get("task_id"), owner_profile, notifier_profile,
)
continue
platform = (sub.get("platform") or "").lower()
if platform not in active_platforms:
logger.debug(
"kanban notifier: subscription for %s owned by profile %s; current profile %s has no adapter for it, skipping",
sub.get("task_id"), owner_profile, notifier_profile,
"kanban notifier: subscription for %s on %s skipped; adapter not connected",
sub.get("task_id"), platform or "<missing>",
)
continue
platform = (sub.get("platform") or "").lower()
if platform not in active_platforms:
logger.debug(
"kanban notifier: subscription for %s on %s skipped; adapter not connected",
sub.get("task_id"), platform or "<missing>",
old_cursor, cursor, events = _kb.claim_unseen_events_for_sub(
conn,
task_id=sub["task_id"],
platform=sub["platform"],
chat_id=sub["chat_id"],
thread_id=sub.get("thread_id") or "",
kinds=TERMINAL_KINDS,
)
if not events:
continue
task = _kb.get_task(conn, sub["task_id"])
logger.debug(
"kanban notifier: claimed %d event(s) for %s on board %s cursor %s→%s",
len(events), sub["task_id"], slug, old_cursor, cursor,
)
deliveries.append({
"sub": sub,
"old_cursor": old_cursor,
"cursor": cursor,
"events": events,
"task": task,
"board": slug,
})
except Exception as sub_exc:
# Isolate per-subscription failures so one
# bad subscription cannot block delivery for
# all other subscriptions in this tick.
logger.warning(
"kanban notifier: subscription for %s on board %s failed: %s",
sub.get("task_id"), slug, sub_exc,
)
continue
old_cursor, cursor, events = _kb.claim_unseen_events_for_sub(
conn,
task_id=sub["task_id"],
platform=sub["platform"],
chat_id=sub["chat_id"],
thread_id=sub.get("thread_id") or "",
kinds=TERMINAL_KINDS,
)
if not events:
continue
task = _kb.get_task(conn, sub["task_id"])
logger.debug(
"kanban notifier: claimed %d event(s) for %s on board %s cursor %s→%s",
len(events), sub["task_id"], slug, old_cursor, cursor,
)
deliveries.append({
"sub": sub,
"old_cursor": old_cursor,
"cursor": cursor,
"events": events,
"task": task,
"board": slug,
})
finally:
conn.close()
return deliveries

43
tests/gateway/test_kanban_notifier.py
View file

@ -480,3 +480,46 @@ def _unseen_terminal_events_for(tid, chat_id):
return events
finally:
conn.close()
def test_kanban_notifier_isolates_per_subscription_failure(tmp_path, monkeypatch):
"""One bad subscription must not block delivery for all others.
Regression for #59269: when claim_unseen_events_for_sub raises for one
subscription, the entire notifier tick used to abort — silently blocking
delivery for every other subscription.
"""
db_path = tmp_path / "isolation.db"
monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path))
kb.init_db()
# Create two tasks with subscriptions and complete both.
conn = kb.connect()
try:
tid_good = kb.create_task(conn, title="good task", assignee="worker")
kb.add_notify_sub(conn, task_id=tid_good, platform="telegram", chat_id="chat-good")
kb.complete_task(conn, tid_good, summary="done")
tid_bad = kb.create_task(conn, title="bad task", assignee="worker")
kb.add_notify_sub(conn, task_id=tid_bad, platform="telegram", chat_id="chat-bad")
kb.complete_task(conn, tid_bad, summary="done")
finally:
conn.close()
original_claim = kb.claim_unseen_events_for_sub
def selective_claim(conn, task_id, **kwargs):
if task_id == tid_bad:
raise RuntimeError("simulated DB corruption for bad task")
return original_claim(conn, task_id=task_id, **kwargs)
monkeypatch.setattr(kb, "claim_unseen_events_for_sub", selective_claim)
adapter = RecordingAdapter()
runner = _make_runner(adapter)
asyncio.run(_run_one_notifier_tick(monkeypatch, runner))
# The good task must still be delivered despite the bad task failing.
assert len(adapter.sent) == 1
assert tid_good in adapter.sent[0]["text"]