fix(kanban): harden delegated-child mutation boundary

agent/delegation_context.py

@@ -17,6 +17,8 @@ _DELEGATED_CHILD_CONTEXT: ContextVar[bool] = ContextVar(
     default=False,
 )
 
+DELEGATED_CHILD_ENV_MARKER = "HERMES_DELEGATED_CHILD_CONTEXT"
+
 KANBAN_ENV_KEYS: tuple[str, ...] = (
     "HERMES_KANBAN_TASK",
     "HERMES_KANBAN_RUN_ID",
@@ -43,9 +45,41 @@ def is_delegated_child_context() -> bool:
     return bool(_DELEGATED_CHILD_CONTEXT.get())
 
 
+def is_delegated_child_process_context() -> bool:
+    """Return True in this process or a subprocess spawned by a child."""
+    import os
+
+    return bool(_DELEGATED_CHILD_CONTEXT.get()) or bool(
+        os.environ.get(DELEGATED_CHILD_ENV_MARKER)
+    )
+
+
 def scrub_kanban_env(env: Mapping[str, str] | MutableMapping[str, str]) -> dict[str, str]:
     """Return *env* with dispatcher-only Kanban variables removed."""
     cleaned = dict(env)
     for key in KANBAN_ENV_KEYS:
         cleaned.pop(key, None)
+    cleaned[DELEGATED_CHILD_ENV_MARKER] = "1"
     return cleaned
+
+
+def delegated_child_subprocess_env(
+    env: Mapping[str, str] | MutableMapping[str, str] | None = None,
+) -> dict[str, str] | None:
+    """Return an env override only when delegated-child lineage must cross fork.
+
+    Most subprocess call sites historically used ``env=None`` to inherit the
+    process environment.  In a ``delegate_task`` child, inheriting as-is leaks
+    parent dispatcher ``HERMES_KANBAN_*`` vars while losing the ContextVar in
+    the new process.  This helper preserves normal ``env=None`` semantics for
+    non-delegated calls, and only materializes a scrubbed env when the lineage
+    marker must be propagated across a child-process boundary.
+    """
+    if not is_delegated_child_process_context():
+        return None if env is None else dict(env)
+
+    if env is None:
+        import os
+
+        env = os.environ
+    return scrub_kanban_env(env)

hermes_cli/kanban.py

@@ -955,6 +955,15 @@ def kanban_command(args: argparse.Namespace) -> int:
             )
         return 0
 
+    # Fast-fail for clearer CLI UX only. The durable trust boundary is lower in
+    # hermes_cli.kanban_db, because children can import DB mutators directly.
+    if _is_delegated_child_cli_mutation(args):
+        print(
+            "kanban: delegate_task child contexts cannot mutate Kanban tasks via the CLI",
+            file=sys.stderr,
+        )
+        return 1
+
     # Board-management commands operate on board metadata and the persisted
     # current-board pointer itself. They must ignore the shared `--board`
     # task-routing override; otherwise `/kanban --board beta boards show`
@@ -1082,6 +1091,66 @@ def _profile_author() -> str:
         return "user"
 
 
+_DELEGATED_CHILD_DENIED_ACTIONS: frozenset[str] = frozenset({
+    "init",
+    "create",
+    "swarm",
+    "assign",
+    "reclaim",
+    "reassign",
+    "link",
+    "unlink",
+    "claim",
+    "comment",
+    "attach",
+    "attach-rm",
+    "complete",
+    "edit",
+    "block",
+    "schedule",
+    "unblock",
+    "promote",
+    "archive",
+    "dispatch",
+    "daemon",
+    "repair",
+    "heartbeat",
+    "notify-subscribe",
+    "notify-unsubscribe",
+    "specify",
+    "decompose",
+    "gc",
+})
+
+_DELEGATED_CHILD_DENIED_BOARD_ACTIONS: frozenset[str] = frozenset({
+    "create",
+    "new",
+    "rm",
+    "remove",
+    "delete",
+    "switch",
+    "use",
+    "rename",
+    "set-default-workdir",
+})
+
+
+def _is_delegated_child_cli_mutation(args: argparse.Namespace) -> bool:
+    action = getattr(args, "kanban_action", None)
+    if action == "boards":
+        boards_action = getattr(args, "boards_action", None) or "list"
+        if boards_action not in _DELEGATED_CHILD_DENIED_BOARD_ACTIONS:
+            return False
+    elif action not in _DELEGATED_CHILD_DENIED_ACTIONS:
+        return False
+    try:
+        from agent.delegation_context import is_delegated_child_process_context
+
+        return is_delegated_child_process_context()
+    except Exception:
+        return bool(os.environ.get("HERMES_DELEGATED_CHILD_CONTEXT"))
+
+
 # ---------------------------------------------------------------------------
 # Boards management (hermes kanban boards …)
 # ---------------------------------------------------------------------------

hermes_cli/kanban_db.py

@@ -138,6 +138,29 @@ _IS_WINDOWS = sys.platform == "win32"
 KANBAN_ATTACHMENT_MAX_BYTES = 25 * 1024 * 1024
 
 
+def _assert_not_delegated_child_mutation() -> None:
+    """Reject Kanban state mutations from ``delegate_task`` child contexts.
+
+    The structured kanban tools and CLI dispatch layer both have fast-fail
+    guards for better UX, but neither is a trust boundary: a delegated child can
+    still shell out to the CLI or import this module directly. The actual
+    invariant belongs at the DB/filesystem mutation layer so every public
+    mutator that uses ``write_txn`` (tasks, runs, comments, attachments,
+    dispatcher claims, repair events, subscriptions, GC, etc.) and every board
+    metadata mutator fails closed before touching durable state.
+    """
+    try:
+        from agent.delegation_context import is_delegated_child_process_context
+
+        delegated = is_delegated_child_process_context()
+    except Exception:
+        delegated = bool(os.environ.get("HERMES_DELEGATED_CHILD_CONTEXT"))
+    if delegated:
+        raise PermissionError(
+            "delegate_task child contexts cannot mutate Kanban tasks or boards"
+        )
+
+
 def _fire_kanban_lifecycle_hook(event: str, task_id: str, **fields: Any) -> None:
     """Fire a kanban lifecycle plugin hook, fully best-effort.
 
@@ -468,6 +491,7 @@ def set_current_board(slug: str) -> Path:
     so that ``hermes kanban boards switch <typo>`` returns an error
     instead of silently pointing at nothing.
     """
+    _assert_not_delegated_child_mutation()
     normed = _normalize_board_slug(slug)
     if not normed:
         raise ValueError("board slug is required")
@@ -479,6 +503,7 @@ def set_current_board(slug: str) -> Path:
 
 def clear_current_board() -> None:
     """Remove ``<root>/kanban/current`` so the active board reverts to ``default``."""
+    _assert_not_delegated_child_mutation()
     try:
         current_board_path().unlink()
     except FileNotFoundError:
@@ -681,6 +706,7 @@ def write_board_metadata(
     Preserves any existing fields not mentioned in the call. Sets
     ``created_at`` on first write. Returns the resulting metadata dict.
     """
+    _assert_not_delegated_child_mutation()
     slug = _normalize_board_slug(board) or DEFAULT_BOARD
     meta = read_board_metadata(slug)
     # Preserve existing DB-derived fields — they get re-computed each
@@ -796,6 +822,7 @@ def remove_board(slug: str, *, archive: bool = True) -> dict:
     Returns a summary dict describing what happened (``{"slug", "action",
     "new_path"}``).
     """
+    _assert_not_delegated_child_mutation()
     normed = _normalize_board_slug(slug)
     if not normed:
         raise ValueError("board slug is required")
@@ -2674,6 +2701,7 @@ def write_txn(conn: sqlite3.Connection):
     a SQLite auto-rollback (which leaves no active transaction) does not
     shadow the original exception with a spurious rollback error.
     """
+    _assert_not_delegated_child_mutation()
     _execute_boundary_with_retry(conn, "BEGIN IMMEDIATE")
     try:
         yield conn

tests/tools/test_delegate_kanban_isolation.py

@@ -2,6 +2,11 @@
 from __future__ import annotations
 
 import json
+import os
+import shlex
+import sys
+
+import pytest
 
 
 def _make_running_kanban_task(monkeypatch, tmp_path):
@@ -131,6 +136,343 @@ def test_delegate_child_terminal_env_scrubs_parent_kanban_keys(monkeypatch):
     assert "HERMES_KANBAN_RUN_ID" not in env
     assert "HERMES_KANBAN_WORKSPACE" not in env
     assert "HERMES_KANBAN_CLAIM_LOCK" not in env
+    assert env["HERMES_DELEGATED_CHILD_CONTEXT"] == "1"
+
+
+def test_delegate_child_foreground_terminal_env_scrubs_parent_kanban_keys(monkeypatch):
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_parent")
+    monkeypatch.setenv("HERMES_KANBAN_RUN_ID", "123")
+    monkeypatch.setenv("HERMES_KANBAN_WORKSPACE", "/tmp/parent-workspace")
+    monkeypatch.setenv("HERMES_KANBAN_CLAIM_LOCK", "lock")
+
+    from agent.delegation_context import delegated_child_context
+    from tools.environments.local import _make_run_env
+
+    with delegated_child_context():
+        env = _make_run_env({"PATH": "/usr/bin"})
+
+    assert "HERMES_KANBAN_TASK" not in env
+    assert "HERMES_KANBAN_RUN_ID" not in env
+    assert "HERMES_KANBAN_WORKSPACE" not in env
+    assert "HERMES_KANBAN_CLAIM_LOCK" not in env
+    assert env["HERMES_DELEGATED_CHILD_CONTEXT"] == "1"
+
+
+def test_delegate_child_process_marker_scrubs_foreground_terminal_kanban_keys(monkeypatch):
+    """A delegated child subprocess has only the env marker, not the ContextVar."""
+    monkeypatch.setenv("HERMES_DELEGATED_CHILD_CONTEXT", "1")
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_parent")
+    monkeypatch.setenv("HERMES_KANBAN_RUN_ID", "123")
+    monkeypatch.setenv("HERMES_KANBAN_DB", "/tmp/parent-kanban.db")
+    monkeypatch.setenv("HERMES_KANBAN_WORKSPACE", "/tmp/parent-workspace")
+    monkeypatch.setenv("HERMES_KANBAN_CLAIM_LOCK", "lock")
+
+    from tools.environments.local import _make_run_env
+
+    env = _make_run_env({"PATH": "/usr/bin"})
+
+    assert "HERMES_KANBAN_TASK" not in env
+    assert "HERMES_KANBAN_RUN_ID" not in env
+    assert "HERMES_KANBAN_DB" not in env
+    assert "HERMES_KANBAN_WORKSPACE" not in env
+    assert "HERMES_KANBAN_CLAIM_LOCK" not in env
+    assert env["HERMES_DELEGATED_CHILD_CONTEXT"] == "1"
+
+
+def test_delegate_child_execute_code_env_preserves_process_marker(monkeypatch, tmp_path):
+    """execute_code has its own env scrubber; it must preserve child lineage."""
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    from tools.code_execution_tool import _scrub_child_env
+
+    env = _scrub_child_env(
+        {
+            "HERMES_HOME": str(home),
+            "HERMES_DELEGATED_CHILD_CONTEXT": "1",
+            "HERMES_KANBAN_TASK": "t_parent",
+            "HERMES_KANBAN_RUN_ID": "123",
+            "HERMES_KANBAN_DB": str(home / "kanban.db"),
+            "HERMES_KANBAN_WORKSPACE": str(tmp_path / "parent-workspace"),
+            "PATH": "/usr/bin",
+        },
+        is_passthrough=lambda _: False,
+        is_windows=False,
+    )
+
+    assert env["HERMES_HOME"] == str(home)
+    assert env["HERMES_DELEGATED_CHILD_CONTEXT"] == "1"
+    assert env["PATH"] == "/usr/bin"
+    assert "HERMES_KANBAN_TASK" not in env
+    assert "HERMES_KANBAN_RUN_ID" not in env
+    assert "HERMES_KANBAN_DB" not in env
+    assert "HERMES_KANBAN_WORKSPACE" not in env
+
+
+def test_delegate_child_execute_code_env_bridges_contextvar_and_scrubs_kanban(
+    monkeypatch,
+    tmp_path,
+):
+    """The real execute_code child-env builder must bridge ContextVar lineage.
+
+    Regression coverage for the vulnerable path: delegate_task marks child
+    execution with a ContextVar, while execute_code used to scrub plain
+    ``os.environ`` and therefore never wrote HERMES_DELEGATED_CHILD_CONTEXT into
+    the sandbox env.
+    """
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_parent")
+    monkeypatch.setenv("HERMES_KANBAN_RUN_ID", "123")
+    monkeypatch.setenv("HERMES_KANBAN_DB", str(home / "kanban.db"))
+    monkeypatch.setenv("HERMES_KANBAN_WORKSPACE", str(tmp_path / "parent-workspace"))
+    monkeypatch.setenv("HERMES_KANBAN_CLAIM_LOCK", "lock")
+    monkeypatch.delenv("HERMES_DELEGATED_CHILD_CONTEXT", raising=False)
+
+    from agent.delegation_context import delegated_child_context
+    from tools.code_execution_tool import _scrub_child_env
+
+    with delegated_child_context():
+        env = _scrub_child_env(
+            dict(os.environ),
+            is_passthrough=lambda k: k.startswith("HERMES_KANBAN_"),
+            is_windows=False,
+        )
+
+    assert os.environ.get("HERMES_DELEGATED_CHILD_CONTEXT") is None
+    assert env["HERMES_HOME"] == str(home)
+    assert env["HERMES_DELEGATED_CHILD_CONTEXT"] == "1"
+    assert "HERMES_KANBAN_TASK" not in env
+    assert "HERMES_KANBAN_RUN_ID" not in env
+    assert "HERMES_KANBAN_DB" not in env
+    assert "HERMES_KANBAN_WORKSPACE" not in env
+    assert "HERMES_KANBAN_CLAIM_LOCK" not in env
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="execute_code UDS sandbox is POSIX-only")
+def test_delegate_child_execute_code_cannot_complete_parent_by_importing_kanban_db(
+    monkeypatch,
+    tmp_path,
+):
+    """E2E: execute_code sandbox inherits child lineage, not parent Kanban env."""
+    kb, tid, _workspace, _attachments_root = _make_running_kanban_task(
+        monkeypatch,
+        tmp_path,
+    )
+
+    from agent.delegation_context import delegated_child_context
+    from tools import code_execution_tool as cet
+
+    code = "\n".join([
+        "import json, os, sqlite3",
+        "from pathlib import Path",
+        "from agent.delegation_context import is_delegated_child_process_context",
+        "from hermes_cli import kanban_db as kb",
+        "observed = {",
+        "    'marker': os.environ.get('HERMES_DELEGATED_CHILD_CONTEXT'),",
+        "    'is_child': is_delegated_child_process_context(),",
+        "    'kanban_keys': sorted(k for k in os.environ if k.startswith('HERMES_KANBAN_')),",
+        "}",
+        "conn = sqlite3.connect(Path(os.environ['HERMES_HOME']) / 'kanban.db')",
+        "conn.row_factory = sqlite3.Row",
+        "try:",
+        f"    kb.complete_task(conn, {tid!r}, summary='child db bypass')",
+        "except PermissionError as exc:",
+        "    observed['permission_error'] = str(exc)",
+        "else:",
+        "    observed['permission_error'] = None",
+        "finally:",
+        "    conn.close()",
+        "print(json.dumps(observed, sort_keys=True))",
+    ])
+
+    monkeypatch.setattr(
+        "tools.approval.check_execute_code_guard",
+        lambda *_args, **_kwargs: {"approved": True},
+    )
+    monkeypatch.setattr(
+        cet,
+        "_load_config",
+        lambda: {"timeout": 15, "max_tool_calls": 50, "mode": "strict"},
+    )
+
+    with delegated_child_context():
+        raw = cet.execute_code(code, task_id="child-execute-code", enabled_tools=[])
+
+    payload = json.loads(raw)
+    assert payload["status"] == "success", payload.get("error", "")
+    observed = json.loads(payload["output"].strip())
+    assert observed["marker"] == "1"
+    assert observed["is_child"] is True
+    assert observed["kanban_keys"] == []
+    assert "delegate_task child contexts cannot mutate Kanban" in observed["permission_error"]
+
+    conn = kb.connect()
+    try:
+        task = kb.get_task(conn, tid)
+        run = kb.latest_run(conn, tid)
+    finally:
+        conn.close()
+
+    assert task.status == "running"
+    assert run.status == "running"
+
+
+def test_delegated_child_subprocess_env_preserves_inherit_semantics_until_needed(monkeypatch):
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_parent")
+    monkeypatch.setenv("HERMES_KANBAN_DB", "/tmp/parent-kanban.db")
+
+    from agent.delegation_context import (
+        delegated_child_context,
+        delegated_child_subprocess_env,
+    )
+
+    assert delegated_child_subprocess_env() is None
+
+    with delegated_child_context():
+        env = delegated_child_subprocess_env()
+
+    assert env is not None
+    assert env["HERMES_DELEGATED_CHILD_CONTEXT"] == "1"
+    assert "HERMES_KANBAN_TASK" not in env
+    assert "HERMES_KANBAN_DB" not in env
+
+
+def test_delegate_child_local_execute_cannot_complete_parent_via_kanban_cli(
+    monkeypatch,
+    tmp_path,
+):
+    kb, tid, _workspace, _attachments_root = _make_running_kanban_task(
+        monkeypatch,
+        tmp_path,
+    )
+
+    from agent.delegation_context import delegated_child_context
+    from tools.environments.local import LocalEnvironment
+
+    code = (
+        "from hermes_cli import kanban; "
+        "import argparse; "
+        "p=argparse.ArgumentParser(); "
+        "sub=p.add_subparsers(dest='cmd'); "
+        "kanban.build_parser(sub); "
+        f"args=p.parse_args(['kanban','complete',{tid!r},'--summary','child cli bypass']); "
+        "raise SystemExit(kanban.kanban_command(args))"
+    )
+    env = LocalEnvironment(cwd=str(tmp_path), timeout=15)
+    try:
+        with delegated_child_context():
+            result = env.execute(
+                f"{shlex.quote(sys.executable)} -c {shlex.quote(code)}",
+                timeout=15,
+            )
+    finally:
+        env.cleanup()
+
+    assert result["returncode"] == 1
+    assert "delegate_task child contexts cannot mutate Kanban tasks" in result["output"]
+
+    conn = kb.connect()
+    try:
+        task = kb.get_task(conn, tid)
+        run = kb.latest_run(conn, tid)
+    finally:
+        conn.close()
+
+    assert task.status == "running"
+    assert run.status == "running"
+
+
+def test_delegate_child_subprocess_cannot_complete_parent_by_importing_kanban_db(
+    monkeypatch,
+    tmp_path,
+):
+    """The DB mutation layer, not only the CLI/tool handlers, is guarded."""
+    kb, tid, _workspace, _attachments_root = _make_running_kanban_task(
+        monkeypatch,
+        tmp_path,
+    )
+
+    from agent.delegation_context import delegated_child_context
+    from tools.environments.local import LocalEnvironment
+
+    code = (
+        "import os, sqlite3; "
+        "from pathlib import Path; "
+        "from hermes_cli import kanban_db as kb; "
+        "conn=sqlite3.connect(Path(os.environ['HERMES_HOME']) / 'kanban.db'); "
+        "conn.row_factory=sqlite3.Row; "
+        "\ntry:\n"
+        f"    kb.complete_task(conn, {tid!r}, summary='child db bypass')\n"
+        "except Exception as exc:\n"
+        "    print(type(exc).__name__ + ': ' + str(exc))\n"
+        "    raise SystemExit(7)\n"
+        "else:\n"
+        "    raise SystemExit(0)\n"
+    )
+    env = LocalEnvironment(cwd=str(tmp_path), timeout=15)
+    try:
+        with delegated_child_context():
+            result = env.execute(
+                f"{shlex.quote(sys.executable)} -c {shlex.quote(code)}",
+                timeout=15,
+            )
+    finally:
+        env.cleanup()
+
+    assert result["returncode"] == 7
+    assert "delegate_task child contexts cannot mutate Kanban tasks or boards" in result["output"]
+
+    conn = kb.connect()
+    try:
+        task = kb.get_task(conn, tid)
+        run = kb.latest_run(conn, tid)
+    finally:
+        conn.close()
+
+    assert task.status == "running"
+    assert run.status == "running"
+
+
+def test_delegate_child_kanban_cli_cannot_delete_parent_board(
+    monkeypatch,
+    tmp_path,
+):
+    kb, _tid, _workspace, _attachments_root = _make_running_kanban_task(
+        monkeypatch,
+        tmp_path,
+    )
+    kb.create_board("victim")
+    assert kb.board_exists("victim")
+
+    from agent.delegation_context import delegated_child_context
+    from tools.environments.local import LocalEnvironment
+
+    code = (
+        "from hermes_cli import kanban; "
+        "import argparse; "
+        "p=argparse.ArgumentParser(); "
+        "sub=p.add_subparsers(dest='cmd'); "
+        "kanban.build_parser(sub); "
+        "args=p.parse_args(['kanban','boards','rm','victim','--delete']); "
+        "raise SystemExit(kanban.kanban_command(args))"
+    )
+    env = LocalEnvironment(cwd=str(tmp_path), timeout=15)
+    try:
+        with delegated_child_context():
+            result = env.execute(
+                f"{shlex.quote(sys.executable)} -c {shlex.quote(code)}",
+                timeout=15,
+            )
+    finally:
+        env.cleanup()
+
+    assert result["returncode"] == 1
+    assert "delegate_task child contexts cannot mutate Kanban tasks" in result["output"]
+    assert kb.board_exists("victim")
+    assert kb.board_dir("victim").is_dir()
 
 
 def test_delegate_child_kanban_mutator_guard_rejects_explicit_task_id(monkeypatch):

tests/tools/test_execute_code_approval_cluster.py

@@ -464,6 +464,7 @@ def test_env_scrub_hermes_allowlist_and_secret_blocks():
         # operational allowlist → kept
         "HERMES_HOME": "/h", "HERMES_PROFILE": "p",
         "HERMES_CONFIG": "/c.yaml", "HERMES_ENV": "/e",
+        "HERMES_DELEGATED_CHILD_CONTEXT": "1",
         # other HERMES_* → dropped (broad prefix removed)
         "HERMES_BASE_URL": "https://x", "HERMES_INTERACTIVE": "1",
         "HERMES_KANBAN_DB": "postgres://u:p@h/db",
@@ -475,7 +476,10 @@ def test_env_scrub_hermes_allowlist_and_secret_blocks():
     }
     out = _scrub_child_env(env, is_passthrough=lambda _: False, is_windows=False)
 
-    for kept in ("HERMES_HOME", "HERMES_PROFILE", "HERMES_CONFIG", "HERMES_ENV", "PATH"):
+    for kept in (
+        "HERMES_HOME", "HERMES_PROFILE", "HERMES_CONFIG", "HERMES_ENV",
+        "HERMES_DELEGATED_CHILD_CONTEXT", "PATH",
+    ):
         assert kept in out, f"{kept} should be kept"
     for dropped in (
         "HERMES_BASE_URL", "HERMES_INTERACTIVE", "HERMES_KANBAN_DB",

tests/tools/test_hermes_subprocess_env.py

@@ -151,6 +151,32 @@ class TestBrowserPassthroughPattern:
         assert "TELEGRAM_BOT_TOKEN" not in env
 
 
+class TestDelegatedChildMarker:
+    def test_delegated_child_context_scrubs_parent_kanban_keys_and_sets_marker(self):
+        from agent.delegation_context import delegated_child_context
+
+        with patch.dict(
+            os.environ,
+            {
+                **_SAFE_SAMPLE,
+                "HERMES_KANBAN_TASK": "t_parent",
+                "HERMES_KANBAN_RUN_ID": "123",
+                "HERMES_KANBAN_DB": "/tmp/parent-kanban.db",
+                "HERMES_KANBAN_WORKSPACE": "/tmp/parent-workspace",
+            },
+            clear=True,
+        ):
+            with delegated_child_context():
+                env = hermes_subprocess_env(inherit_credentials=True)
+
+        assert env["HERMES_DELEGATED_CHILD_CONTEXT"] == "1"
+        assert "HERMES_KANBAN_TASK" not in env
+        assert "HERMES_KANBAN_RUN_ID" not in env
+        assert "HERMES_KANBAN_DB" not in env
+        assert "HERMES_KANBAN_WORKSPACE" not in env
+        assert env["MY_APP_VAR"] == "keep-me"
+
+
 _INTERNAL_DYNAMIC_SAMPLE = {
     "AUXILIARY_VISION_API_KEY": "sk-vision",
     "AUXILIARY_VISION_BASE_URL": "http://internal:1234/v1",

tests/tools/test_voice_mode.py

@@ -134,6 +134,7 @@ class TestDetectAudioEnvironment:
         monkeypatch.delenv("SSH_CLIENT", raising=False)
         monkeypatch.delenv("SSH_TTY", raising=False)
         monkeypatch.delenv("SSH_CONNECTION", raising=False)
+        monkeypatch.setattr("hermes_constants.is_container", lambda: False)
         monkeypatch.setattr("tools.voice_mode._import_audio",
                             lambda: (MagicMock(), MagicMock()))
         monkeypatch.setattr("builtins.open", _non_wsl_proc_version(open))
@@ -424,6 +425,7 @@ class TestDetectAudioEnvironment:
         monkeypatch.delenv("SSH_CLIENT", raising=False)
         monkeypatch.delenv("SSH_TTY", raising=False)
         monkeypatch.delenv("SSH_CONNECTION", raising=False)
+        monkeypatch.setattr("hermes_constants.is_container", lambda: False)
         monkeypatch.setattr("tools.voice_mode.shutil.which", lambda cmd: "/data/data/com.termux/files/usr/bin/termux-microphone-record" if cmd == "termux-microphone-record" else None)
         monkeypatch.setattr("tools.voice_mode._termux_api_app_installed", lambda: True)
         monkeypatch.setattr("tools.voice_mode._import_audio", lambda: (_ for _ in ()).throw(ImportError("no audio libs")))

tools/code_execution_tool.py

@@ -135,7 +135,10 @@ def _truncate_stdout_text(stdout_text: str) -> Tuple[str, Dict[str, Any]]:
 # Environment variable scrubbing rules (shared between the local + remote
 # backends).  Secret-substring block is applied first; anything left must
 # match a safe prefix, the operational HERMES_ allowlist, or (on Windows) an
-# OS-essential name.
+# OS-essential name.  Delegate-task child context is also an exact-name
+# operational marker: without it, a sandbox script that spawns/imports Hermes
+# code can lose the DB-layer Kanban mutation guard while still inheriting
+# HERMES_HOME.
 #
 # NB: the broad "HERMES_" prefix was deliberately removed (#27303) — it leaked
 # HERMES_*-named config that lacks a secret substring (e.g. HERMES_BASE_URL,
@@ -166,6 +169,7 @@ _HERMES_CHILD_ALLOWED = frozenset({
     "HERMES_PROFILE",
     "HERMES_CONFIG",
     "HERMES_ENV",
+    "HERMES_DELEGATED_CHILD_CONTEXT",
 })
 
 # Windows-only: a handful of variables are required by the OS/CRT itself.
@@ -261,6 +265,22 @@ def _scrub_child_env(source_env, is_passthrough=None, is_windows=None):
             len(_dropped_hermes),
             ", ".join(sorted(_dropped_hermes)),
         )
+
+    # delegate_task children are marked with a ContextVar, not os.environ, while
+    # the execute_code sandbox crosses a process boundary. Bridge that context
+    # into the child env and strip dispatcher-owned Kanban variables after the
+    # normal secret/passthrough 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
 
 
@@ -1736,6 +1756,8 @@ def _is_usable_python(python_path: str) -> bool:
     Cached so we don't fork a subprocess on every execute_code call.
     """
     try:
+        from agent.delegation_context import delegated_child_subprocess_env
+
         result = subprocess.run(
             [python_path, "-c",
              "import sys; sys.exit(0 if sys.version_info >= (3, 8) else 1)"],
@@ -1743,6 +1765,7 @@ def _is_usable_python(python_path: str) -> bool:
             capture_output=True,
             creationflags=subprocess.CREATE_NO_WINDOW if _IS_WINDOWS else 0,
             stdin=subprocess.DEVNULL,
+            env=delegated_child_subprocess_env(),
         )
         return result.returncode == 0
     except (OSError, subprocess.TimeoutExpired, subprocess.SubprocessError):

tools/environments/local.py

@@ -490,18 +490,24 @@ def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = Non
 
     _apply_windows_msys_bash_env_defaults(sanitized)
 
+    sanitized = _scrub_delegated_child_kanban_env(sanitized)
+
+    return sanitized
+
+
+def _scrub_delegated_child_kanban_env(env: dict[str, str]) -> dict[str, str]:
+    """Strip dispatcher-owned Kanban env from delegate_task child subprocesses."""
     try:
         from agent.delegation_context import (
-            is_delegated_child_context,
+            is_delegated_child_process_context,
             scrub_kanban_env,
         )
 
-        if is_delegated_child_context():
-            sanitized = scrub_kanban_env(sanitized)
+        if is_delegated_child_process_context():
+            return scrub_kanban_env(env)
     except Exception:
         pass
-
-    return sanitized
+    return env
 
 
 # Tier-1 secrets: stripped from EVERY spawned subprocess unconditionally —
@@ -623,6 +629,12 @@ def hermes_subprocess_env(*, inherit_credentials: bool = False) -> dict[str, str
     # happen; single uniform policy across every spawn surface.
     _inject_session_context_env(env)
 
+    # Non-terminal subprocess helpers (browser, lazy-deps, TUI/ACP hosts, etc.)
+    # also need the delegate_task child lineage marker.  Otherwise a child
+    # context that later imports Kanban DB code in the spawned process would
+    # still see the parent's HERMES_HOME but lose the DB mutation guard.
+    env = _scrub_delegated_child_kanban_env(env)
+
     return env
 
 
@@ -1185,6 +1197,8 @@ def _make_run_env(env: dict) -> dict:
 
     _apply_windows_msys_bash_env_defaults(run_env)
 
+    run_env = _scrub_delegated_child_kanban_env(run_env)
+
     return run_env
 
 

tools/transcription_tools.py

@@ -549,11 +549,14 @@ def _run_command_stt(command: str, timeout: float) -> subprocess.CompletedProces
 
     Mirrors ``tools.tts_tool._run_command_tts``.
     """
+    from agent.delegation_context import delegated_child_subprocess_env
+
     popen_kwargs: Dict[str, Any] = {
         "shell": True,
         "stdout": subprocess.PIPE,
         "stderr": subprocess.PIPE,
         "text": True,
+        "env": delegated_child_subprocess_env(),
     }
     if os.name == "nt":
         popen_kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)

tools/tts_tool.py

@@ -775,11 +775,14 @@ def _terminate_command_tts_process_tree(proc: subprocess.Popen) -> None:
 
 def _run_command_tts(command: str, timeout: float) -> subprocess.CompletedProcess:
     """Run a command-provider shell command with process-tree timeout cleanup."""
+    from agent.delegation_context import delegated_child_subprocess_env
+
     popen_kwargs: Dict[str, Any] = {
         "shell": True,
         "stdout": subprocess.PIPE,
         "stderr": subprocess.PIPE,
         "text": True,
+        "env": delegated_child_subprocess_env(),
     }
     if os.name == "nt":
         popen_kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)