fix(kanban): inherit notify subs for child tasks

Copy gateway notification subscriptions from parent tasks to child tasks created by create_task(..., parents=...), link_tasks(), and decompose_triage_task().

Inherited subscriptions start at the child's current event cursor, so linking an existing child does not replay pre-link task events.

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).

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):
     """