fix(gateway): close cross-session HERMES_SESSION_* leak into subprocess env

Session vars (HERMES_SESSION_*) have a process-global os.environ mirror written
last-writer-wins as a CLI/cron fallback and never cleared. Under a concurrent
multi-session host (messaging gateway, ACP adapter, API server, TUI) that global
belongs to whichever turn wrote it last. A subprocess spawned from a task whose
session ContextVar is _UNSET (a sibling task that never bound, or one that
inherited another session's context) inherited the FOREIGN global and acted on
another session's identity.

Add a session_context_engaged() latch (set once any host calls set_session_vars)
and route both terminal spawn paths through a single _inject_session_context_env
chokepoint: once engaged, a bound ContextVar (incl. "") is authoritative and an
_UNSET var is STRIPPED rather than inheriting the possibly-foreign global. Pure
single-process CLI/one-shot (never engaged) keeps the inherited fallback.

Salvaged from #50531 (supersedes #49922). local.py hunk re-applied by intent
onto the current hermes_subprocess_env refactor.

Co-authored-by: PolyphonyRequiem <3107779+PolyphonyRequiem@users.noreply.github.com>

gateway/run.py

@@ -8163,6 +8163,23 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         """
         source = event.source
 
+        # 🔴 Cross-session leak guard. This handler runs inside a per-message
+        # asyncio task created via create_task(), which snapshots the spawning
+        # context with copy_context(). If a *concurrent* message had already
+        # bound its session via set_session_vars() when this task was created,
+        # we inherited ITS HERMES_SESSION_* ContextVars. Until we bind our own
+        # (a few steps down, in _set_session_env), any subprocess spawned here
+        # would read the foreign session's identity via the subprocess-env
+        # bridge — the _UNSET-strip guard there can't help because the vars are
+        # set-to-foreign, not _UNSET. Reset to _UNSET now so that window strips
+        # safe (no session) instead of leaking the sibling's. See
+        # gateway/session_context.reset_session_vars + the inheritance test.
+        try:
+            from gateway.session_context import reset_session_vars
+            reset_session_vars()
+        except Exception:
+            logger.debug("reset_session_vars failed at handler entry", exc_info=True)
+
         if (
             getattr(self, "_startup_restore_in_progress", False)
             and not getattr(event, "internal", False)

gateway/session_context.py

@@ -44,6 +44,28 @@ from typing import Any
 # When it holds "" (after clear_session_vars resets it), we return "" — no fallback.
 _UNSET: Any = object()
 
+# Process-level flag: has any code in this process bound a session via
+# set_session_vars()? Concurrent multi-session hosts (the messaging gateway, the
+# ACP adapter, the API server, the TUI, cron) all do; a pure single-process
+# CLI/one-shot that never engages the session-context system does not.
+#
+# The subprocess-env bridge (tools/environments/local.py) reads this to choose
+# its leak policy: when engaged, the ContextVars are authoritative and an _UNSET
+# var means "no session bound in THIS task" — so a process-global os.environ
+# mirror (written last-writer-wins by whatever concurrent session ran most
+# recently) must NOT be inherited into a child process. When never engaged, the
+# os.environ fallback is preserved (no concurrency to leak across). Monotonic
+# latch — once any host binds a session, the process stays engaged for life.
+_session_context_engaged: bool = False
+
+
+def session_context_engaged() -> bool:
+    """True if any session has been bound via set_session_vars in this process.
+
+    See the ``_session_context_engaged`` comment for the leak-policy rationale.
+    """
+    return _session_context_engaged
+
 # ---------------------------------------------------------------------------
 # Per-task session variables
 # ---------------------------------------------------------------------------
@@ -150,6 +172,11 @@ def set_session_vars(
     ``_SESSION_ASYNC_DELIVERY`` / ``async_delivery_supported``). Stateless
     request/response adapters (the API server) pass ``False``.
     """
+    # Mark the session-context machinery engaged for this process. The
+    # subprocess-env bridge uses this to switch from "os.environ fallback" to
+    # "ContextVar-authoritative, strip on _UNSET" — see session_context_engaged.
+    global _session_context_engaged
+    _session_context_engaged = True
     tokens = [
         _SESSION_PLATFORM.set(platform),
         _SESSION_SOURCE.set(source),
@@ -209,6 +236,54 @@ def clear_session_vars(tokens: list) -> None:
         pass
 
 
+def reset_session_vars() -> None:
+    """Reset every session context variable to ``_UNSET`` for THIS context.
+
+    Distinct from :func:`clear_session_vars`, which sets the vars to ``""``
+    ("explicitly cleared" — suppresses the os.environ fallback and is used when
+    a handler *finishes*).  This helper restores the ``_UNSET`` sentinel
+    ("never bound in this context"), which is what a freshly-spawned task should
+    look like *before* it binds its own session.
+
+    🔴 Why this exists — the cross-session ContextVar inheritance leak.
+    Each gateway message is processed in its own ``asyncio`` task, created via
+    ``create_task`` (which snapshots the *current* context with
+    ``copy_context``).  When message B's task is spawned from a context where a
+    concurrent message A had already called :func:`set_session_vars`, B inherits
+    A's **set** ContextVars.  Until B calls its own ``set_session_vars`` there is
+    a window where any subprocess B spawns (e.g. a tool shelling out) reads
+    *A's* ``HERMES_SESSION_*`` identity via the subprocess-env bridge.  The
+    bridge's ``_UNSET``-strip guard cannot help: the vars are not ``_UNSET``,
+    they are set-to-A.  Calling ``reset_session_vars`` at the top of the
+    per-message handler drops the inherited identity so the window strips safe
+    (no session) instead of leaking the foreign one; the handler then binds its
+    own via ``set_session_vars`` a few steps later.  See
+    tests/tools/test_local_env_session_leak.py and
+    tests/gateway/test_session_context_inheritance.py.
+
+    Note ``_SESSION_ASYNC_DELIVERY`` lives outside ``_VAR_MAP`` (it is a bool
+    capability flag read via :func:`async_delivery_supported`, not a string
+    ``HERMES_SESSION_*`` env var read via :func:`get_session_env`), so it is
+    reset explicitly below. Without it, a task spawned from a context where a
+    sibling adapter bound ``async_delivery=False`` (the stateless API server)
+    inherits that ``False`` through the pre-bind window, and
+    ``async_delivery_supported`` wrongly reports the new turn's channel as
+    unable to route a background completion until ``set_session_vars`` runs.
+    """
+    for var in _VAR_MAP.values():
+        var.set(_UNSET)
+    # Reset the async-delivery capability to "never bound here" (_UNSET) for the
+    # same inheritance-leak reason as the mapped vars above — see clear_session_vars,
+    # which resets this var on the handler-exit path for the symmetric concern.
+    _SESSION_ASYNC_DELIVERY.set(_UNSET)
+    try:
+        from agent.runtime_cwd import clear_session_cwd
+
+        clear_session_cwd()
+    except Exception:
+        pass
+
+
 def get_session_env(name: str, default: str = "") -> str:
     """Read a session context variable by its legacy ``HERMES_SESSION_*`` name.
 

tests/gateway/test_session_context_inheritance.py

@@ -0,0 +1,262 @@
+"""Cross-session ContextVar *inheritance* leak guard.
+
+Companion to ``tests/tools/test_local_env_session_leak.py``. That file covers
+the ``os.environ``-mirror leak (a subprocess inheriting a foreign *global* when
+this task's ContextVar is ``_UNSET``). THIS file covers a distinct, subtler
+variant that the ``_UNSET``-strip guard does NOT catch:
+
+    Each gateway message is processed in its own asyncio task, created via
+    ``create_task`` — which snapshots the spawning context with
+    ``copy_context()``. If message B's task is created from a context where a
+    *concurrent* message A had ALREADY called ``set_session_vars``, B inherits
+    A's **set** ContextVars. Between B's task start and B's own
+    ``set_session_vars`` call, any subprocess B spawns reads A's
+    ``HERMES_SESSION_*`` identity through the subprocess-env bridge. The bridge's
+    strip-on-``_UNSET`` rule is no help: the inherited vars are set-to-A, not
+    ``_UNSET``.
+
+Verified in production 2026-06-21: a ``/bug`` turn ran ``bug_thread.py whoami``
+and read a concurrent session's ticket (``cursor-captive-modals``) instead of
+its own, because its task inherited that session's bound ContextVars.
+
+The fix: ``gateway.session_context.reset_session_vars`` resets every session var
+to ``_UNSET`` at the top of the per-message handler (``GatewayRunner._handle_message``),
+*before* any work, so an inherited identity is dropped and the pre-bind window
+strips safe instead of leaking the sibling's. The handler then binds its own
+session a few steps later.
+"""
+import asyncio
+from contextvars import copy_context
+
+import pytest
+
+import gateway.session_context as sc
+from gateway.session_context import (
+    _SESSION_ASYNC_DELIVERY,
+    _UNSET,
+    _VAR_MAP,
+    async_delivery_supported,
+    reset_session_vars,
+    set_session_vars,
+)
+from tools.environments.local import _make_run_env
+
+SESSION_VARS = list(_VAR_MAP.keys())
+
+MINE = dict(
+    session_key="agent:main:discord:thread:MINE:MINE",
+    platform="discord",
+    chat_id="MINE_CHAT",
+    thread_id="MINE_THREAD",
+    user_id="MINE_USER",
+    chat_name="mine",
+    message_id="MINE_MSG",
+)
+FOREIGN = dict(
+    session_key="agent:main:discord:thread:FOREIGN:FOREIGN",
+    platform="discord",
+    chat_id="FOREIGN_CHAT",
+    thread_id="FOREIGN_THREAD",
+    user_id="FOREIGN_USER",
+    chat_name="foreign",
+    message_id="FOREIGN_MSG",
+)
+
+
+@pytest.fixture(autouse=True)
+def _isolate_session_context():
+    """Clean ContextVar + engaged-latch slate per test, restored afterwards."""
+    import os
+
+    saved_env = {k: os.environ.get(k) for k in SESSION_VARS}
+    saved_ctx = {name: var.get() for name, var in _VAR_MAP.items()}
+    saved_async = _SESSION_ASYNC_DELIVERY.get()
+    saved_engaged = sc._session_context_engaged
+    for var in _VAR_MAP.values():
+        var.set(_UNSET)
+    _SESSION_ASYNC_DELIVERY.set(_UNSET)
+    sc._session_context_engaged = True  # a concurrent multi-session host is engaged
+    try:
+        yield
+    finally:
+        for var, val in zip(_VAR_MAP.values(), saved_ctx.values()):
+            var.set(val)
+        _SESSION_ASYNC_DELIVERY.set(saved_async)
+        sc._session_context_engaged = saved_engaged
+        for k, v in saved_env.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+
+
+def _spawn_view():
+    """What a subprocess spawned right now would see for the session vars."""
+    env = _make_run_env({})
+    return {
+        "HERMES_SESSION_CHAT_ID": env.get("HERMES_SESSION_CHAT_ID"),
+        "HERMES_SESSION_THREAD_ID": env.get("HERMES_SESSION_THREAD_ID"),
+        "HERMES_SESSION_KEY": env.get("HERMES_SESSION_KEY"),
+    }
+
+
+async def _child_turn(reset_first: bool):
+    """Simulate message B's processing task: created (copy_context) from a
+    parent context where message A already bound its session.
+
+    Returns the subprocess view from the *pre-bind window* — before B calls its
+    own set_session_vars. With ``reset_first`` (the fix), B resets at entry.
+    """
+    captured = {}
+
+    def _b_body():
+        if reset_first:
+            reset_session_vars()  # THE FIX: handler-entry reset
+        captured["window"] = _spawn_view()  # pre-bind window
+        set_session_vars(**FOREIGN)  # B binds its own session
+        captured["bound"] = _spawn_view()
+
+    # create_task snapshots the CURRENT (A-bound) context, exactly like the
+    # gateway's per-message dispatch.
+    await asyncio.create_task(_async_noop(_b_body))
+    return captured
+
+
+async def _async_noop(fn):
+    fn()
+
+
+def test_child_task_inherits_foreign_session_without_reset():
+    """REPRODUCER: without the entry reset, B's pre-bind window leaks A's id.
+
+    This is the production hijack. Asserting the leak EXISTS documents the bug
+    the fix closes; the next test proves the fix.
+    """
+    set_session_vars(**MINE)  # parent A binds in the current context
+
+    captured = asyncio.run(_child_turn(reset_first=False))
+
+    # The pre-bind window inherited A's (MINE) identity — the leak.
+    assert captured["window"]["HERMES_SESSION_CHAT_ID"] == "MINE_CHAT", (
+        "Expected to reproduce the inheritance leak (window sees parent's "
+        f"MINE_CHAT); got {captured['window']!r}"
+    )
+
+
+def test_reset_session_vars_closes_inheritance_leak():
+    """THE FIX: resetting at handler entry strips the inherited identity.
+
+    After reset_session_vars(), the pre-bind window must see NO session vars
+    (stripped, because they are _UNSET in this context and the process is
+    engaged) — NOT the parent's MINE_*. B's own bind then takes effect normally.
+    """
+    set_session_vars(**MINE)  # parent A binds in the current context
+
+    captured = asyncio.run(_child_turn(reset_first=True))
+
+    window = captured["window"]
+    for var in ("HERMES_SESSION_CHAT_ID", "HERMES_SESSION_THREAD_ID", "HERMES_SESSION_KEY"):
+        assert window[var] is None, (
+            f"{var} leaked the parent session after reset: {window[var]!r}"
+        )
+
+    # B's own session still binds correctly after the reset window.
+    assert captured["bound"]["HERMES_SESSION_CHAT_ID"] == "FOREIGN_CHAT"
+    assert captured["bound"]["HERMES_SESSION_KEY"] == FOREIGN["session_key"]
+
+
+def test_reset_session_vars_restores_unset_not_empty():
+    """reset_session_vars sets _UNSET (not "" like clear_session_vars).
+
+    The distinction matters: "" is 'explicitly cleared' (suppresses os.environ
+    fallback, used when a handler finishes); _UNSET is 'never bound here' (lets
+    the bridge strip and a CLI fallback resolve). Entry-reset must use _UNSET.
+    """
+    set_session_vars(**MINE)
+    reset_session_vars()
+    for name, var in _VAR_MAP.items():
+        assert var.get() is _UNSET, f"{name} is {var.get()!r}, expected _UNSET"
+
+
+# ---------------------------------------------------------------------------
+# Async-delivery capability inheritance (the sibling var outside _VAR_MAP)
+# ---------------------------------------------------------------------------
+#
+# ``_SESSION_ASYNC_DELIVERY`` is NOT in ``_VAR_MAP`` — it is a bool capability
+# flag read via ``async_delivery_supported()``, not a string ``HERMES_SESSION_*``
+# var read via ``get_session_env``. So the ``for var in _VAR_MAP.values()`` loop
+# in ``reset_session_vars`` does not touch it; it must be reset explicitly.
+#
+# Without that explicit reset, a task created (copy_context) from a context where
+# a *concurrent* sibling A had bound ``async_delivery=False`` (the stateless API
+# server) inherits A's ``False``. In B's pre-bind window
+# ``async_delivery_supported()`` then wrongly reports B's channel as unable to
+# route a background completion — even though B is e.g. a real gateway turn that
+# CAN. Tools (terminal notify_on_complete / watch_patterns, delegate_task
+# background=True) would refuse a promise the channel could actually keep.
+
+
+async def _child_async_delivery(reset_first: bool):
+    """Simulate message B's task created from a parent context where a stateless
+    sibling A bound ``async_delivery=False``.
+
+    Returns ``async_delivery_supported()`` as seen in B's pre-bind window.
+    """
+    captured = {}
+
+    def _b_body():
+        if reset_first:
+            reset_session_vars()  # THE FIX: handler-entry reset
+        captured["window"] = async_delivery_supported()  # pre-bind window
+
+    await asyncio.create_task(_async_noop(_b_body))
+    return captured
+
+
+def test_child_task_inherits_foreign_async_delivery_without_reset():
+    """REPRODUCER: without the entry reset, B inherits A's async_delivery=False.
+
+    A stateless adapter (API server) opts out with async_delivery=False. A task
+    spawned from that context sees the inherited False in its pre-bind window —
+    the leak the explicit reset closes.
+    """
+    set_session_vars(**FOREIGN, async_delivery=False)  # stateless sibling A
+
+    captured = asyncio.run(_child_async_delivery(reset_first=False))
+
+    assert captured["window"] is False, (
+        "Expected to reproduce the async-delivery inheritance leak (window "
+        f"inherits A's async_delivery=False); got {captured['window']!r}"
+    )
+
+
+def test_reset_session_vars_closes_async_delivery_leak():
+    """THE FIX: resetting at handler entry drops the inherited async_delivery.
+
+    After reset_session_vars(), the pre-bind window must fall back to the
+    default-supported behavior (True) — NOT the stateless sibling's False — so a
+    real gateway turn isn't wrongly told its channel can't route async delivery.
+    """
+    set_session_vars(**FOREIGN, async_delivery=False)  # stateless sibling A
+
+    captured = asyncio.run(_child_async_delivery(reset_first=True))
+
+    assert captured["window"] is True, (
+        "After reset, async delivery must default to supported; "
+        f"got {captured['window']!r}"
+    )
+
+
+def test_reset_session_vars_restores_async_delivery_unset():
+    """reset_session_vars restores _SESSION_ASYNC_DELIVERY to the _UNSET sentinel.
+
+    The capability flag must read 'never bound here' (_UNSET), not a falsy value,
+    so async_delivery_supported() resolves to the default-supported path rather
+    than being mistaken for an opted-out stateless adapter.
+    """
+    set_session_vars(**FOREIGN, async_delivery=False)
+    reset_session_vars()
+    assert _SESSION_ASYNC_DELIVERY.get() is _UNSET, (
+        f"_SESSION_ASYNC_DELIVERY is {_SESSION_ASYNC_DELIVERY.get()!r}, expected _UNSET"
+    )
+    assert async_delivery_supported() is True

tests/tools/test_local_env_session_leak.py

@@ -0,0 +1,254 @@
+"""Cross-session HERMES_SESSION_* leak guard for the local terminal backend.
+
+Regression coverage for the bug where a terminal subprocess could observe a
+*different concurrent session's* ``HERMES_SESSION_KEY`` (and the other
+``HERMES_SESSION_*`` vars).
+
+Root cause: the session vars have a process-global ``os.environ`` mirror (written
+last-writer-wins as a CLI/cron fallback, never cleared), while the
+concurrency-safe source of truth is a task-local ``ContextVar``. The subprocess
+env was built from ``os.environ`` and only *overrode* the session vars when the
+ContextVar was set+truthy. When the subprocess was spawned from a thread/context
+that never inherited the agent's copied context (ContextVar ``_UNSET``), the
+override no-op'd and the stale, foreign ``os.environ`` value leaked into the
+child — so e.g. ``bug_thread.py whoami`` read another session's thread id.
+
+The fix: once the session-context machinery is engaged in this process (any
+concurrent host — gateway, ACP, API server, TUI, cron — has called
+``set_session_vars``), the session vars are ContextVar-authoritative. The
+subprocess-env bridge resolves each ``HERMES_SESSION_*`` from the ContextVar and,
+when it is ``_UNSET``, STRIPS the var from the child env rather than inheriting
+the process-global value that may belong to another session. A pure
+single-process CLI/one-shot that never engaged the session-context system keeps
+the ``os.environ`` fallback.
+"""
+
+import os
+
+import pytest
+
+import gateway.session_context as sc
+from gateway.session_context import _VAR_MAP, clear_session_vars, set_session_vars
+from tools.environments.local import _make_run_env, _sanitize_subprocess_env
+
+# The full set of session vars the bridge owns.
+SESSION_VARS = list(_VAR_MAP.keys())
+
+
+@pytest.fixture(autouse=True)
+def _isolate_session_context():
+    """Clean ContextVar + os.environ + engaged-latch slate per test, restored."""
+    saved_env = {k: os.environ.get(k) for k in SESSION_VARS}
+    saved_ctx = {name: var.get() for name, var in _VAR_MAP.items()}
+    saved_engaged = sc._session_context_engaged
+    for var in _VAR_MAP.values():
+        var.set(sc._UNSET)
+    sc._session_context_engaged = False
+    try:
+        yield
+    finally:
+        for var, val in zip(_VAR_MAP.values(), saved_ctx.values()):
+            var.set(val)
+        sc._session_context_engaged = saved_engaged
+        for k, v in saved_env.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+
+
+def _engage():
+    """Mark the session-context machinery engaged, like a concurrent host would."""
+    sc._session_context_engaged = True
+
+
+# --------------------------------------------------------------------------- #
+# Foreground path (_make_run_env)
+# --------------------------------------------------------------------------- #
+
+def test_engaged_unset_contextvar_strips_foreign_session_key(monkeypatch):
+    """Engaged host + UNSET ContextVar must NOT inherit a foreign global.
+
+    This is the production hijack: a concurrent session wrote
+    os.environ["HERMES_SESSION_KEY"], this task's ContextVar is unset, and the
+    subprocess must see NO key rather than the foreign one.
+    """
+    _engage()
+    monkeypatch.setenv(
+        "HERMES_SESSION_KEY",
+        "agent:main:discord:thread:FOREIGN_CONCURRENT:FOREIGN_CONCURRENT",
+    )
+
+    env = _make_run_env({})
+
+    assert "HERMES_SESSION_KEY" not in env, (
+        "Foreign concurrent session key leaked into subprocess env: "
+        f"{env.get('HERMES_SESSION_KEY')!r}"
+    )
+
+
+def test_set_session_vars_engages_and_overrides_foreign_global(monkeypatch):
+    """set_session_vars itself engages the latch and the bound value wins.
+
+    Mirrors a real host: calling set_session_vars both marks the process engaged
+    and binds the ContextVar, so the bound value overrides the foreign global.
+    """
+    monkeypatch.setenv(
+        "HERMES_SESSION_KEY",
+        "agent:main:discord:thread:FOREIGN:FOREIGN",
+    )
+
+    tokens = set_session_vars(
+        session_key="agent:main:discord:group:MY_BUGS_ROOT:111",
+        platform="discord",
+        chat_id="MY_BUGS_ROOT",
+    )
+    try:
+        assert sc.session_context_engaged() is True
+        env = _make_run_env({})
+    finally:
+        clear_session_vars(tokens)
+
+    assert env.get("HERMES_SESSION_KEY") == "agent:main:discord:group:MY_BUGS_ROOT:111"
+
+
+def test_engaged_strips_all_session_vars_when_unset(monkeypatch):
+    """The strip covers every HERMES_SESSION_* mirror, not just the key."""
+    _engage()
+    monkeypatch.setenv("HERMES_SESSION_KEY", "foreign-key")
+    monkeypatch.setenv("HERMES_SESSION_THREAD_ID", "foreign-thread")
+    monkeypatch.setenv("HERMES_SESSION_CHAT_ID", "foreign-chat")
+    monkeypatch.setenv("HERMES_SESSION_USER_ID", "foreign-user")
+
+    env = _make_run_env({})
+
+    for var in (
+        "HERMES_SESSION_KEY",
+        "HERMES_SESSION_THREAD_ID",
+        "HERMES_SESSION_CHAT_ID",
+        "HERMES_SESSION_USER_ID",
+    ):
+        assert var not in env, f"{var} leaked from a foreign global: {env.get(var)!r}"
+
+
+def test_unengaged_process_preserves_os_environ_fallback(monkeypatch):
+    """A process that never engaged the session-context system keeps the fallback.
+
+    Pure single-process CLI/one-shot sets HERMES_SESSION_* directly in os.environ
+    and relies on the subprocess inheriting them; there is no concurrency to leak
+    across, so the strip must NOT apply.
+    """
+    # _isolate_session_context already forced engaged=False.
+    monkeypatch.setenv("HERMES_SESSION_KEY", "cli-session-key")
+    monkeypatch.setenv("HERMES_SESSION_ID", "cli-session-id")
+
+    env = _make_run_env({})
+
+    assert env.get("HERMES_SESSION_KEY") == "cli-session-key"
+    assert env.get("HERMES_SESSION_ID") == "cli-session-id"
+
+
+def test_engaged_explicit_empty_contextvar_clears(monkeypatch):
+    """An explicitly-cleared ContextVar ("" via clear_session_vars) clears the var.
+
+    After a handler finishes it calls clear_session_vars which sets each var to
+    "" (distinct from _UNSET). A subprocess spawned in that window must see the
+    empty value (which overrides the foreign global), NOT the foreign global —
+    an empty key is safe (whoami reads "" → no thread).
+    """
+    monkeypatch.setenv("HERMES_SESSION_KEY", "foreign-after-clear")
+
+    tokens = set_session_vars(session_key="real-key", platform="discord", chat_id="c")
+    clear_session_vars(tokens)  # sets vars to "" (explicitly cleared); stays engaged
+
+    env = _make_run_env({})
+
+    # Explicit-empty wins over the foreign global: either stripped or "" — never
+    # the foreign value. Both outcomes are safe for the consumer.
+    assert env.get("HERMES_SESSION_KEY", "") == "", (
+        f"Foreign key survived an explicit clear: {env.get('HERMES_SESSION_KEY')!r}"
+    )
+
+
+def test_explicit_empty_thread_id_overrides_stale_value(monkeypatch):
+    """A bound-but-empty thread id must override a stale inherited value.
+
+    This is the complementary case (the #38507 scenario): a top-level post with
+    no thread id binds HERMES_SESSION_THREAD_ID="" and that empty value must win
+    over an older non-empty value left in os.environ.
+    """
+    monkeypatch.setenv("HERMES_SESSION_THREAD_ID", "stale-thread-from-prior-turn")
+
+    tokens = set_session_vars(
+        session_key="mm:chan",
+        platform="mattermost",
+        chat_id="chan",
+        thread_id="",  # explicitly no thread
+    )
+    try:
+        env = _make_run_env({})
+    finally:
+        clear_session_vars(tokens)
+
+    assert env.get("HERMES_SESSION_THREAD_ID") == "", (
+        "Bound-empty thread id did not override the stale value: "
+        f"{env.get('HERMES_SESSION_THREAD_ID')!r}"
+    )
+    assert env.get("HERMES_SESSION_KEY") == "mm:chan"
+
+
+# --------------------------------------------------------------------------- #
+# Background / PTY path (_sanitize_subprocess_env via process_registry)
+# --------------------------------------------------------------------------- #
+
+def test_sanitize_subprocess_env_strips_foreign_session_key_when_engaged(monkeypatch):
+    """The background/PTY spawn path gets the same cross-session strip.
+
+    process_registry.spawn_local() builds its env via _sanitize_subprocess_env(
+    os.environ, env_vars). A background subprocess spawned with an UNSET
+    ContextVar in an engaged process must not inherit a foreign session key.
+    """
+    _engage()
+    stale_base = {
+        "PATH": "/usr/bin:/bin",
+        "HERMES_SESSION_KEY": "agent:main:discord:thread:FOREIGN_BG:FOREIGN_BG",
+        "HERMES_SESSION_THREAD_ID": "FOREIGN_BG",
+    }
+
+    sanitized = _sanitize_subprocess_env(stale_base)
+
+    assert "HERMES_SESSION_KEY" not in sanitized, (
+        f"Background subprocess inherited foreign key: {sanitized.get('HERMES_SESSION_KEY')!r}"
+    )
+    assert "HERMES_SESSION_THREAD_ID" not in sanitized
+
+
+def test_sanitize_subprocess_env_set_contextvar_wins_when_engaged():
+    """Background path: a SET ContextVar overrides the foreign global base."""
+    stale_base = {
+        "PATH": "/usr/bin:/bin",
+        "HERMES_SESSION_KEY": "agent:main:discord:thread:FOREIGN_BG:FOREIGN_BG",
+    }
+    tokens = set_session_vars(
+        session_key="agent:main:discord:group:REAL_BG:222",
+        platform="discord",
+        chat_id="REAL_BG",
+    )
+    try:
+        sanitized = _sanitize_subprocess_env(stale_base)
+    finally:
+        clear_session_vars(tokens)
+
+    assert sanitized.get("HERMES_SESSION_KEY") == "agent:main:discord:group:REAL_BG:222"
+
+
+def test_sanitize_subprocess_env_unengaged_preserves_fallback(monkeypatch):
+    """Background path in an unengaged process keeps the inherited value."""
+    stale_base = {
+        "PATH": "/usr/bin:/bin",
+        "HERMES_SESSION_KEY": "cli-bg-key",
+    }
+
+    sanitized = _sanitize_subprocess_env(stale_base)
+
+    assert sanitized.get("HERMES_SESSION_KEY") == "cli-bg-key"

tools/environments/local.py

@@ -265,6 +265,53 @@ def _inject_context_hermes_home(env: dict) -> None:
         pass
 
 
+def _inject_session_context_env(env: dict) -> None:
+    """Bridge gateway session ContextVars into a subprocess environment dict.
+
+    ContextVars don't propagate to child processes, so the live session vars
+    (HERMES_SESSION_*) are bridged onto the child env here.
+
+    🔴 Cross-session leak guard. The session vars also have a process-global
+    os.environ mirror (written last-writer-wins as a CLI/cron fallback, never
+    cleared). Under a concurrent multi-session host (the messaging gateway, ACP
+    adapter, API server, TUI) that global belongs to *whichever turn wrote it
+    last* — NOT necessarily this task. A subprocess spawned from a task whose
+    ContextVar is _UNSET (e.g. a sibling message task that never bound, or one
+    that inherited another session's context) would otherwise inherit the
+    FOREIGN global and act on another session's identity.
+
+    So once the session-context machinery is engaged in this process (any host
+    has called set_session_vars), the session vars are ContextVar-authoritative:
+    - ContextVar set (incl. explicitly-empty "") → that value wins, overriding
+      any stale snapshot/global value.
+    - ContextVar _UNSET → STRIP the var from the child env rather than inherit
+      the possibly-foreign process-global.
+    In a pure single-process CLI/one-shot that never engaged the session-context
+    system there is no concurrency to leak across, so the inherited fallback is
+    kept. See gateway/session_context.session_context_engaged and
+    tests/tools/test_local_env_session_leak.py.
+    """
+    try:
+        from gateway.session_context import (
+            _UNSET,
+            _VAR_MAP,
+            session_context_engaged,
+        )
+    except Exception:
+        return
+
+    _engaged = session_context_engaged()
+    for var_name, var in _VAR_MAP.items():
+        value = var.get()
+        if value is not _UNSET:
+            # Explicitly bound (including "") — authoritative for this task.
+            env[var_name] = "" if value is None else str(value)
+        elif _engaged:
+            # Unset for THIS task while a concurrent host is engaged: drop any
+            # inherited global so a sibling session's value can't leak in.
+            env.pop(var_name, None)
+
+
 def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = None) -> dict:
     """Filter Hermes-managed secrets from a subprocess environment."""
     try:
@@ -298,6 +345,10 @@ def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = Non
     from hermes_constants import apply_subprocess_home_env
     apply_subprocess_home_env(sanitized)
 
+    # Same cross-session leak guard as _make_run_env, for the background/PTY
+    # spawn path (process_registry.spawn_local builds env via this function).
+    _inject_session_context_env(sanitized)
+
     for _marker in _ACTIVE_VENV_MARKER_VARS:
         sanitized.pop(_marker, None)
 
@@ -700,16 +751,10 @@ def _make_run_env(env: dict) -> dict:
     from hermes_constants import apply_subprocess_home_env
     apply_subprocess_home_env(run_env)
 
-    # Inject ContextVar-based session vars into subprocess env.
-    # ContextVars don't propagate to child processes, so we bridge them here.
-    try:
-        from gateway.session_context import _UNSET, _VAR_MAP
-        for var_name, var in _VAR_MAP.items():
-            value = var.get()
-            if value is not _UNSET and value:
-                run_env[var_name] = value
-    except Exception:
-        pass
+    # Bridge ContextVar-based session vars into the subprocess env (with the
+    # cross-session leak guard — strips _UNSET vars when a concurrent host is
+    # engaged so a sibling session's os.environ mirror can't leak in).
+    _inject_session_context_env(run_env)
 
     for _marker in _ACTIVE_VENV_MARKER_VARS:
         run_env.pop(_marker, None)