fix(delegation): prevent child HERMES_SESSION_ID leak into parent process env

AIAgent.__init__ calls set_current_session_id(self.session_id), which
mutated both the task-local ContextVar and the process-global os.environ.
Because _build_child_agent wraps construction in delegated_child_context(),
the ContextVar write is harmless (task-local), but the os.environ write
clobbered the parent's HERMES_SESSION_ID for the rest of the process —
leaking the child id into parent tools and subprocesses spawned after
the child was built.

Root cause of HermesPRDelegationSessionContext: parent
20260729_212118_5d797e dispatched child 20260730_160515_736ea1; later
parent terminal inherited HERMES_SESSION_ID=the child.

Fix: set_current_session_id() skips the process-global os.environ write
when called from within a delegated_child_context(). The child's own
tools and subprocesses still resolve their id through the ContextVar
(task-local), while the parent's process-wide env keeps the parent's
session identity. Root agents (CLI, gateway, cron) retain both paths.

Adds 7 regression tests covering single child, concurrent children (8
parallel), parent-tool observation after construction, and root-agent
session rotation backward compatibility. All pass; ruff clean.

agent/agent_init.py

@@ -1465,7 +1465,17 @@ def init_agent(
 
         set_current_session_id(agent.session_id)
     except Exception:
-        os.environ["HERMES_SESSION_ID"] = agent.session_id
+        # Preserve the root-agent legacy fallback, but never let delegated
+        # construction publish a child ID process-wide even if the ContextVar
+        # bridge itself failed to import.
+        try:
+            from agent.delegation_context import is_delegated_child_context
+
+            delegated_child = is_delegated_child_context()
+        except Exception:
+            delegated_child = False
+        if not delegated_child:
+            os.environ["HERMES_SESSION_ID"] = agent.session_id
 
     # Session logs go into ~/.hermes/sessions/ alongside gateway sessions
     hermes_home = get_hermes_home()

agent/delegation_context.py

@@ -31,11 +31,21 @@ KANBAN_ENV_KEYS: tuple[str, ...] = (
 
 
 @contextmanager
-def delegated_child_context() -> Iterator[None]:
-    """Mark the current execution context as a delegate_task child."""
+def delegated_child_context(session_id: str | None = None) -> Iterator[None]:
+    """Mark child execution and isolate its task-local session identity.
+
+    Child construction calls ``set_current_session_id`` internally, so even a
+    context entered without an id must restore the parent's ContextVar.  Child
+    execution passes its explicit id and receives it only for this scope.
+    """
     token = _DELEGATED_CHILD_CONTEXT.set(True)
     try:
-        yield
+        # Import lazily: session_context calls is_delegated_child_context() when
+        # deciding whether the compatibility os.environ mirror is safe.
+        from gateway.session_context import scoped_current_session_id
+
+        with scoped_current_session_id(session_id):
+            yield
     finally:
         _DELEGATED_CHILD_CONTEXT.reset(token)
 

gateway/session_context.py

@@ -36,8 +36,9 @@ needs to replace the import + call site:
     platform = get_session_env("HERMES_SESSION_PLATFORM", "")
 """
 
+from contextlib import contextmanager
 from contextvars import ContextVar
-from typing import Any
+from typing import Any, Iterator
 
 # Sentinel to distinguish "never set in this context" from "explicitly set to empty".
 # When a contextvar holds _UNSET, we fall back to os.environ (CLI/cron compat).
@@ -148,12 +149,52 @@ def set_current_session_id(session_id: str) -> None:
     reconstructing the entire agent. Tools still consult
     ``get_session_env("HERMES_SESSION_ID")`` with an ``os.environ`` fallback,
     so both storage paths must move together when the active session changes.
+
+    Delegated subagent children are the exception: they are constructed inside
+    the parent process within ``delegated_child_context()``, and their
+    ``AIAgent.__init__`` calls this same helper. Writing a child's internal
+    session id to ``os.environ`` (process-global) would clobber the parent's
+    ``HERMES_SESSION_ID`` for the rest of the process — leaking the child id
+    into parent tools and subprocesses spawned after the child was built. The
+    ContextVar write below is task-local and safe for concurrent children; only
+    the process-global ``os.environ`` mirror is suppressed for delegated
+    children. Root agents (CLI, gateway, cron) keep both paths.
     """
     import os
 
-    os.environ["HERMES_SESSION_ID"] = session_id
     _SESSION_ID.set(session_id)
 
+    # Skip the process-global os.environ write for delegated children. The
+    # child's own tools and subprocesses still resolve their id through the
+    # ContextVar (task-local), while the parent's process-wide env keeps the
+    # parent's session identity. See HermesPRDelegationSessionContext task.
+    try:
+        from agent.delegation_context import is_delegated_child_context
+
+        if is_delegated_child_context():
+            return
+    except Exception:
+        pass
+
+    os.environ["HERMES_SESSION_ID"] = session_id
+
+
+@contextmanager
+def scoped_current_session_id(session_id: str | None = None) -> Iterator[None]:
+    """Bind a task-local session id and restore the prior value on exit.
+
+    With ``session_id=None`` this acts as a save/restore boundary around code
+    that may call :func:`set_current_session_id` itself (notably delegated
+    ``AIAgent`` construction).  It intentionally never mutates ``os.environ``.
+    """
+    previous = _SESSION_ID.get()
+    if session_id is not None:
+        _SESSION_ID.set(session_id)
+    try:
+        yield
+    finally:
+        _SESSION_ID.set(previous)
+
 
 def set_session_vars(
     platform: str = "",

tests/gateway/test_delegation_session_id_leak.py

@@ -0,0 +1,143 @@
+"""Delegated children must not replace their parent's session identity."""
+
+from __future__ import annotations
+
+import os
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+from agent.delegation_context import delegated_child_context
+from gateway.session_context import (
+    _SESSION_ID,
+    _UNSET,
+    get_session_env,
+    set_current_session_id,
+)
+from tools.environments.local import build_subprocess_env
+
+
+@pytest.fixture(autouse=True)
+def _isolate_session_context():
+    saved_env = os.environ.get("HERMES_SESSION_ID")
+    saved_ctx = _SESSION_ID.get()
+    _SESSION_ID.set(_UNSET)
+    os.environ.pop("HERMES_SESSION_ID", None)
+    try:
+        yield
+    finally:
+        _SESSION_ID.set(saved_ctx)
+        if saved_env is None:
+            os.environ.pop("HERMES_SESSION_ID", None)
+        else:
+            os.environ["HERMES_SESSION_ID"] = saved_env
+
+
+def _construct_child(session_id: str) -> tuple[object, str | None]:
+    """Model the session mutation performed by AIAgent.__init__."""
+    with delegated_child_context():
+        set_current_session_id(session_id)
+        return _SESSION_ID.get(), os.environ.get("HERMES_SESSION_ID")
+
+
+def test_root_agent_keeps_contextvar_and_environment_in_sync():
+    set_current_session_id("parent-session")
+
+    assert _SESSION_ID.get() == "parent-session"
+    assert os.environ["HERMES_SESSION_ID"] == "parent-session"
+    assert get_session_env("HERMES_SESSION_ID") == "parent-session"
+
+
+def test_child_construction_restores_both_parent_id_paths():
+    set_current_session_id("parent-session")
+
+    inside_context, inside_environment = _construct_child("child-session")
+
+    assert inside_context == "child-session"
+    assert inside_environment == "parent-session"
+    assert _SESSION_ID.get() == "parent-session"
+    assert os.environ["HERMES_SESSION_ID"] == "parent-session"
+    assert get_session_env("HERMES_SESSION_ID") == "parent-session"
+
+
+def test_child_execution_binds_own_id_then_restores_parent():
+    set_current_session_id("parent-session")
+
+    with delegated_child_context("child-session"):
+        assert _SESSION_ID.get() == "child-session"
+        assert get_session_env("HERMES_SESSION_ID") == "child-session"
+        assert os.environ["HERMES_SESSION_ID"] == "parent-session"
+
+    assert _SESSION_ID.get() == "parent-session"
+    assert get_session_env("HERMES_SESSION_ID") == "parent-session"
+    assert os.environ["HERMES_SESSION_ID"] == "parent-session"
+
+
+def test_child_subprocess_environment_receives_child_id():
+    set_current_session_id("parent-session")
+
+    with delegated_child_context("child-session"):
+        child_env = build_subprocess_env(
+            base={"HERMES_SESSION_ID": "foreign-session"},
+        )
+
+    assert child_env["HERMES_SESSION_ID"] == "child-session"
+    assert os.environ["HERMES_SESSION_ID"] == "parent-session"
+    assert _SESSION_ID.get() == "parent-session"
+
+
+def test_parallel_children_keep_parent_environment_and_own_contexts():
+    set_current_session_id("parent-session")
+    child_ids = [f"child-{index}" for index in range(8)]
+
+    def run_child(child_id: str) -> tuple[str, str, str | None, object]:
+        with delegated_child_context(child_id):
+            observed_context = str(_SESSION_ID.get())
+            observed_subprocess = build_subprocess_env(base={}).get(
+                "HERMES_SESSION_ID"
+            )
+            observed_environment = os.environ.get("HERMES_SESSION_ID")
+        return (
+            observed_context,
+            str(observed_subprocess),
+            observed_environment,
+            _SESSION_ID.get(),
+        )
+
+    with ThreadPoolExecutor(max_workers=len(child_ids)) as pool:
+        observations = list(pool.map(run_child, child_ids))
+
+    for child_id, observation in zip(child_ids, observations):
+        context_id, subprocess_id, environment_id, after_context = observation
+        assert context_id == child_id
+        assert subprocess_id == child_id
+        assert environment_id == "parent-session"
+        # ThreadPoolExecutor workers do not inherit the caller's ContextVar;
+        # the scope must restore that worker's original unset value.
+        assert after_context is _UNSET
+
+    assert _SESSION_ID.get() == "parent-session"
+    assert os.environ["HERMES_SESSION_ID"] == "parent-session"
+
+
+def test_nested_child_scope_restores_outer_child_then_parent():
+    set_current_session_id("parent-session")
+
+    with delegated_child_context("child-outer"):
+        assert get_session_env("HERMES_SESSION_ID") == "child-outer"
+        with delegated_child_context("child-inner"):
+            assert get_session_env("HERMES_SESSION_ID") == "child-inner"
+        assert get_session_env("HERMES_SESSION_ID") == "child-outer"
+
+    assert get_session_env("HERMES_SESSION_ID") == "parent-session"
+    assert os.environ["HERMES_SESSION_ID"] == "parent-session"
+
+
+def test_root_rotation_still_updates_both_paths_after_child():
+    set_current_session_id("parent-v1")
+    _construct_child("child-session")
+
+    set_current_session_id("parent-v2")
+
+    assert _SESSION_ID.get() == "parent-v2"
+    assert os.environ["HERMES_SESSION_ID"] == "parent-v2"

tools/delegate_tool.py

@@ -2176,7 +2176,7 @@ def _run_single_child(
             _worker_thread_holder["t"] = threading.current_thread()
             from agent.delegation_context import delegated_child_context
 
-            with delegated_child_context():
+            with delegated_child_context(str(getattr(child, "session_id", "") or "")):
                 return child.run_conversation(
                     user_message=goal,
                     task_id=child_task_id,