fix(kanban): preserve durable origins for worker-created tasks

Carry the owning task's notification subscriptions independently of dependency
edges, within the creation transaction. Prefer its durable session over worker
and request-local sessions while preserving explicit overrides. Cover worker
CLI create and built-in decomposition, and retain conversation route anchors.
Auto-subscribe no longer upgrades an inherited passive subscription.

Slim adaptation of Christopher-Schulze's session-precedence fix in #85687,
expanded to durable subscription provenance and sibling creation paths.
Related: #85575, #85687

Validation: strict RED/GREEN (7 failing cases before; 7 passing after), then
58 Kanban test files: 383 passed, 2 skipped. Real dispatcher-spawn subprocess
probe covers direct, linked, unlinked, explicit-session, worker CLI, built-in
children and a plain CLI negative control, with recording transport only.

Co-authored-by: Christopher <210261288+Christopher-Schulze@users.noreply.github.com>

hermes_cli/kanban.py

@@ -341,6 +341,8 @@ def _cmd_assignees(args: argparse.Namespace) -> int:
 
 
 def _cmd_create(args: argparse.Namespace) -> int:
+    from agent.delegation_context import is_dispatcher_owned_worker_context
+
     try:
         ws_kind, ws_path = _parse_workspace_flag(args.workspace)
         branch_name = _parse_branch_flag(getattr(args, "branch", None))
@@ -371,6 +373,8 @@ def _cmd_create(args: argparse.Namespace) -> int:
             goal_max_turns=getattr(args, "goal_max_turns", None),
             completion_contract=getattr(args, "completion_contract", None),
             initial_status=getattr(args, "initial_status", "running"),
+            creator_task_id=(os.environ.get("HERMES_KANBAN_TASK")
+                             if is_dispatcher_owned_worker_context() else None),
         )
         task = kb.get_task(conn, task_id)
     if getattr(args, "json", False):

hermes_cli/kanban_db.py

@@ -1229,6 +1229,7 @@ def create_task(
     goal_mode: bool = False, goal_max_turns: Optional[int] = None, initial_status: str = "running",
     session_id: Optional[str] = None, board: Optional[str] = None, project_id: Optional[str] = None,
     project_source_task_id: Optional[str] = None,
+    creator_task_id: Optional[str] = None,
     completion_contract: Optional[str] = None,
 ) -> str:
     """Create a task (optionally under ``parents``); returns its id.
@@ -1239,10 +1240,12 @@ def create_task(
     instead of a duplicate. ``max_runtime_seconds``: cap before the dispatcher
     SIGTERMs and re-queues. ``model_override``/``provider_override`` pin the
     worker model (provider requires model); ``reasoning_effort`` is independent.
+    ``creator_task_id``: inherit durable session/subscriptions independently of
+    dependency edges; an explicit ``session_id`` still wins.
     ``project_source_task_id``: cross-profile fallback when ``project_id`` is not
     in the active profile's projects.db — see ``_resolve_project_link``.
     """
-    from hermes_cli.kanban_db_graph import initial_task_state
+    from hermes_cli.kanban_db_graph import initial_task_state, inherit_creator_origin
     from hermes_cli.kanban_pr_acceptance import validate_contract
 
     completion_contract = validate_contract(completion_contract)
@@ -1345,6 +1348,7 @@ def create_task(
                         "assignee": assignee,
                         "status": task_status,
                         "parents": list(parents),
+                        "creator_task_id": creator_task_id,
                         "tenant": tenant,
                         "workspace_kind": workspace_kind,
                         "workspace_path": workspace_path,
@@ -1357,6 +1361,7 @@ def create_task(
                     },
                 )
                 # ACK-edge: the originating channel hears a child BLOCK, not just the fan-in.
+                inherit_creator_origin(conn, task_id, creator_task_id, created_at=now)
                 _inherit_notify_subs(conn, task_id, parents, created_at=now)
             return task_id
         except sqlite3.IntegrityError:

hermes_cli/kanban_db_graph.py

@@ -5,6 +5,23 @@ import sqlite3
 import time
 from typing import Any, Optional
 
+def inherit_creator_origin(
+    conn: sqlite3.Connection, task_id: str, creator_task_id: Optional[str], *,
+    created_at: int,
+) -> None:
+    """Copy durable origin inside creation's transaction, never adding dependencies."""
+    if not creator_task_id:
+        return
+    from hermes_cli.kanban_db import _inherit_notify_subs
+
+    conn.execute(
+        "UPDATE tasks SET session_id = COALESCE(session_id, "
+        "(SELECT session_id FROM tasks WHERE id = ?)) WHERE id = ?",
+        (creator_task_id, task_id),
+    )
+    _inherit_notify_subs(conn, task_id, (creator_task_id,), created_at=created_at)
+
+
 def initial_task_state(
     conn: sqlite3.Connection, parents: tuple[str, ...], initial_status: str,
     triage: bool, tenant: Optional[str],
@@ -164,7 +181,7 @@ def _insert_decomposed_child(
     ``<repo>/.worktrees/<child-id>`` per child from the board anchor.
     """
     from hermes_cli.kanban_db import (
-        _new_task_id, _canonical_assignee, _append_event, _inherit_notify_subs,
+        _new_task_id, _canonical_assignee, _append_event,
     )
 
     root_ws_kind = root_row["workspace_kind"] or "scratch"
@@ -193,5 +210,5 @@ def _insert_decomposed_child(
     _append_event(
         conn, new_id, "created", {"by": author or "decomposer", "from_decompose_of": root_id},
     )
-    _inherit_notify_subs(conn, new_id, (root_id,), created_at=now)
+    inherit_creator_origin(conn, new_id, root_id, created_at=now)
     return new_id

tests/hermes_cli/test_kanban_creator_origin.py

@@ -0,0 +1,39 @@
+"""All explicit creator paths share durable lineage without graph coupling."""
+import pytest
+
+
+@pytest.mark.parametrize("surface", ["db", "builtin", "cli"])
+def test_creator_origin_survives_without_dependency_parent(tmp_path, monkeypatch, capsys, surface):
+    from hermes_cli import kanban_db as kb, kanban_db_connect as kbc, kanban_db_notify as kn
+    from hermes_cli.kanban_db_graph import decompose_triage_task
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    kb.init_db()
+    with kbc.connect_closing() as conn:
+        owner = kb.create_task(conn, title="owner", session_id="durable", triage=True)
+        kn.add_notify_sub(conn, task_id=owner, platform="telegram", chat_id="chat",
+                         delivery_mode="wake", notifier_profile="default")
+        if surface == "builtin":
+            tid = decompose_triage_task(conn, owner, root_assignee="default",
+                                       children=[{"title": "child"}])[0]
+        elif surface == "db":
+            tid = kb.create_task(conn, title="child", creator_task_id=owner)
+        else:
+            import json
+            import argparse
+            from hermes_cli.kanban import kanban_command
+            from hermes_cli.kanban_parser import build_parser
+            parser = argparse.ArgumentParser()
+            build_parser(parser.add_subparsers())
+            monkeypatch.setenv("HERMES_KANBAN_TASK", owner)
+            assert kanban_command(parser.parse_args(["kanban", "create", "child", "--json"])) == 0
+            tid = json.loads(capsys.readouterr().out)["id"]
+        assert kb.get_task(conn, tid).session_id == "durable"
+        subs = kn.list_notify_subs(conn, tid)
+        assert len(subs) == 1 and subs[0]["delivery_mode"] == "wake"
+        assert not conn.execute("SELECT 1 FROM task_links WHERE child_id = ?", (tid,)).fetchone()
+        # No ambient identity guessing in the storage API.
+        plain = kb.create_task(conn, title="plain", session_id="explicit")
+        assert kb.get_task(conn, plain).session_id == "explicit"
+        assert not kn.list_notify_subs(conn, plain)

tests/tools/test_kanban_provenance.py

@@ -0,0 +1,61 @@
+"""Worker provenance is not a dependency edge or a transient runtime session."""
+import json
+
+import pytest
+
+
+@pytest.mark.parametrize("linked,explicit", [(False, None), (True, None), (False, "override")])
+def test_worker_create_keeps_durable_origin(tmp_path, monkeypatch, linked, explicit):
+    from hermes_cli import kanban_db as kb, kanban_db_connect as kbc, kanban_db_notify as kn
+    from tools import kanban_tools as kt, async_delegation
+    from gateway.session_context import set_session_vars, clear_session_vars
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    kb.init_db()
+    with kbc.connect_closing() as conn:
+        owner = kb.create_task(conn, title="owner", session_id="durable")
+        kn.add_notify_sub(conn, task_id=owner, platform="discord", chat_id="chat",
+                          user_id="user", notifier_profile="default", delivery_mode="notify",
+                          delivery_metadata={"scope_id": "guild", "parent_chat_id": "forum"})
+        expected = kn.list_notify_subs(conn, owner)[0]
+    monkeypatch.setenv("HERMES_KANBAN_TASK", owner)
+    monkeypatch.setenv("HERMES_SESSION_ID", "ephemeral")
+    monkeypatch.setattr(async_delegation, "_current_origin_session_id", lambda: "api-origin")
+    # Even a matching current channel must not upgrade an inherited passive policy.
+    tokens = set_session_vars(platform="discord", chat_id="chat", profile="default")
+    try:
+        result = json.loads(kt._handle_create(dict(title="child", assignee="default",
+                            parents=[owner] if linked else [], session_id=explicit)))
+    finally:
+        clear_session_vars(tokens)
+    assert result["ok"], result
+    with kbc.connect_closing() as conn:
+        child = kb.get_task(conn, result["task_id"])
+        assert child.session_id == (explicit or "durable")
+        subs = kn.list_notify_subs(conn, child.id)
+        assert len(subs) == 1
+        for key in ("platform", "chat_id", "user_id", "delivery_mode", "delivery_metadata", "notifier_profile"):
+            assert subs[0][key] == expected[key]
+        assert bool(conn.execute("SELECT 1 FROM task_links WHERE child_id = ?", (child.id,)).fetchone()) == linked
+
+
+def test_tool_subscription_captures_conversation_anchors(tmp_path, monkeypatch):
+    from hermes_cli import kanban_db as kb, kanban_db_connect as kbc, kanban_db_notify as kn
+    from tools import kanban_tools as kt
+    from gateway.session_context import set_session_vars, clear_session_vars
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    kb.init_db()
+    tokens = set_session_vars(platform="discord", chat_id="thread", chat_type="thread",
+                             scope_id="guild", parent_chat_id="forum", profile="default")
+    try:
+        result = json.loads(kt._handle_create(dict(title="direct", assignee="default")))
+    finally:
+        clear_session_vars(tokens)
+    assert result["ok"], result
+    with kbc.connect_closing() as conn:
+        metadata = kn.list_notify_subs(conn, result["task_id"])[0]["delivery_metadata"]
+        assert metadata["scope_id"] == "guild"
+        assert metadata["parent_chat_id"] == "forum"

tools/kanban_tools.py

@@ -810,12 +810,6 @@ def _handle_create(args: dict, **kw) -> str:
     assignee = args.get("assignee")
     _check(assignee, "assignee is required — name the profile that should execute this "
                      "task (the dispatcher will only spawn tasks with an assignee)")
-    # Prefer the request-scoped api_server origin binding over HERMES_SESSION_ID: the env
-    # var is clobbered with a subagent's internal id whenever a child agent is constructed
-    # in-process, which would stamp — and later wake — the wrong session.
-    from tools.async_delegation import _current_origin_session_id
-    session_id = (args.get("session_id") or _current_origin_session_id()
-                  or os.environ.get("HERMES_SESSION_ID"))
     # Workspace sharing is always explicit: omitted fields mean a fresh scratch workspace
     # even for a dispatcher-spawned creator (reusing the parent's path would let a child
     # mutate review evidence or race its checkout). Project identity is the one safe thing
@@ -831,9 +825,14 @@ def _handle_create(args: dict, **kw) -> str:
     _check(model_override or not provider_override, "'provider' requires 'model' to be set as well")
     parents = _coerce_str_list(args.get("parents") or [], "parents", "task ids")
     with _board(args.get("board")) as (kb, conn):
+        from tools.async_delegation import _current_origin_session_id
+        self_tid = (os.environ.get("HERMES_KANBAN_TASK")
+                    if _is_dispatcher_owned_worker() else None)
+        self_task = kb.get_task(conn, self_tid) if self_tid else None
+        # The worker/API runtime may be transient; the owning task's origin is durable.
+        session_id = (args.get("session_id") or (self_task.session_id if self_task else None)
+                      or _current_origin_session_id() or os.environ.get("HERMES_SESSION_ID"))
         if project_id is None and workspace_kind is None and workspace_path is None:
-            self_tid = os.environ.get("HERMES_KANBAN_TASK")
-            self_task = kb.get_task(conn, self_tid) if self_tid else None
             if self_task is not None and self_task.project_id:
                 project_id, project_source_task_id = self_task.project_id, self_task.id
         new_tid = kb.create_task(
@@ -843,6 +842,7 @@ def _handle_create(args: dict, **kw) -> str:
             workspace_kind=str(workspace_kind if workspace_kind is not None else "scratch"),
             workspace_path=workspace_path, project_id=project_id,
             project_source_task_id=project_source_task_id, triage=triage,
+            creator_task_id=self_tid,
             idempotency_key=args.get("idempotency_key"),
             max_runtime_seconds=_opt_int(args.get("max_runtime_seconds")), skills=skills,
             model_override=model_override, provider_override=provider_override,
@@ -878,7 +878,11 @@ def _resolve_notify_target() -> Optional[dict[str, Any]]:
         except Exception:
             notifier_profile = "default"
     delivery_metadata: dict[str, Any] = {
-        k: v for k, v in (("thread_id", thread_id), ("chat_type", chat_type)) if v}
+        k: v for k, v in (
+            ("thread_id", thread_id), ("chat_type", chat_type),
+            ("scope_id", env("HERMES_SESSION_SCOPE_ID", "")),
+            ("parent_chat_id", env("HERMES_SESSION_PARENT_CHAT_ID", "")),
+        ) if v}
     if (platform.lower() == "telegram" and thread_id
             and (chat_type or "").lower() in {"dm", "direct", "private"}):
         delivery_metadata["telegram_dm_topic_reply_fallback"] = True
@@ -910,8 +914,13 @@ def _maybe_auto_subscribe(conn: Any, task_id: str) -> bool:
         target = _resolve_notify_target()
         if target is None:
             return False  # CLI / cron / test — no persistent channel
-        from hermes_cli import kanban_db as _kb
         from hermes_cli import kanban_db_notify as _kbn
+        # Inheritance and explicit subscriptions already encode the delivery policy.
+        # Auto-subscribe must not turn a passive destination into an agent wake.
+        if any(sub["platform"] == target["platform"] and sub["chat_id"] == target["chat_id"]
+               and (sub["thread_id"] or "") == (target["thread_id"] or "")
+               for sub in _kbn.list_notify_subs(conn, task_id)):
+            return True
         _kbn.add_notify_sub(conn, task_id=task_id, **target)
         return True
     except Exception as _exc:

website/docs/user-guide/features/kanban.md

@@ -952,6 +952,21 @@ bot> ✓ t_9fc1a3 completed by transcriber
 
 Subscriptions survive a task reaching `done` — completion is reversible (a reviewer or controller can reopen a done task), so the origin session keeps getting notified through reopen cycles. They auto-remove on `archived` (the irreversible end state). On boards that never archive, a GC sweep purges subscriptions for tasks that have sat in `done` or `blocked` with no new activity for `kanban.done_sub_retention_days` days (default 30; set 0 to disable), so stale rows don't accumulate forever. If you script a create with `--json` (machine output) the auto-subscribe is skipped — the assumption is that scripted callers want to manage subscriptions explicitly via `/kanban notify-subscribe`.
 
+Dispatcher workers creating tasks through `kanban_create` or `hermes kanban create`
+copy the owning task's durable notification subscriptions even without `parents`
+dependency links. Destinations, route anchors, and delivery modes are preserved;
+a passive subscription is not upgraded to a wake by auto-subscribe. This copies
+existing subscriptions independently of `auto_subscribe_on_create`, which controls
+adding the current conversation as a new destination. No destination is invented
+for a bare CLI session or a worker whose owning task has no subscriptions.
+
+For `kanban_create`, session lineage resolves in this order: explicit `session_id`,
+the owning worker task's durable session, request-scoped API origin, then the
+current process session. Built-in decomposition also inherits its root's durable
+session. Session lineage is not itself a notification destination: changing
+`session_id` does not replace existing subscriptions; use `notify-subscribe` and
+`notify-unsubscribe` to change where events are delivered.
+
 A chat-originated auto-subscribe is created in `notify+wake` mode: on a terminal event the destination agent both receives the passive message **and** takes a real turn, so it can read the board context and reply in its own voice. See [Delivery modes](#delivery-modes) below.
 
 ### Output truncation in messaging