From 44f19dfe78d2ef3c5b61fcb28fbc173356611aaf Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 3 Sep 2026 00:09:20 -0700 Subject: [PATCH] refactor(tui): per-board kanban poll owns probe/connect/close; dedupe boards via dict --- tui_gateway/session_notifications.py | 77 +++++++++++++--------------- 1 file changed, 37 insertions(+), 40 deletions(-) diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py index 14f1133426..d4161bb950 100644 --- a/tui_gateway/session_notifications.py +++ b/tui_gateway/session_notifications.py @@ -273,29 +273,43 @@ def _kb_board_key(_kb, board_meta) -> tuple[str, str]: return slug, f"slug:{slug}" -def _kb_poll_board(_kb, conn, slug: str, session_key: str) -> list: - """Claim + format this session's unseen events on one open board connection.""" +def _kb_poll_board(_kb, slug: str, session_key: str) -> list: + """Claim + format this session's unseen events on one board. One poller per live session: the board + is not opened writable unless it has a subscription owned by this exact session (if the read-only + probe fails — locked/corrupt DB — delivery is preserved by falling through).""" try: - subs = _kb.list_notify_subs(conn) + if _kb.count_notify_subs(board=slug, platform="tui", chat_id=session_key) == 0: + return [] + except Exception: + pass + try: + conn = _kb.connect(board=slug) except Exception: return [] texts: list = [] - for sub in subs: - if (sub.get("platform") or "").lower() != "tui" or sub.get("chat_id") != session_key: - continue - sub_ident = dict( - task_id=sub["task_id"], platform=sub["platform"], chat_id=sub["chat_id"], thread_id=sub.get("thread_id") or "" - ) - _old, _new, events = _kb.claim_unseen_events_for_sub(conn, kinds=_KANBAN_NOTIFY_KINDS, **sub_ident) - if not events: - continue - task = _kb.get_task(conn, sub["task_id"]) - texts.extend(t for t in (_format_kanban_event_text(sub, task, ev, slug) for ev in events) if t) - # Unsubscribe only on archive: ``done`` is reversible in review/controller flows, so keeping the - # sub lets a later reopen notify the same session. The claimed cursor prevents replay. - if task and getattr(task, "status", "") == "archived": - with contextlib.suppress(Exception): - _kb.remove_notify_sub(conn, **sub_ident) + try: + try: + subs = _kb.list_notify_subs(conn) + except Exception: + return [] + for sub in subs: + if (sub.get("platform") or "").lower() != "tui" or sub.get("chat_id") != session_key: + continue + sub_ident = dict( + task_id=sub["task_id"], platform=sub["platform"], chat_id=sub["chat_id"], thread_id=sub.get("thread_id") or "" + ) + _old, _new, events = _kb.claim_unseen_events_for_sub(conn, kinds=_KANBAN_NOTIFY_KINDS, **sub_ident) + if not events: + continue + task = _kb.get_task(conn, sub["task_id"]) + texts.extend(t for t in (_format_kanban_event_text(sub, task, ev, slug) for ev in events) if t) + # Unsubscribe only on archive: ``done`` is reversible in review/controller flows, so keeping the + # sub lets a later reopen notify the same session. The claimed cursor prevents replay. + if task and getattr(task, "status", "") == "archived": + with contextlib.suppress(Exception): + _kb.remove_notify_sub(conn, **sub_ident) + finally: + conn.close() return texts @@ -318,28 +332,11 @@ def _collect_kanban_notifications(session: dict) -> list: boards = [_kb.read_board_metadata(_kb.DEFAULT_BOARD)] except Exception: return [] - texts: list = [] - seen_db_paths: set = set() + # dict keyed by resolved DB identity: first slug per DB wins (a pinned HERMES_KANBAN_DB aliases slugs). + unique = {} for slug, resolved in (_kb_board_key(_kb, board_meta) for board_meta in boards): - if resolved in seen_db_paths: - continue - seen_db_paths.add(resolved) - # One poller per live session: don't open this board writable unless it has a subscription owned - # by this exact session. If the read-only probe fails (locked/corrupt DB), preserve delivery. - try: - if _kb.count_notify_subs(board=slug, platform="tui", chat_id=session_key) == 0: - continue - except Exception: - pass - try: - conn = _kb.connect(board=slug) - except Exception: - continue - try: - texts.extend(_kb_poll_board(_kb, conn, slug, session_key)) - finally: - conn.close() - return texts + unique.setdefault(resolved, slug) + return [t for slug in unique.values() for t in _kb_poll_board(_kb, slug, session_key)] def _notif_poll_kanban(sid: str, session: dict) -> None: