diff --git a/gateway/run.py b/gateway/run.py index 9050dd741615..e5426c1417e7 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -4590,22 +4590,16 @@ def _collect(): logger.debug("kanban notifier: cannot open board %s: %s", slug, exc) continue try: - # `connect()` runs the schema + idempotent migration - # on first open per process, so an explicit - # `init_db()` here would be redundant. Worse: - # `init_db()` deliberately busts the per-process - # cache and re-runs the migration on a *second* - # connection, which races the first and used to - # log a benign but noisy `duplicate column name` - # traceback (and intermittent "database is locked" - # ā€” issue #21378) on every gateway start against - # a legacy DB. `_add_column_if_missing` now - # tolerates that race, but we still skip the - # redundant call to avoid the wasted work. subs = _kb.list_notify_subs(conn) if not subs: logger.debug("kanban notifier: board %s has no subscriptions", slug) - for sub in subs: + + # Split into regular (per-task) and board-level subs. + regular_subs = [s for s in subs if s.get("task_id") != _kb.BOARD_SUB_TASK_ID] + board_subs = [s for s in subs if s.get("task_id") == _kb.BOARD_SUB_TASK_ID] + + # --- Process regular (per-task) subs first --- + for sub in regular_subs: owner_profile = sub.get("notifier_profile") or None if owner_profile and owner_profile != notifier_profile: logger.debug( @@ -4643,11 +4637,63 @@ def _collect(): "task": task, "board": slug, }) + + # --- Process board-level subs --- + for sub in board_subs: + owner_profile = sub.get("notifier_profile") or None + if owner_profile and owner_profile != notifier_profile: + continue + platform = (sub.get("platform") or "").lower() + if platform not in active_platforms: + continue + # Parse the per-sub kinds filter (comma-separated + # string stored in the DB). Falls back to + # TERMINAL_KINDS when NULL/empty. + sub_kinds_raw = sub.get("kinds") or None + if sub_kinds_raw: + sub_kinds: tuple[str, ...] = tuple( + k.strip() for k in sub_kinds_raw.split(",") if k.strip() + ) + else: + sub_kinds = TERMINAL_KINDS + old_cursor, cursor, events = _kb.claim_unseen_board_events( + conn, + platform=sub["platform"], + chat_id=sub["chat_id"], + thread_id=sub.get("thread_id") or "", + kinds=sub_kinds, + ) + if not events: + continue + # Board subs produce one delivery dict per event + # so each gets its own task context for message + # formatting. We mark them as board subs for the + # delivery loop. + for ev in events: + ev_task = _kb.get_task(conn, ev.task_id) + deliveries.append({ + "sub": sub, + "old_cursor": old_cursor, + "cursor": cursor, + "events": [ev], + "task": ev_task, + "board": slug, + "_board_sub": True, + }) + logger.debug( + "kanban notifier: claimed %d board event(s) on board %s cursor %sā†’%s", + len(events), slug, old_cursor, cursor, + ) finally: conn.close() return deliveries deliveries = await asyncio.to_thread(_collect) + + # Track event keys delivered via per-task subs so board subs + # don't duplicate them to the same destination. + delivered_event_keys: set[tuple] = set() + for d in deliveries: sub = d["sub"] task = d["task"] @@ -4677,8 +4723,18 @@ def _collect(): ) continue title = (task.title if task else sub["task_id"])[:120] + is_board_sub = d.get("_board_sub", False) for ev in d["events"]: kind = ev.kind + # Dedup: if this is a board-level sub, skip events + # already delivered by a per-task sub to the same + # destination. + event_dest_key = ( + ev.id, sub["platform"], + sub["chat_id"], sub.get("thread_id") or "", + ) + if is_board_sub and event_dest_key in delivered_event_keys: + continue # Identity prefix: attribute terminal pings to the # worker that did the work. Makes fleets (where one # chat subscribes to many tasks) legible at a glance. @@ -4772,6 +4828,8 @@ def _collect(): ) # Reset the failure counter on success. sub_fail_counts.pop(sub_key, None) + # Track delivered events for board-sub dedup. + delivered_event_keys.add(event_dest_key) except Exception as exc: fails = sub_fail_counts.get(sub_key, 0) + 1 sub_fail_counts[sub_key] = fails @@ -4815,8 +4873,10 @@ def _collect(): # 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. + # Board-level subs never auto-unsubscribe ā€” they + # are permanent until explicitly removed. task_terminal = task and task.status in {"done", "archived"} - if task_terminal: + if task_terminal and not is_board_sub: await asyncio.to_thread( self._kanban_unsub, sub, board_slug, ) diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index 4e975bb3e8d7..eb6550315c6d 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -629,7 +629,12 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu help="Subscribe a gateway source to a task's terminal events " "(used by /kanban subscribe in the gateway adapter)", ) - p_nsub.add_argument("task_id") + p_nsub.add_argument("task_id", nargs="?", default=None, + help="Task ID to subscribe to (omit when using --board)") + p_nsub.add_argument( + "--board", action="store_true", default=False, + help="Subscribe to ALL tasks on the board (board-level subscription)", + ) p_nsub.add_argument("--platform", required=True) p_nsub.add_argument("--chat-id", required=True) p_nsub.add_argument("--thread-id", default=None) @@ -638,19 +643,29 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu "--notifier-profile", default=None, help="Profile gateway that owns/delivers this subscription (default: active profile)", ) + p_nsub.add_argument( + "--kinds", default=None, + help="Comma-separated event kinds to filter on (e.g. 'completed,blocked'). " + "Default: all terminal kinds.", + ) p_nlist = sub.add_parser( "notify-list", help="List notification subscriptions (optionally for a single task)", ) p_nlist.add_argument("task_id", nargs="?", default=None) + p_nlist.add_argument("--board", action="store_true", default=False, + help="List only board-level subscriptions") p_nlist.add_argument("--json", action="store_true") p_nrm = sub.add_parser( "notify-unsubscribe", help="Remove a gateway subscription from a task", ) - p_nrm.add_argument("task_id") + p_nrm.add_argument("task_id", nargs="?", default=None, + help="Task ID to unsubscribe from (omit when using --board)") + p_nrm.add_argument("--board", action="store_true", default=False, + help="Unsubscribe from board-level subscription") p_nrm.add_argument("--platform", required=True) p_nrm.add_argument("--chat-id", required=True) p_nrm.add_argument("--thread-id", default=None) @@ -2264,25 +2279,42 @@ def _cmd_stats(args: argparse.Namespace) -> int: def _cmd_notify_subscribe(args: argparse.Namespace) -> int: + board_flag = getattr(args, "board", False) + task_id = getattr(args, "task_id", None) + if not board_flag and not task_id: + print("error: either task_id or --board is required", file=sys.stderr) + return 2 + if board_flag and task_id: + print("error: --board and task_id are mutually exclusive", file=sys.stderr) + return 2 + kinds_str = getattr(args, "kinds", None) or None with kb.connect() as conn: - if kb.get_task(conn, args.task_id) is None: - print(f"no such task: {args.task_id}", file=sys.stderr) - return 1 + if board_flag: + effective_task_id = kb.BOARD_SUB_TASK_ID + else: + if kb.get_task(conn, task_id) is None: + print(f"no such task: {task_id}", file=sys.stderr) + return 1 + effective_task_id = task_id kb.add_notify_sub( - conn, task_id=args.task_id, + conn, task_id=effective_task_id, platform=args.platform, chat_id=args.chat_id, thread_id=args.thread_id, user_id=args.user_id, notifier_profile=args.notifier_profile or _profile_author(), + kinds=kinds_str, ) + target = "[BOARD]" if board_flag else effective_task_id + kinds_display = f" (kinds={kinds_str})" if kinds_str else "" print(f"Subscribed {args.platform}:{args.chat_id}" + (f":{args.thread_id}" if args.thread_id else "") - + f" to {args.task_id}") + + f" to {target}{kinds_display}") return 0 def _cmd_notify_list(args: argparse.Namespace) -> int: + board_flag = getattr(args, "board", False) with kb.connect() as conn: - subs = kb.list_notify_subs(conn, args.task_id) + subs = kb.list_notify_subs(conn, args.task_id, board_only=board_flag) if getattr(args, "json", False): print(json.dumps(subs, indent=2, ensure_ascii=False)) return 0 @@ -2292,22 +2324,35 @@ def _cmd_notify_list(args: argparse.Namespace) -> int: for s in subs: thr = f":{s['thread_id']}" if s.get("thread_id") else "" owner = f" owner={s['notifier_profile']}" if s.get("notifier_profile") else "" - print(f" {s['task_id']:10s} {s['platform']}:{s['chat_id']}{thr}" - f" (since event {s['last_event_id']}){owner}") + kinds_info = f" kinds={s['kinds']}" if s.get("kinds") else "" + is_board = s.get("task_id") == kb.BOARD_SUB_TASK_ID + task_label = "[BOARD] " if is_board else f"{s['task_id']:10s}" + print(f" {task_label} {s['platform']}:{s['chat_id']}{thr}" + f" (since event {s['last_event_id']}){owner}{kinds_info}") return 0 def _cmd_notify_unsubscribe(args: argparse.Namespace) -> int: + board_flag = getattr(args, "board", False) + task_id = getattr(args, "task_id", None) + if not board_flag and not task_id: + print("error: either task_id or --board is required", file=sys.stderr) + return 2 + if board_flag and task_id: + print("error: --board and task_id are mutually exclusive", file=sys.stderr) + return 2 + effective_task_id = kb.BOARD_SUB_TASK_ID if board_flag else task_id with kb.connect() as conn: ok = kb.remove_notify_sub( - conn, task_id=args.task_id, + conn, task_id=effective_task_id, platform=args.platform, chat_id=args.chat_id, thread_id=args.thread_id, ) + target = "[BOARD]" if board_flag else effective_task_id if not ok: - print("(no such subscription)", file=sys.stderr) + print(f"(no such subscription for {target})", file=sys.stderr) return 1 - print(f"Unsubscribed from {args.task_id}") + print(f"Unsubscribed from {target}") return 0 diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index edeae51707b0..74fba621db71 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -149,6 +149,7 @@ def _resolve_claim_ttl_seconds(ttl_seconds: Optional[int] = None) -> int: # --------------------------------------------------------------------------- DEFAULT_BOARD = "default" +BOARD_SUB_TASK_ID = "__board__" # Slug validator: lowercase alphanumerics, digits, hyphens; 1ā€“64 chars. # Strict enough to stop traversal (`..`) and embedded path separators, loose @@ -1200,6 +1201,10 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: _add_column_if_missing( conn, "kanban_notify_subs", "notifier_profile", "notifier_profile TEXT" ) + if "kinds" not in notify_cols: + _add_column_if_missing( + conn, "kanban_notify_subs", "kinds", "kinds TEXT" + ) # One-shot backfill: any task that is 'running' before runs existed # had its claim_lock / claim_expires / worker_pid on the task row. @@ -5680,18 +5685,26 @@ def add_notify_sub( thread_id: Optional[str] = None, user_id: Optional[str] = None, notifier_profile: Optional[str] = None, + kinds: Optional[str] = None, ) -> None: """Register a gateway source that wants terminal-state notifications - for ``task_id``. Idempotent on (task, platform, chat, thread).""" + for ``task_id``. Idempotent on (task, platform, chat, thread). + + For board-level subscriptions, caller passes + ``task_id=BOARD_SUB_TASK_ID`` (``"__board__"``). + + ``kinds`` is a comma-separated string of event kinds to filter on + (e.g. ``"completed,blocked"``). ``None`` means all terminal kinds. + """ now = int(time.time()) with write_txn(conn): conn.execute( """ INSERT OR IGNORE INTO kanban_notify_subs - (task_id, platform, chat_id, thread_id, user_id, notifier_profile, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?) + (task_id, platform, chat_id, thread_id, user_id, notifier_profile, kinds, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, - (task_id, platform, chat_id, thread_id or "", user_id, notifier_profile, now), + (task_id, platform, chat_id, thread_id or "", user_id, notifier_profile, kinds, now), ) if notifier_profile: # Self-heal legacy rows that predate notifier ownership by @@ -5709,8 +5722,13 @@ def add_notify_sub( def list_notify_subs( conn: sqlite3.Connection, task_id: Optional[str] = None, + *, board_only: bool = False, ) -> list[dict]: - if task_id is not None: + if board_only: + rows = conn.execute( + "SELECT * FROM kanban_notify_subs WHERE task_id = ?", (BOARD_SUB_TASK_ID,), + ).fetchall() + elif task_id is not None: rows = conn.execute( "SELECT * FROM kanban_notify_subs WHERE task_id = ?", (task_id,), ).fetchall() @@ -5836,6 +5854,74 @@ def claim_unseen_events_for_sub( return old_cursor, new_cursor, events +def claim_unseen_board_events( + conn: sqlite3.Connection, + *, + platform: str, + chat_id: str, + thread_id: Optional[str] = None, + kinds: Optional[Iterable[str]] = None, +) -> tuple[int, int, list[Event]]: + """Atomically claim unseen events across ALL tasks for a board-level sub. + + Same CAS pattern as :func:`claim_unseen_events_for_sub` but queries the + global ``task_events`` table without a ``task_id`` filter, and uses + ``task_id=BOARD_SUB_TASK_ID`` (``"__board__"``) for the subscription row. + + ``kinds`` filters events by ``kind IN (...)``. When ``None``, all event + kinds are returned. Results are capped at 200 rows per call to prevent + backfill runaway on first subscribe. + + Returns ``(old_cursor, new_cursor, events)``. + """ + task_id = BOARD_SUB_TASK_ID + with write_txn(conn): + row = conn.execute( + "SELECT last_event_id FROM kanban_notify_subs " + "WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ?", + (task_id, platform, chat_id, thread_id or ""), + ).fetchone() + if row is None: + return 0, 0, [] + old_cursor = int(row["last_event_id"]) + + kind_list = list(kinds) if kinds else None + q = ( + "SELECT * FROM task_events WHERE id > ? " + + ("AND kind IN (" + ",".join("?" * len(kind_list)) + ") " if kind_list else "") + + "ORDER BY id ASC LIMIT 200" + ) + params: list[Any] = [old_cursor] + if kind_list: + params.extend(kind_list) + rows = conn.execute(q, params).fetchall() + + out: list[Event] = [] + max_id = old_cursor + for r in rows: + try: + payload = json.loads(r["payload"]) if r["payload"] else None + except Exception: + payload = None + out.append(Event( + id=r["id"], task_id=r["task_id"], kind=r["kind"], + payload=payload, created_at=r["created_at"], + run_id=(int(r["run_id"]) if "run_id" in r.keys() and r["run_id"] is not None else None), + )) + max_id = max(max_id, int(r["id"])) + + if not out: + return old_cursor, old_cursor, [] + + conn.execute( + "UPDATE kanban_notify_subs SET last_event_id = ? " + "WHERE task_id = ? AND platform = ? AND chat_id = ? AND thread_id = ? " + "AND last_event_id = ?", + (int(max_id), task_id, platform, chat_id, thread_id or "", int(old_cursor)), + ) + return old_cursor, max_id, out + + def advance_notify_cursor( conn: sqlite3.Connection, *,