From badb240ffa30d08e7aa19d6e3fce8d644cb4b004 Mon Sep 17 00:00:00 2001 From: falkoro <39274208+falkoro@users.noreply.github.com> Date: Tue, 7 Jul 2026 10:03:08 +0200 Subject: [PATCH] fix(tui): deliver kanban notify subscriptions to TUI/desktop sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit kanban_create auto-subscribes TUI/desktop sessions with platform="tui" and chat_id=HERMES_SESSION_KEY, and tools/kanban_tools.py documents that the TUI notification poller (tui_gateway/server.py) reads kanban_notify_subs and posts completion messages into the running session — but that reader was never implemented. The poller only watched process_registry completion events, and the gateway notifier skips "tui" rows because no such messaging adapter exists. Result: subscriptions accumulate with last_event_id=0 forever and no task event is ever delivered (18 subs, 29 terminal events, 0 deliveries in the report). Implement the missing delivery path in the TUI notification poller: - every 5s, claim unseen terminal events for this session's platform="tui" subscriptions via claim_unseen_events_for_sub — the same atomic cursor-claim the gateway notifier uses, so an event is delivered exactly once even with a gateway polling the same board DB - format events with the same wording as the gateway notifier (done/blocked/gave up/crashed/timed out/status; archived and unblocked are claimed but silent, so they can't wedge the cursor) - emit a status.update for user visibility, then chain an agent turn when the session is idle — mirroring process-completion handling; claimed events buffer in the session until it goes idle since the cursor (unlike the process queue) cannot re-queue - unsubscribe only at a truly final task status (done/archived), matching the gateway rule so respawned tasks keep notifying - multi-board: iterate boards, polling each resolved DB path once Fixes #59890 --- .../tui_gateway/test_kanban_notify_poller.py | 140 +++++++++++++ tui_gateway/server.py | 197 ++++++++++++++++++ 2 files changed, 337 insertions(+) create mode 100644 tests/tui_gateway/test_kanban_notify_poller.py diff --git a/tests/tui_gateway/test_kanban_notify_poller.py b/tests/tui_gateway/test_kanban_notify_poller.py new file mode 100644 index 000000000000..fb5896a4ca96 --- /dev/null +++ b/tests/tui_gateway/test_kanban_notify_poller.py @@ -0,0 +1,140 @@ +"""Tests for the TUI-side kanban notification poller (issue #59890). + +``kanban_create`` auto-subscribes TUI/desktop sessions with +``platform="tui"`` / ``chat_id=HERMES_SESSION_KEY``, but no component ever +read those rows back: the gateway notifier skips them (no "tui" messaging +adapter) and the TUI notification poller only watched process completions. +``last_event_id`` stayed 0 forever and no notification was ever delivered. + +These tests cover the delivery half that now lives in tui_gateway/server.py: +``_collect_kanban_notifications`` (cursor claim + formatting + terminal +unsubscribe) and ``_format_kanban_event_text``. +""" + +from types import SimpleNamespace + +from hermes_cli import kanban_db as kb +from tui_gateway.server import ( + _collect_kanban_notifications, + _format_kanban_event_text, +) + +SESSION_KEY = "tui-session-key-1" + + +def _session(key: str = SESSION_KEY) -> dict: + return {"session_key": key} + + +def _create_subscribed_task(*, chat_id: str = SESSION_KEY, platform: str = "tui"): + conn = kb.connect() + try: + tid = kb.create_task(conn, title="notify tui", assignee="worker") + kb.add_notify_sub(conn, task_id=tid, platform=platform, chat_id=chat_id) + return tid + finally: + conn.close() + + +def _complete(tid: str, summary: str = "all done") -> None: + conn = kb.connect() + try: + kb.complete_task(conn, tid, summary=summary) + finally: + conn.close() + + +def _sub_rows(tid: str) -> list: + conn = kb.connect() + try: + return kb.list_notify_subs(conn, task_id=tid) + finally: + conn.close() + + +class TestCollectKanbanNotifications: + def test_delivers_completed_event_and_unsubscribes(self): + tid = _create_subscribed_task() + _complete(tid, summary="shipped the fix") + + texts = _collect_kanban_notifications(_session()) + + assert len(texts) == 1 + assert tid in texts[0] + assert "done" in texts[0] + assert "shipped the fix" in texts[0] + # Task is at a final status -> subscription removed. + assert _sub_rows(tid) == [] + + def test_claim_advances_cursor_so_second_poll_is_empty(self): + tid = _create_subscribed_task() + conn = kb.connect() + try: + kb.block_task(conn, tid, reason="waiting on review") + finally: + conn.close() + + first = _collect_kanban_notifications(_session()) + second = _collect_kanban_notifications(_session()) + + assert len(first) == 1 + assert "blocked" in first[0] + assert "waiting on review" in first[0] + assert second == [] + # Blocked is not a final status -> subscription stays alive so a + # respawned task's next terminal event still reaches the user. + assert len(_sub_rows(tid)) == 1 + + def test_ignores_other_sessions_and_platforms(self): + tid_other_session = _create_subscribed_task(chat_id="some-other-session") + tid_gateway = _create_subscribed_task(platform="telegram", chat_id="chat-1") + _complete(tid_other_session) + _complete(tid_gateway) + + texts = _collect_kanban_notifications(_session()) + + assert texts == [] + # Foreign subscriptions untouched: cursors still 0, rows still there. + for tid in (tid_other_session, tid_gateway): + rows = _sub_rows(tid) + assert len(rows) == 1 + assert rows[0]["last_event_id"] == 0 + + def test_no_session_key_is_a_noop(self): + tid = _create_subscribed_task() + _complete(tid) + + assert _collect_kanban_notifications({"session_key": ""}) == [] + assert _collect_kanban_notifications({"session_key": None}) == [] + assert len(_sub_rows(tid)) == 1 + + +class TestFormatKanbanEventText: + SUB = {"task_id": "t_abc123"} + TASK = SimpleNamespace(title="build the thing", assignee="worker", result=None) + + def test_silent_kinds_return_none(self): + for kind in ("archived", "unblocked"): + ev = SimpleNamespace(kind=kind, payload={}) + assert _format_kanban_event_text(self.SUB, self.TASK, ev, "main") is None + + def test_blocked_includes_reason(self): + ev = SimpleNamespace(kind="blocked", payload={"reason": "needs creds"}) + text = _format_kanban_event_text(self.SUB, self.TASK, ev, "main") + assert "t_abc123" in text + assert "blocked" in text + assert "needs creds" in text + assert "[main]" in text + assert "@worker" in text + + def test_completed_prefers_payload_summary(self): + ev = SimpleNamespace(kind="completed", payload={"summary": "first line\nsecond"}) + text = _format_kanban_event_text(self.SUB, self.TASK, ev, "") + assert "done" in text + assert "first line" in text + assert "second" not in text + + def test_timed_out_with_bad_payload_does_not_raise(self): + ev = SimpleNamespace(kind="timed_out", payload={"limit_seconds": "not-a-number"}) + text = _format_kanban_event_text(self.SUB, self.TASK, ev, "") + assert "timed out" in text diff --git a/tui_gateway/server.py b/tui_gateway/server.py index daf6ef4597b4..10ea2538a631 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -11033,6 +11033,158 @@ def _notification_event_dedup_key(evt: dict) -> tuple: return (evt_sid, evt_type) +# Mirror gateway/kanban_watchers.py TERMINAL_KINDS: claim silent kinds too so +# the cursor advances past them and they can't wedge a later completed/blocked +# event behind an unclaimed row. +_KANBAN_NOTIFY_KINDS = ( + "completed", "blocked", "gave_up", "crashed", "timed_out", + "status", "archived", "unblocked", +) +_KANBAN_SILENT_KINDS = frozenset({"archived", "unblocked"}) +_KANBAN_POLL_SECONDS = 5.0 + + +def _format_kanban_event_text(sub: dict, task, ev, board_slug: str) -> Optional[str]: + """Single-line notification text for one kanban event. + + Wording mirrors the gateway notifier (gateway/kanban_watchers.py) so a + task completion reads the same in the TUI as it does on Telegram. + Returns None for kinds that are claimed but intentionally silent. + """ + kind = getattr(ev, "kind", "") + if not kind or kind in _KANBAN_SILENT_KINDS: + return None + task_id = sub.get("task_id", "") + title = (getattr(task, "title", None) or task_id)[:120] + board_tag = f"[{board_slug}] " if board_slug else "" + who = getattr(task, "assignee", None) or "" + tag = f"@{who} " if who else "" + payload = getattr(ev, "payload", None) or {} + if kind == "completed": + handoff = "" + summary = payload.get("summary") + if summary: + lines = str(summary).strip().splitlines() + handoff = f"\n{lines[0][:200]}" if lines else "" + elif getattr(task, "result", None): + lines = str(task.result).strip().splitlines() + handoff = f"\n{lines[0][:160]}" if lines else "" + return f"✔ {board_tag}{tag}Kanban {task_id} done — {title}{handoff}" + if kind == "blocked": + reason = f": {str(payload.get('reason'))[:160]}" if payload.get("reason") else "" + return f"⏸ {board_tag}{tag}Kanban {task_id} blocked{reason}" + if kind == "gave_up": + err = f"\n{str(payload.get('error'))[:200]}" if payload.get("error") else "" + return f"✖ {board_tag}{tag}Kanban {task_id} gave up after repeated spawn failures{err}" + if kind == "crashed": + return f"✖ {board_tag}{tag}Kanban {task_id} worker crashed (pid gone); dispatcher will retry" + if kind == "timed_out": + limit = 0 + try: + limit = int(payload.get("limit_seconds") or 0) + except (TypeError, ValueError): + pass + return f"⏱ {board_tag}{tag}Kanban {task_id} timed out (max_runtime={limit}s); will retry" + if kind == "status": + return f"🔄 {board_tag}{tag}Kanban {task_id} → {payload.get('status') or ''}" + return None + + +def _collect_kanban_notifications(session: dict) -> list: + """Claim unseen terminal kanban events for this TUI session's subscriptions. + + ``kanban_create`` auto-subscribes TUI/desktop sessions with + ``platform="tui"`` and ``chat_id=HERMES_SESSION_KEY`` (see + tools/kanban_tools.py ``_maybe_auto_subscribe``). The gateway notifier + can't deliver those — there is no "tui" messaging adapter — so this + poller is the delivery path for them (issue #59890). Uses the same + atomic cursor-claim (``claim_unseen_events_for_sub``) as the gateway + notifier, so a subscription is delivered exactly once even if a gateway + and a TUI poll the same board DB. + + Returns the list of formatted notification texts (may be empty). + """ + session_key = str(session.get("session_key") or "") + if not session_key or session.get("_finalized"): + return [] + try: + from hermes_cli import kanban_db as _kb + except Exception: + return [] + texts: list = [] + try: + boards = _kb.list_boards(include_archived=False) + except Exception: + try: + boards = [_kb.read_board_metadata(_kb.DEFAULT_BOARD)] + except Exception: + return [] + # Poll each resolved DB path once — multiple slugs can point at the same + # DB when HERMES_KANBAN_DB pins the board path (same guard as the gateway + # notifier). + seen_db_paths: set = set() + for board_meta in boards: + slug = (board_meta or {}).get("slug") or _kb.DEFAULT_BOARD + db_path = (board_meta or {}).get("db_path") + try: + resolved = ( + str(Path(db_path).expanduser().resolve()) + if db_path else str(_kb.kanban_db_path(slug).resolve()) + ) + except Exception: + resolved = f"slug:{slug}" + if resolved in seen_db_paths: + continue + seen_db_paths.add(resolved) + try: + conn = _kb.connect(board=slug) + except Exception: + continue + try: + try: + subs = _kb.list_notify_subs(conn) + except Exception: + continue + for sub in subs: + if (sub.get("platform") or "").lower() != "tui": + continue + if sub.get("chat_id") != session_key: + continue + _old, _new, 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=_KANBAN_NOTIFY_KINDS, + ) + if not events: + continue + task = _kb.get_task(conn, sub["task_id"]) + for ev in events: + text = _format_kanban_event_text(sub, task, ev, slug) + if text: + texts.append(text) + # Unsubscribe only at a truly final status (done/archived); + # blocked/crashed subs stay live so a respawned task's next + # terminal event still reaches the user (same rule as the + # gateway notifier). + if task and getattr(task, "status", "") in {"done", "archived"}: + try: + _kb.remove_notify_sub( + conn, + task_id=sub["task_id"], + platform=sub["platform"], + chat_id=sub["chat_id"], + thread_id=sub.get("thread_id") or "", + ) + except Exception: + pass + finally: + conn.close() + return texts + + def _notification_poller_loop( stop_event: threading.Event, sid: str, session: dict ) -> None: @@ -11045,11 +11197,56 @@ def _notification_poller_loop( The completion_queue is process-global. In multi-session Desktop each poller requeues events owned by another live session and drops addressed events whose owner is gone; ownerless legacy notifications remain global. + + Also polls ``kanban_notify_subs`` every ``_KANBAN_POLL_SECONDS`` for this + session's TUI kanban subscriptions and delivers terminal task events the + same way (status.update + agent turn) — the delivery path + tools/kanban_tools.py documents for platform="tui" rows (issue #59890). """ from tools.process_registry import process_registry, format_process_notification _emitted = set() # dedup re-queued events so same completion isn't emitted 50 times while session is busy + _last_kanban_poll = 0.0 while not stop_event.is_set() and not session.get("_finalized"): + _now = time.monotonic() + if _now - _last_kanban_poll >= _KANBAN_POLL_SECONDS: + _last_kanban_poll = _now + try: + _kanban_texts = _collect_kanban_notifications(session) + except Exception as _kb_exc: + print( + f"[tui_gateway] kanban notification poll failed: " + f"{type(_kb_exc).__name__}: {_kb_exc}", + file=sys.stderr, + ) + _kanban_texts = [] + if _kanban_texts: + for _kb_text in _kanban_texts: + _emit("status.update", sid, {"kind": "process", "text": _kb_text}) + # Events are cursor-claimed (never re-queued), so buffer them + # until the session is idle instead of dropping the agent turn. + session.setdefault("_kanban_pending", []).extend(_kanban_texts) + _pending = session.get("_kanban_pending") or [] + if _pending: + _batch: list = [] + with session["history_lock"]: + if not session.get("running"): + session["running"] = True + _batch = list(_pending) + session["_kanban_pending"] = [] + if _batch: + rid = f"__notif__{int(time.time() * 1000)}" + try: + _emit("message.start", sid) + _run_prompt_submit(rid, sid, session, "\n".join(_batch)) + except Exception as exc: + print( + f"[tui_gateway] kanban notification dispatch failed: " + f"{type(exc).__name__}: {exc}", + file=sys.stderr, + ) + with session["history_lock"]: + session["running"] = False try: evt = process_registry.completion_queue.get(timeout=0.5) except Exception: