fix: keep Kanban worker scope out of descendant processes

Carry the existing write fence across Hermes-owned spawn boundaries without
dropping board routing or changing credential policy. Grant dispatcher and
managed tool runtimes explicit task scope; align CLI task mutations with tools.

Verify real shell/CLI descendants, dispatcher startup, and supervised stdio
transport against isolated SQLite boards. This is cooperative runtime scoping,
not OS confinement.

Refs #103974, #104058, #104904

agent/delegation_context.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 import os
 from contextlib import contextmanager
 from contextvars import ContextVar, Token
-from typing import Iterator, Mapping, MutableMapping
+from typing import Iterator, Mapping, MutableMapping, overload
 
 _DELEGATED_CHILD_CONTEXT: ContextVar[bool] = ContextVar("hermes_delegated_child_context", default=False)
 # Any in-process execution that is NOT the dispatcher-owned worker (cron jobs). Kept separate
@@ -20,8 +20,8 @@ _NON_DISPATCHER_OWNED_CONTEXT: ContextVar[bool] = ContextVar("hermes_non_dispatc
 DELEGATED_CHILD_ENV_MARKER = "HERMES_DELEGATED_CHILD_CONTEXT"
 
 KANBAN_ENV_KEYS: tuple[str, ...] = (
-    "HERMES_KANBAN_TASK", "HERMES_KANBAN_RUN_ID", "HERMES_KANBAN_WORKSPACE", "HERMES_KANBAN_WORKSPACES_ROOT",
-    "HERMES_KANBAN_CLAIM_LOCK", "HERMES_KANBAN_BOARD", "HERMES_KANBAN_DB",
+    "HERMES_KANBAN_TASK", "HERMES_KANBAN_RUN_ID", "HERMES_KANBAN_CLAIM_LOCK",
+    "HERMES_KANBAN_GOAL_MODE", "HERMES_KANBAN_GOAL_MAX_TURNS",
 )
 
 
@@ -70,7 +70,7 @@ def non_dispatcher_owned_context() -> Iterator[None]:
 
 def is_dispatcher_owned_worker_context() -> bool:
     """The single predicate every ``HERMES_KANBAN_*`` identity gate should use."""
-    return not (_DELEGATED_CHILD_CONTEXT.get() or _NON_DISPATCHER_OWNED_CONTEXT.get())
+    return not (is_delegated_child_process_context() or _NON_DISPATCHER_OWNED_CONTEXT.get())
 
 
 def is_delegated_child_process_context() -> bool:
@@ -79,17 +79,34 @@ def is_delegated_child_process_context() -> bool:
 
 
 def scrub_kanban_env(env: Mapping[str, str] | MutableMapping[str, str]) -> dict[str, str]:
-    """Return *env* with dispatcher-only Kanban variables removed and the lineage marker set."""
+    """Remove worker identity, retaining board/location and an inherited write fence.
+
+    TASK absence alone would promote a descendant to an orchestrator. The marker
+    survives later execs, including scripts that remove TASK themselves. This is
+    cooperative runtime scoping, not confinement of code with direct SQLite access.
+    """
     cleaned = {k: v for k, v in env.items() if k not in KANBAN_ENV_KEYS}
     cleaned[DELEGATED_CHILD_ENV_MARKER] = "1"
     return cleaned
 
 
+@overload
+def delegated_child_subprocess_env(env: Mapping[str, str]) -> dict[str, str]: ...
+
+
+@overload
+def delegated_child_subprocess_env(env: None = None) -> dict[str, str] | None: ...
+
+
 def delegated_child_subprocess_env(
     env: Mapping[str, str] | MutableMapping[str, str] | None = None,
 ) -> dict[str, str] | None:
-    """Env override only when delegated-child lineage must cross fork: preserves ``env=None``
-    inherit semantics for non-delegated calls; in a child, a scrubbed env carrying the marker."""
-    if not is_delegated_child_process_context():
+    """Carry worker/delegate descendant denial across a real process spawn.
+
+    Location and credentials are untouched; callers retain their existing secret policy.
+    Dispatcher workers and supervised tool transports grant their own explicit scope.
+    """
+    if not (is_delegated_child_process_context() or os.environ.get("HERMES_KANBAN_TASK")
+            or (env and (env.get("HERMES_KANBAN_TASK") or env.get(DELEGATED_CHILD_ENV_MARKER)))):
         return None if env is None else dict(env)
     return scrub_kanban_env(os.environ if env is None else env)

agent/lsp/client.py

@@ -206,6 +206,7 @@ class LSPClient:
             raise
 
     async def _spawn(self) -> None:
+        from agent.delegation_context import delegated_child_subprocess_env
         cmd = self._command
         if sys.platform == "win32" and cmd[0].lower().endswith((".cmd", ".bat")):
             cmd = ["cmd.exe", "/c", *cmd]  # CreateProcess can't run .cmd/.bat shims directly
@@ -217,7 +218,7 @@ class LSPClient:
             self._proc = await asyncio.create_subprocess_exec(
                 cmd[0], *cmd[1:],
                 stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
-                env={**os.environ, **(self._env or {})}, cwd=self._cwd,
+                env=delegated_child_subprocess_env({**os.environ, **(self._env or {})}), cwd=self._cwd,
                 start_new_session=True, creationflags=windows_hide_flags(),
             )
         except FileNotFoundError as e:

agent/shell_hooks.py

@@ -301,9 +301,11 @@ def _spawn(spec: ShellHookSpec, stdin_json: str) -> Dict[str, Any]:
     # Own process group on POSIX so a timed-out hook's descendants are reaped with it (Windows: kill_process_tree
     # / taskkill /T). Hooks that finish in time keep detached helpers alive.
     popen_kwargs: Dict[str, Any] = {"creationflags": windows_hide_flags()} if IS_WINDOWS else {"process_group": 0}
+    from agent.delegation_context import delegated_child_subprocess_env
     try:
         proc = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-                                text=True, encoding='utf-8', errors='replace', shell=False, **popen_kwargs)
+                                text=True, encoding='utf-8', errors='replace', shell=False,
+                                env=delegated_child_subprocess_env(), **popen_kwargs)
     except Exception as exc:
         return failed(next((msg for cls, msg in _POPEN_ERRORS if isinstance(exc, cls)), str(exc)))
     try:

agent/skill_preprocessing.py

@@ -47,6 +47,7 @@ def run_inline_shell(command: str, cwd: Path | None, timeout: int) -> str:
     stdout is empty). Failures return an ``[inline-shell ...]`` marker instead
     of raising, so one bad snippet can't wreck the whole skill message."""
     _popen_kwargs = {"creationflags": windows_hide_flags()} if IS_WINDOWS else {}
+    from agent.delegation_context import delegated_child_subprocess_env
     try:
         completed = subprocess.run(
             ["bash", "-c", command],
@@ -56,6 +57,7 @@ def run_inline_shell(command: str, cwd: Path | None, timeout: int) -> str:
             timeout=max(1, int(timeout)),
             check=False,
             stdin=subprocess.DEVNULL,
+            env=delegated_child_subprocess_env(),
             **_popen_kwargs,
         )
     except subprocess.TimeoutExpired:

agent/transports/codex_app_server.py

@@ -63,9 +63,23 @@ class CodexAppServerClient:
             spawn_env["CODEX_HOME"] = codex_home
 
         cmd = [codex_bin, "app-server", *(extra_args or [])]
+        from agent.delegation_context import (
+            DELEGATED_CHILD_ENV_MARKER, KANBAN_ENV_KEYS,
+            delegated_child_subprocess_env, is_dispatcher_owned_worker_context,
+        )
+        # Native shell children remain unowned. Only Hermes' managed MCP tool
+        # endpoint acts for this worker; grant it scope via its existing per-server
+        # environment, never by granting the whole executor process ownership.
+        owned_task = os.environ.get("HERMES_KANBAN_TASK") and is_dispatcher_owned_worker_context()
+        if owned_task:
+            for key in (*KANBAN_ENV_KEYS, "HERMES_KANBAN_DB", "HERMES_KANBAN_BOARD"):
+                if key in os.environ:
+                    cmd += ["-c", f"mcp_servers.hermes-mcp.env.{key}={json.dumps(os.environ[key])}"]
+            cmd += ["-c", f'mcp_servers.hermes-mcp.env.{DELEGATED_CHILD_ENV_MARKER}=""']
+        spawn_env = delegated_child_subprocess_env(spawn_env)
         # Kanban workers must write handoff/status to the board DB outside the
         # workspace: keep the sandbox on, add the Kanban root as writable.
-        if spawn_env.get("HERMES_KANBAN_TASK"):
+        if owned_task:
             kanban_db = spawn_env.get("HERMES_KANBAN_DB")
             default_root = os.path.join(spawn_env.get("HERMES_HOME", os.path.expanduser("~/.hermes")), "kanban")
             kanban_root = os.path.dirname(kanban_db) if kanban_db else spawn_env.get("HERMES_KANBAN_ROOT", default_root)

cron/scheduler_delivery.py

@@ -665,7 +665,8 @@ def _deliver_to_bot_chat(job: dict, content: str, profile: str) -> Optional[str]
         logger.warning("Job '%s': %s", job_id, msg, **log_kwargs)
         return msg
 
-    env = os.environ.copy()
+    from agent.delegation_context import delegated_child_subprocess_env
+    env = delegated_child_subprocess_env(os.environ)
     if profile:
         argv += ["-p", profile]
         # -p owns profile resolution; this scheduler's HERMES_HOME must not shadow it.

evals/kanban_scope_probe.py

@@ -0,0 +1,60 @@
+"""Credential-free app-server/MCP scope probe; writes only its temporary board."""
+import json
+import os
+from pathlib import Path
+import subprocess
+import shutil
+import sys
+import tempfile
+
+repo = Path(sys.argv[1]).resolve()
+if len(sys.argv) == 2:
+    with tempfile.TemporaryDirectory(prefix="kanban-transport-") as home:
+        env = {"HOME": home, "HERMES_HOME": home + "/hermes", "PATH": os.environ["PATH"], "PYTHONDONTWRITEBYTECODE": "1"}
+        p = subprocess.run([sys.executable, __file__, str(repo), "isolated"], cwd=home, env=env, stdin=subprocess.DEVNULL)
+        sys.exit(p.returncode)
+sys.path.insert(0, str(repo))
+from hermes_cli import kanban_db as kb
+from hermes_cli.kanban_db_connect import connect
+from agent.transports.codex_app_server import CodexAppServerClient
+
+home = Path(os.environ["HOME"])
+hh = Path(os.environ["HERMES_HOME"])
+hh.mkdir(exist_ok=True)
+(hh / "config.yaml").write_text("toolsets: [kanban]\n")
+db = home / "assigned.db"
+conn = connect(db)
+own, foreign = [kb.create_task(conn, title=x) for x in ("owner", "foreign")]
+for t in (own, foreign):
+    kb.claim_task(conn, t)
+task = kb.get_task(conn, own)
+os.environ.update({"HERMES_KANBAN_DB": str(db), "HERMES_KANBAN_BOARD": "default", "HERMES_KANBAN_TASK": own, "HERMES_KANBAN_RUN_ID": str(task.current_run_id), "HERMES_KANBAN_CLAIM_LOCK": task.claim_lock})
+ch = home / "codex"
+ch.mkdir()
+(ch / "config.toml").write_text(
+    'model="fixture"\nmodel_provider="fixture"\n'
+    '[model_providers.fixture]\nname="fixture"\nbase_url="http://127.0.0.1:9/v1"\nwire_api="responses"\n'
+    '[mcp_servers.hermes-mcp]\ncommand=' + json.dumps(sys.executable) + '\nargs=["-m","agent.transports.hermes_tools_mcp_server"]\nstartup_timeout_sec=40\n'
+    '[mcp_servers.hermes-mcp.env]\nPYTHONPATH=' + json.dumps(str(repo)) + '\nHERMES_HOME=' + json.dumps(str(hh)) + '\n'
+)
+report = {}
+with CodexAppServerClient(codex_bin=shutil.which("codex") or "codex", codex_home=str(ch)) as c:
+    try:
+        report["initialize"] = c.initialize(capabilities={"experimentalApi": True})
+    except Exception:
+        print(json.dumps({"stderr": c._stderr_lines, "args": c._proc.args, "rc": c._proc.poll()}), flush=True)
+        raise
+    script = "import os,sys,json;sys.path.insert(0," + repr(str(repo)) + ");from tools import kanban_tools as kt;print(json.dumps({'marker':os.getenv('HERMES_DELEGATED_CHILD_CONTEXT'),'db':os.getenv('HERMES_KANBAN_DB'),'mutation':json.loads(kt._handle_complete({'task_id':" + repr(own) + ",'summary':'must refuse'}))}))"
+    report["native_child"] = c.request("command/exec", {"command": [sys.executable, "-c", script], "cwd": str(home), "sandboxPolicy": {"type": "dangerFullAccess"}}, timeout=40)
+    thread = c.request("thread/start", {"model": "fixture", "modelProvider": "fixture", "cwd": str(home), "approvalPolicy": "never", "sandbox": "workspace-write"}, timeout=50)
+    thread_id = thread["thread"]["id"]
+    status = c.request("mcpServerStatus/list", {"threadId": thread_id}, timeout=50)
+    report["mcp_servers"] = [{"name": x.get("name"), "tools": list(x.get("tools", {}))} for x in status.get("data", [])]
+    for name, tid in (("foreign", foreign), ("own", own)):
+        report[name] = c.request("mcpServer/tool/call", {"threadId": thread_id, "server": "hermes-mcp", "tool": "kanban_complete", "arguments": {"task_id": tid, "summary": "supervised parent handoff"}}, timeout=50)
+    report["stderr"] = c._stderr_lines[-8:]
+report["readback"] = {"own": kb.get_task(conn, own).status, "foreign": kb.get_task(conn, foreign).status, "integrity": conn.execute("PRAGMA integrity_check").fetchone()[0]}
+print(json.dumps(report, indent=2))
+assert report["readback"] == {"own": "done", "foreign": "running", "integrity": "ok"}, report
+native = json.loads(report["native_child"]["stdout"])
+assert native["marker"] == "1" and native["db"] == str(db) and "error" in native["mutation"], report

hermes_cli/kanban.py

@@ -191,7 +191,7 @@ def kanban_command(args: argparse.Namespace) -> int:
             return _err(f"kanban: unknown action {action!r}", 2)
         try:
             return int(handler(args) or 0)
-        except (ValueError, RuntimeError) as exc:
+        except (ValueError, RuntimeError, PermissionError) as exc:
             return _err(f"kanban: {exc}")
 
 
@@ -215,12 +215,13 @@ _DELEGATED_CHILD_DENIED_ACTIONS: frozenset[str] = frozenset({
     "claim", "comment", "attach", "attach-rm", "complete", "edit", "block",
     "schedule", "unblock", "promote", "archive", "dispatch", "daemon", "repair",
     "heartbeat", "notify-subscribe", "notify-unsubscribe", "specify", "decompose",
+    "request-review", "request-changes", "reopen-review",
     "gc",
 })
 
 _DELEGATED_CHILD_DENIED_BOARD_ACTIONS: frozenset[str] = frozenset({
     "create", "new", "rm", "remove", "delete", "switch", "use", "rename",
-    "set-default-workdir",
+    "set-default-workdir", "import",
 })
 
 
@@ -738,6 +739,7 @@ def _cmd_attach(args: argparse.Namespace) -> int:
     """Attach a local file via the shared ``store_attachment_bytes`` path (same 25 MB cap and name
     sanitisation as the dashboard upload and agent tool)."""
     import mimetypes
+    _worker_run_id_for(args.task_id)
 
     src = Path(args.path).expanduser()
     if not src.is_file():
@@ -784,6 +786,9 @@ def _cmd_attach_rm(args: argparse.Namespace) -> int:
 
 
 def _worker_run_id_for(task_id: str) -> Optional[int]:
+    env_tid = os.environ.get("HERMES_KANBAN_TASK")
+    if env_tid and env_tid != task_id:
+        raise ValueError(f"worker is scoped to task {env_tid}; refusing to mutate {task_id}")
     raw = os.environ.get("HERMES_KANBAN_RUN_ID")
     if os.environ.get("HERMES_KANBAN_TASK") != task_id or not raw:
         return None
@@ -929,6 +934,8 @@ def _cmd_schedule(args: argparse.Namespace) -> int:
 
 
 def _cmd_unblock(args: argparse.Namespace) -> int:
+    if os.environ.get("HERMES_KANBAN_TASK"):
+        return _err("kanban unblock is orchestrator-only; workers must hand off their assigned task")
     ids, rc = _require_ids(args)
     if rc:
         return rc

hermes_cli/kanban_db_connect.py

@@ -672,6 +672,16 @@ def connect(db_path: Optional[Path] = None, *, board: Optional[str] = None) -> s
     :func:`kanban_db_path` (``HERMES_KANBAN_DB`` -> ``HERMES_KANBAN_BOARD`` ->
     ``<root>/kanban/current`` -> ``default``)."""
     path = db_path if db_path is not None else _kb.kanban_db_path(board=board)
+    from agent.delegation_context import is_delegated_child_process_context
+    if is_delegated_child_process_context():
+        # Reads must not enter schema/backfill write transactions. Never create a
+        # missing board or migrate on a descendant's behalf; the owner initializes it.
+        conn = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)
+        conn.row_factory = sqlite3.Row
+        if not _schema_is_present(conn):
+            conn.close()
+            raise PermissionError("Kanban descendants require an initialized board; ask its owner to initialize it")
+        return conn
     path.parent.mkdir(parents=True, exist_ok=True)
 
     # Fast path: once THIS process has initialized this path, skip the

hermes_cli/kanban_db_dispatch.py

@@ -2246,6 +2246,9 @@ def _default_spawn(task: Task, workspace: str, *, board: Optional[str] = None) -
     # kanban_comment reads HERMES_PROFILE for its default author; `-p` alone
     # doesn't set the env var.
     env["HERMES_PROFILE"] = profile_arg
+    # This is the grant boundary: the dispatcher assigned this new worker's task.
+    from agent.delegation_context import DELEGATED_CHILD_ENV_MARKER
+    env.pop(DELEGATED_CHILD_ENV_MARKER, None)
     # `--cli` is the highest-precedence TUI override; dropping HERMES_TUI covers
     # older hermes builds on PATH that predate the flag's precedence.
     env.pop("HERMES_TUI", None)

tests/cron/test_cron_kanban_env_isolation.py

@@ -28,7 +28,6 @@ is left completely untouched.
 
 from __future__ import annotations
 
-import ast
 import os
 import threading
 
@@ -380,73 +379,38 @@ class TestRunJobKanbanIsolation:
         assert after == before, "worker identity must survive concurrent cron jobs"
 
 
-# ---------------------------------------------------------------------------
-# Drift guard
-# ---------------------------------------------------------------------------
+@pytest.mark.linux_only
+def test_dispatcher_grants_only_the_assigned_worker_scope(tmp_path, monkeypatch):
+    import json
+    from pathlib import Path
+    import sys
+    from hermes_cli import kanban_db as kb
+    from hermes_cli.kanban_db_connect import connect
+    from hermes_cli.kanban_db_dispatch import _default_spawn
 
-def test_every_dispatcher_kanban_var_is_identity_gated():
-    """Invariant: every HERMES_KANBAN_* var the dispatcher injects is covered by
-    the canonical KANBAN_ENV_KEYS, so the delegate_task subprocess scrubber and
-    any future consumer stay in sync with ``_default_spawn``.
-
-    Fails loudly if a new dispatcher var is added without registering it.
-    """
-    import hermes_cli.kanban_db_dispatch as kanban_db_dispatch
-    from agent.delegation_context import KANBAN_ENV_KEYS
-
-    source = ast.parse(open(kanban_db_dispatch.__file__, encoding="utf-8").read())
-    spawn = next(
-        node for node in ast.walk(source)
-        if isinstance(node, ast.FunctionDef) and node.name == "_default_spawn"
-    )
-
-    injected = set()
-    for node in ast.walk(spawn):
-        # env["HERMES_KANBAN_X"] = ...  and the annotated form
-        if isinstance(node, (ast.Assign, ast.AnnAssign)):
-            targets = node.targets if isinstance(node, ast.Assign) else [node.target]
-            for target in targets:
-                if not isinstance(target, ast.Subscript):
-                    continue
-                if ast.unparse(target.value) != "env":
-                    continue
-                key = ast.unparse(target.slice).strip("\"'")
-                if key.startswith("HERMES_KANBAN_"):
-                    injected.add(key)
-        # env.update({"HERMES_KANBAN_X": ...}) / env.setdefault("HERMES_KANBAN_X", ...)
-        elif isinstance(node, ast.Call):
-            func = ast.unparse(node.func)
-            if func not in ("env.update", "env.setdefault"):
-                continue
-            literals = []
-            for arg in node.args:
-                if isinstance(arg, ast.Dict):
-                    literals.extend(
-                        k for k in arg.keys if isinstance(k, ast.Constant)
-                    )
-                elif isinstance(arg, ast.Constant):
-                    literals.append(arg)
-            for kw in node.keywords:
-                if kw.arg and kw.arg.startswith("HERMES_KANBAN_"):
-                    injected.add(kw.arg)
-            for lit in literals:
-                if isinstance(lit.value, str) and lit.value.startswith(
-                    "HERMES_KANBAN_"
-                ):
-                    injected.add(lit.value)
-
-    assert injected, "failed to parse dispatcher kanban env injection"
-
-    # These are worker-behaviour knobs rather than board/task identity; they are
-    # intentionally not part of KANBAN_ENV_KEYS. Listed explicitly so adding a
-    # new var forces a decision instead of silently passing.
-    behaviour_only = {
-        "HERMES_KANBAN_BRANCH",
-        "HERMES_KANBAN_GOAL_MODE",
-        "HERMES_KANBAN_GOAL_MAX_TURNS",
-    }
-    uncovered = injected - set(KANBAN_ENV_KEYS) - behaviour_only
-    assert not uncovered, (
-        f"dispatcher injects {sorted(uncovered)} which is neither in "
-        "KANBAN_ENV_KEYS nor explicitly classified as behaviour-only"
+    monkeypatch.setenv("HOME", str(tmp_path))
+    db = tmp_path / "board.db"
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(db))
+    conn = connect(db)
+    tid = kb.create_task(conn, title="assigned child", assignee="default")
+    kb.claim_task(conn, tid)
+    task = kb.get_task(conn, tid)
+    output = tmp_path / "worker-result.json"
+    worker = tmp_path / "fixture-worker"
+    root = str(Path(__file__).resolve().parents[2])
+    worker.write_text(
+        f"#!{sys.executable}\nimport sys, os, json;sys.path.insert(0, {root!r})\n"
+        "from tools.kanban_tools import _handle_complete\n"
+        f"result=_handle_complete({{'summary':'assigned worker'}});open({str(output)!r}, 'w').write(result)\n"
     )
+    worker.chmod(0o700)
+    monkeypatch.setenv("HERMES_BIN", str(worker))
+    # Building a new worker under an existing task must replace, not inherit, its scope.
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "prior-task")
+    pid = _default_spawn(task, str(tmp_path), board="default")
+    assert pid is not None
+    os.waitpid(pid, 0)  # windows-footgun: ok — Linux-only real dispatcher spawn
+    assert json.loads(output.read_text())["ok"]
+    assert kb.get_task(conn, tid).status == "done"
+    assert os.environ["HERMES_KANBAN_TASK"] == "prior-task"
+    conn.close()

tests/tools/test_kanban_descendant_scope.py

@@ -0,0 +1,109 @@
+"""Real shell/CLI ingress: inherited context is not a board-write grant."""
+import json
+import os
+from pathlib import Path
+import shlex
+import subprocess
+import sys
+
+from hermes_cli import kanban_db as kb
+from hermes_cli.kanban_db_connect import connect
+from tools import kanban_tools
+from tools.environments.local import LocalEnvironment
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def _worker_board(tmp_path, monkeypatch):
+    db = tmp_path / "assigned.db"
+    conn = connect(db)
+    own, foreign = [kb.create_task(conn, title=title) for title in ("own", "foreign")]
+    for tid in (own, foreign):
+        kb.claim_task(conn, tid)
+    task = kb.get_task(conn, own)
+    for key, value in {
+        "HERMES_KANBAN_DB": str(db), "HERMES_KANBAN_BOARD": "default",
+        "HERMES_KANBAN_TASK": own, "HERMES_KANBAN_RUN_ID": str(task.current_run_id),
+        "HERMES_KANBAN_CLAIM_LOCK": task.claim_lock, "HOME": str(tmp_path),
+    }.items():
+        monkeypatch.setenv(key, value)
+    monkeypatch.delenv("HERMES_DELEGATED_CHILD_CONTEXT", raising=False)
+    return conn, own, foreign
+
+
+def test_terminal_descendants_cannot_mutate_even_after_task_is_removed(tmp_path, monkeypatch):
+    conn, own, foreign = _worker_board(tmp_path, monkeypatch)
+    script = tmp_path / "descendant.py"
+    script.write_text(
+        "import os, sys, json, subprocess\n"
+        f"sys.path.insert(0, {str(ROOT)!r})\n"
+        "from tools import kanban_tools as kt\n"
+        "from agent.delegation_context import is_dispatcher_owned_worker_context\n"
+        f"own, foreign = {own!r}, {foreign!r}\n"
+        "out = {'owner': is_dispatcher_owned_worker_context(), 'default': kt._default_task_id(None),"
+        " 'db': os.getenv('HERMES_KANBAN_DB'), 'board': os.getenv('HERMES_KANBAN_BOARD')}\n"
+        "out['show'] = json.loads(kt._handle_show({'task_id':own}))\n"
+        "out['tools'] = [json.loads(kt._handle_complete({'task_id': t, 'summary':'must refuse'})) for t in (own,foreign)]\n"
+        "os.environ.pop('HERMES_KANBAN_TASK', None)\n"
+        f"p = subprocess.run([sys.executable, '-m', 'hermes_cli.main', 'kanban', 'complete', foreign, '--result', 'must refuse'], cwd={str(ROOT)!r}, capture_output=True, text=True, stdin=subprocess.DEVNULL, timeout=45)\n"
+        "out['later_cli'] = {'rc': p.returncode, 'out':p.stdout, 'err':p.stderr}\n"
+        "print('SCOPE_RESULT=' + json.dumps(out))\n"
+    )
+    terminal = LocalEnvironment(cwd=str(tmp_path))
+    try:
+        result = terminal.execute(f"{shlex.quote(sys.executable)} {shlex.quote(str(script))}")
+    finally:
+        terminal.cleanup()
+    from agent.skill_preprocessing import run_inline_shell
+    from agent.shell_hooks import ShellHookSpec, _spawn
+    from tools.code_execution_env import _build_child_env
+    from tools.mcp_tool_config import _build_safe_env
+
+    command = f"{shlex.quote(sys.executable)} {shlex.quote(str(script))}"
+    outputs = [result.get("output", ""), run_inline_shell(command, tmp_path, 45),
+               _spawn(ShellHookSpec(event="session:start", command=command, timeout=45), "{}")['stdout']]
+    child_envs = [
+        _build_child_env(rpc_endpoint="fixture", rpc_token="fixture", tmpdir=str(tmp_path), child_python=sys.executable),
+        _build_safe_env({"HERMES_HOME": os.environ["HERMES_HOME"]}),
+    ]
+    for env in child_envs:
+        proc = subprocess.run([sys.executable, str(script)], env=env, cwd=tmp_path,
+                              stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=45)
+        assert proc.returncode == 0, proc.stderr
+        outputs.append(proc.stdout)
+    for output in outputs:
+        row = json.loads(next(line.split("SCOPE_RESULT=", 1)[1] for line in output.splitlines() if "SCOPE_RESULT=" in line))
+        assert row["show"]["task"]["id"] == own, row
+        assert not row["owner"] and row["default"] is None, row
+        assert row["db"] == str(tmp_path / "assigned.db") and row["board"] == "default"
+        assert all("error" in value for value in row["tools"]), row
+        assert row["later_cli"]["rc"] != 0, row
+    assert [kb.get_task(conn, t).status for t in (own, foreign)] == ["running", "running"]
+    assert json.loads(kanban_tools._handle_complete({"summary": "parent handoff"}))["ok"]
+    assert kb.get_task(conn, own).status == "done"
+    assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok"
+    conn.close()
+
+
+def test_worker_cli_cannot_use_foreign_task_to_drop_run_scope(tmp_path, monkeypatch):
+    conn, own, foreign = _worker_board(tmp_path, monkeypatch)
+    assert "error" in json.loads(kanban_tools._handle_complete({"task_id": foreign, "summary": "no"}))
+    proc = subprocess.run(
+        [sys.executable, "-m", "hermes_cli.main", "kanban", "complete", foreign, "--result", "no"],
+        cwd=ROOT, env=dict(os.environ), stdin=subprocess.DEVNULL,
+        capture_output=True, text=True, timeout=45,
+    )
+    assert proc.returncode != 0 and "worker is scoped to task" in proc.stderr, (proc.stdout, proc.stderr)
+    assert kb.get_task(conn, foreign).status == "running"
+    attachment = tmp_path / "note.txt"
+    attachment.write_text("fixture")
+    assert kb.block_task(conn, foreign, reason="fixture awaiting orchestrator")
+    for arguments in (["attach", foreign, str(attachment)], ["unblock", foreign]):
+        proc = subprocess.run([sys.executable, "-m", "hermes_cli.main", "kanban", *arguments],
+                              cwd=ROOT, env=dict(os.environ), stdin=subprocess.DEVNULL,
+                              capture_output=True, text=True, timeout=45)
+        assert proc.returncode != 0, (arguments, proc.stdout, proc.stderr)
+    assert not kb.list_attachments(conn, foreign)
+    assert json.loads(kanban_tools._handle_complete({"task_id": own, "summary": "parent"}))["ok"]
+    conn.close()

tools/code_execution_env.py

@@ -96,13 +96,17 @@ def _scrub_child_env(source_env, is_passthrough=None, is_windows=None):
     # delegate_task children are marked by a ContextVar, not os.environ, and the sandbox crosses
     # a process boundary: strip dispatcher-owned Kanban vars AFTER the scrub so an explicit
     # passthrough cannot re-grant a delegated child the parent's board mutation capability.
-    try:
-        from agent.delegation_context import is_delegated_child_process_context, scrub_kanban_env
-        if is_delegated_child_process_context():
-            scrubbed = scrub_kanban_env(scrubbed)
-    except Exception:
-        pass
-    return scrubbed
+    from agent.delegation_context import (
+        DELEGATED_CHILD_ENV_MARKER, delegated_child_subprocess_env,
+    )
+    scoped = delegated_child_subprocess_env(source_env)
+    # Preserve location only when carrying the descendant fence, not for arbitrary
+    # non-allowlisted HERMES_* values in otherwise ordinary execution environments.
+    if scoped.get(DELEGATED_CHILD_ENV_MARKER):
+        for key in (DELEGATED_CHILD_ENV_MARKER, "HERMES_KANBAN_DB", "HERMES_KANBAN_BOARD"):
+            if key in scoped:
+                scrubbed[key] = scoped[key]
+    return delegated_child_subprocess_env(scrubbed)
 
 
 def _build_child_env(*, rpc_endpoint: str, rpc_token: str, tmpdir: str,

tools/environments/local.py

@@ -272,13 +272,8 @@ def _finalize_child_env(env: dict) -> dict:
     _inject_session_context_env(env)
     _strip_hermes_owned_pythonpath_and_runtime_markers(env)
     _apply_windows_msys_bash_env_defaults(env)
-    try:  # strip dispatcher-owned Kanban env from delegate_task child subprocesses
-        from agent.delegation_context import is_delegated_child_process_context, scrub_kanban_env
-        if is_delegated_child_process_context():
-            return scrub_kanban_env(env)
-    except Exception:
-        pass
-    return env
+    from agent.delegation_context import delegated_child_subprocess_env
+    return delegated_child_subprocess_env(env)
 
 
 def _scrubbed_env(parts, plugin_strip: frozenset, fix_path) -> dict:
@@ -338,7 +333,8 @@ def build_subprocess_env(
         _apply_profile_home(env)
     if extra:
         env.update(extra)
-    return env
+    from agent.delegation_context import delegated_child_subprocess_env
+    return delegated_child_subprocess_env(env)
 
 
 # --- Shell discovery ---

tools/kanban_tools.py

@@ -118,7 +118,7 @@ def _kanban_handler(tool_name: str) -> Callable:
 def _reject_delegated_child_mutation(tool_name: str) -> None:
     """A delegate_task child shares the parent's process, so inherited HERMES_KANBAN_*
     env is not proof of ownership: it may report findings but must not mutate."""
-    if _is_delegated_child_context():
+    if _delegation_ctx("is_delegated_child_process_context", False):
         raise _Reject(
             f"{tool_name} refused: delegate_task child agents are not Kanban run owners. "
             "Return findings to the parent agent; the dispatcher worker or an explicitly "

tools/mcp_tool_config.py

@@ -103,9 +103,13 @@ def _build_safe_env(user_env: Optional[dict]) -> dict:
         key: value for key, value in os.environ.items()
         if key in _SAFE_ENV_KEYS or key.upper() in _SAFE_ENV_KEYS_CASE_INSENSITIVE
         or key.startswith("XDG_") or (get_secret_source is not None and get_secret_source(key))}
+    for key in ("HERMES_KANBAN_DB", "HERMES_KANBAN_BOARD"):
+        if key in os.environ:
+            env[key] = os.environ[key]
     if user_env:
         env.update(user_env)
-    return env
+    from agent.delegation_context import delegated_child_subprocess_env
+    return delegated_child_subprocess_env(env)
 
 
 def _which_with_config_pathext(command: str, path_arg, env: dict):

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

@@ -46,6 +46,28 @@ For Hermes profile lanes, the dispatcher's `_default_spawn` runs `hermes -p <ass
 
 For non-Hermes lanes (registered via a plugin), the plugin supplies its own `spawn_fn` callable that gets `task`, `workspace`, and `board` and returns an optional pid for crash detection.
 
+### Descendant process scope
+
+A task assignment belongs to the dispatcher worker, not to every program it starts.
+Hermes subprocess helpers carry a non-owner fence into shells, execution kernels,
+cron deliveries, hooks, language servers, and ordinary stdio MCP servers. Later
+children remain fenced even when a script removes the inherited task ID: CLI and
+tool mutations are rejected, rather than treating that script as an orchestrator.
+Board/database routing and workspace paths are retained. Descendants can read an
+existing board without running schema migrations; its owner must initialize it.
+
+The dispatcher explicitly grants a newly assigned worker its own scope. The managed
+Hermes-tools MCP endpoint can likewise act for its supervising worker, while the
+executor's ordinary shell children remain fenced. Workers may only perform lifecycle
+handoffs and attach files to their assigned task; `unblock` remains orchestrator-only.
+Cross-task comments and follow-up task creation retain their existing behavior.
+
+Integration authors spawning code should use
+`agent.delegation_context.delegated_child_subprocess_env` at the actual spawn, after
+merging environment overrides. It preserves the caller's credential/profile policy.
+This is cooperative runtime scoping, **not OS confinement**: it does not prevent
+arbitrary code from deliberately erasing lineage metadata or opening SQLite directly.
+
 ### 3. A lifecycle terminator
 
 Every claim must end in exactly one of: