diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index f0f31338aa..052c2fff59 100644 --- a/hermes_cli/kanban.py +++ b/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): diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 046681515f..210c6eed3c 100644 --- a/hermes_cli/kanban_db.py +++ b/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: diff --git a/hermes_cli/kanban_db_graph.py b/hermes_cli/kanban_db_graph.py index ea329ed7b2..e3a1f306c4 100644 --- a/hermes_cli/kanban_db_graph.py +++ b/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( ``/.worktrees/`` 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 diff --git a/tests/hermes_cli/test_kanban_creator_origin.py b/tests/hermes_cli/test_kanban_creator_origin.py new file mode 100644 index 0000000000..4402b201a8 --- /dev/null +++ b/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) diff --git a/tests/tools/test_kanban_provenance.py b/tests/tools/test_kanban_provenance.py new file mode 100644 index 0000000000..eeb69a0bbe --- /dev/null +++ b/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" diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index cd4943e221..b9ff4645ad 100644 --- a/tools/kanban_tools.py +++ b/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: diff --git a/website/docs/user-guide/features/kanban.md b/website/docs/user-guide/features/kanban.md index 4eda0372d8..413d355e58 100644 --- a/website/docs/user-guide/features/kanban.md +++ b/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