diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index f8735f55d0f6..4fcc4fd610ef 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -3177,6 +3177,7 @@ def create_task( "provider_override": provider_override, }, ) + _inherit_notify_subs(conn, task_id, parents, created_at=now) return task_id except sqlite3.IntegrityError: if attempt == 1: @@ -3199,6 +3200,47 @@ def _find_missing_parents(conn: sqlite3.Connection, parents: Iterable[str]) -> l return [p for p in parents if p not in present] +def _inherit_notify_subs( + conn: sqlite3.Connection, + child_id: str, + parents: Iterable[str], + *, + created_at: Optional[int] = None, +) -> None: + """Copy gateway notification subscriptions from parent tasks to a child. + + The inherited subscription starts caught up to the child's current event + cursor. This makes manual `link_tasks(parent, existing_child)` safe: the + parent chat receives future child terminal events without replaying the + child's pre-link history. + """ + parent_ids = tuple(dict.fromkeys(p for p in parents if p)) + if not parent_ids: + return + row = conn.execute( + "SELECT COALESCE(MAX(id), 0) AS cursor FROM task_events WHERE task_id = ?", + (child_id,), + ).fetchone() + cursor = int(row["cursor"] if row is not None else 0) + placeholders = ",".join("?" * len(parent_ids)) + conn.execute( + f""" + INSERT OR IGNORE INTO kanban_notify_subs + (task_id, platform, chat_id, thread_id, user_id, + notifier_profile, created_at, last_event_id) + SELECT ?, platform, chat_id, thread_id, user_id, notifier_profile, ?, ? + FROM kanban_notify_subs + WHERE task_id IN ({placeholders}) + """, + ( + child_id, + int(created_at if created_at is not None else time.time()), + cursor, + *parent_ids, + ), + ) + + def get_task(conn: sqlite3.Connection, task_id: str) -> Optional[Task]: row = conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone() return Task.from_row(row) if row else None @@ -3380,6 +3422,7 @@ def link_tasks(conn: sqlite3.Connection, parent_id: str, child_id: str) -> None: conn, child_id, "linked", {"parent": parent_id, "child": child_id}, ) + _inherit_notify_subs(conn, child_id, (parent_id,)) def _would_cycle(conn: sqlite3.Connection, parent_id: str, child_id: str) -> bool: @@ -6017,6 +6060,7 @@ def decompose_triage_task( conn, new_id, "created", {"by": author or "decomposer", "from_decompose_of": task_id}, ) + _inherit_notify_subs(conn, new_id, (task_id,), created_at=now) child_ids.append(new_id) # Link children to their sibling parents (within the decomposed graph). diff --git a/tests/hermes_cli/test_kanban_notify.py b/tests/hermes_cli/test_kanban_notify.py index 07b1fa201bac..15eb2671b249 100644 --- a/tests/hermes_cli/test_kanban_notify.py +++ b/tests/hermes_cli/test_kanban_notify.py @@ -26,6 +26,109 @@ def kanban_home(tmp_path, monkeypatch): return home +def _assert_inherited_notify_sub(subs: list[dict]) -> None: + assert len(subs) == 1 + assert subs[0]["platform"] == "telegram" + assert subs[0]["chat_id"] == "chat1" + assert subs[0]["thread_id"] == "topic1" + assert subs[0]["user_id"] == "user1" + assert subs[0]["notifier_profile"] == "default" + + +def test_create_task_inherits_parent_notify_subscriptions(kanban_home): + conn = kb.connect() + try: + parent = kb.create_task(conn, title="parent", assignee="worker1") + kb.add_notify_sub( + conn, + task_id=parent, + platform="telegram", + chat_id="chat1", + thread_id="topic1", + user_id="user1", + notifier_profile="default", + ) + + child = kb.create_task(conn, title="child", parents=[parent], assignee="worker1") + + subs = kb.list_notify_subs(conn, child) + finally: + conn.close() + + _assert_inherited_notify_sub(subs) + + +def test_link_tasks_inherits_parent_notify_subscriptions_without_replaying_old_child_events(kanban_home): + conn = kb.connect() + try: + parent = kb.create_task(conn, title="parent", assignee="worker1") + child = kb.create_task(conn, title="child", assignee="worker1") + with kb.write_txn(conn): + kb._append_event(conn, child, kind="blocked", payload={"reason": "old"}) + kb.add_notify_sub( + conn, + task_id=parent, + platform="telegram", + chat_id="chat1", + thread_id="topic1", + user_id="user1", + notifier_profile="default", + ) + + kb.link_tasks(conn, parent, child) + + subs = kb.list_notify_subs(conn, child) + _, old_events = kb.unseen_events_for_sub( + conn, + task_id=child, + platform="telegram", + chat_id="chat1", + thread_id="topic1", + kinds=["blocked"], + ) + finally: + conn.close() + + _assert_inherited_notify_sub(subs) + assert old_events == [] + + +def test_decompose_triage_task_inherits_root_notify_subscriptions(kanban_home): + conn = kb.connect() + try: + root = kb.create_task(conn, title="triage root", triage=True, assignee="orchestrator") + kb.add_notify_sub( + conn, + task_id=root, + platform="telegram", + chat_id="chat1", + thread_id="topic1", + user_id="user1", + notifier_profile="default", + ) + + child_ids = kb.decompose_triage_task( + conn, + root, + root_assignee="orchestrator", + children=[ + {"title": "first child", "assignee": "worker1"}, + {"title": "second child", "assignee": "worker2", "parents": [0]}, + ], + author="triager", + auto_promote=False, + ) + + assert child_ids is not None + child_subs = [kb.list_notify_subs(conn, child_id) for child_id in child_ids] + finally: + conn.close() + + assert len(child_subs) == 2 + for subs in child_subs: + _assert_inherited_notify_sub(subs) + + @pytest.mark.asyncio async def test_notifier_unsubs_after_completed_event(kanban_home): """