Skip to content

feat: board-level kanban notification subscriptions (ADR-0011) #28748

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
KulltivateAI wants to merge 1 commit into from
Closed

feat: board-level kanban notification subscriptions (ADR-0011) #28748

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 74 additions & 14 deletions gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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,

@teknium1 teknium1 Jun 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This supplies the real task object, but the message formatter below still interpolates sub['task_id']; for a board subscription that is the board sentinel, so the user-facing notification would not identify the actual task. Please render board events with ev.task_id or the loaded task id.

"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"]
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
)
Expand Down
71 changes: 58 additions & 13 deletions hermes_cli/kanban.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
Expand Down Expand Up @@ -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
Expand All @@ -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


Expand Down
Loading