fix(tui): deliver kanban notify subscriptions to TUI/desktop sessions

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

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

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: