fix(security): scope passthrough env to routed profile

tests/tools/test_docker_environment.py

@@ -222,6 +222,78 @@ def test_init_env_args_uses_hermes_dotenv_for_empty_shell_env(monkeypatch):
     assert "MY_SECRET=" not in args
 
 
+def test_init_env_args_uses_active_profile_for_forwarded_env(monkeypatch):
+    """Docker forwarding must resolve the routed profile's secret scope."""
+    from agent import secret_scope as ss
+
+    env = _make_execute_only_env(forward_env=["SERVICE_TOKEN"])
+    monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
+    monkeypatch.setattr(docker_env, "_load_hermes_env_vars", lambda: {})
+    ss.set_multiplex_active(True)
+    token = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-routed-profile"})
+    try:
+        args = env._build_init_env_args()
+    finally:
+        ss.reset_secret_scope(token)
+        ss.set_multiplex_active(False)
+
+    assert "SERVICE_TOKEN=token-for-routed-profile" in args
+    assert "SERVICE_TOKEN=token-for-default" not in args
+
+
+def test_init_env_args_omits_missing_scoped_forwarded_env(monkeypatch):
+    """A missing routed secret must not reintroduce the process env value."""
+    from agent import secret_scope as ss
+
+    env = _make_execute_only_env(forward_env=["SERVICE_TOKEN"])
+    monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
+    monkeypatch.setattr(docker_env, "_load_hermes_env_vars", lambda: {})
+    ss.set_multiplex_active(True)
+    token = ss.set_secret_scope({})
+    try:
+        args = env._build_init_env_args()
+    finally:
+        ss.reset_secret_scope(token)
+        ss.set_multiplex_active(False)
+
+    assert "SERVICE_TOKEN=token-for-default" not in args
+    assert "SERVICE_TOKEN" not in args
+
+
+def test_runtime_exec_tracks_scope_and_clears_missing_value(monkeypatch):
+    """Shared Docker containers must refresh and clear profile-scoped values."""
+    from agent import secret_scope as ss
+
+    env = _make_execute_only_env(forward_env=["SERVICE_TOKEN"])
+    monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
+    monkeypatch.setattr(docker_env, "_load_hermes_env_vars", lambda: {})
+    calls = []
+    monkeypatch.setattr(
+        docker_env,
+        "_popen_bash",
+        lambda cmd, stdin_data=None: calls.append((cmd, stdin_data)) or object(),
+    )
+    ss.set_multiplex_active(True)
+    token = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-profile-a"})
+    try:
+        env._run_bash("printf '%s' \"$SERVICE_TOKEN\"")
+    finally:
+        ss.reset_secret_scope(token)
+
+    token = ss.set_secret_scope({})
+    try:
+        env._run_bash("printf '%s' \"${SERVICE_TOKEN-unset}\"")
+    finally:
+        ss.reset_secret_scope(token)
+        ss.set_multiplex_active(False)
+
+    first_cmd = calls[0][0]
+    assert "SERVICE_TOKEN=token-for-profile-a" in first_cmd
+    second_cmd = calls[1][0]
+    assert "SERVICE_TOKEN=token-for-profile-a" not in second_cmd
+    assert "unset SERVICE_TOKEN" in second_cmd[-1]
+
+
 # ── docker_env tests ──────────────────────────────────────────────
 
 

tests/tools/test_env_passthrough.py

@@ -4,12 +4,14 @@ import os
 import pytest
 import yaml
 
+from agent import secret_scope as ss
 import tools.env_passthrough as _ep_mod
 from tools.env_passthrough import (
     clear_env_passthrough,
     get_all_passthrough,
     is_env_passthrough,
     register_env_passthrough,
+    resolve_passthrough_value,
 )
 
 
@@ -18,9 +20,11 @@ def _clean_passthrough():
     """Ensure a clean passthrough state for every test."""
     clear_env_passthrough()
     _ep_mod._config_passthrough = None
+    ss.set_multiplex_active(False)
     yield
     clear_env_passthrough()
     _ep_mod._config_passthrough = None
+    ss.set_multiplex_active(False)
 
 
 class TestSkillScopedPassthrough:
@@ -40,7 +44,7 @@ class TestConfigPassthrough:
     def test_reads_from_config(self, tmp_path, monkeypatch):
         config = {"terminal": {"env_passthrough": ["MY_CUSTOM_KEY", "ANOTHER_TOKEN"]}}
         config_path = tmp_path / "config.yaml"
-        config_path.write_text(yaml.dump(config))
+        config_path.write_text(yaml.dump(config), encoding="utf-8")
         monkeypatch.setenv("HERMES_HOME", str(tmp_path))
         _ep_mod._config_passthrough = None
 
@@ -52,7 +56,7 @@ class TestConfigPassthrough:
     def test_union_of_skill_and_config(self, tmp_path, monkeypatch):
         config = {"terminal": {"env_passthrough": ["CONFIG_KEY"]}}
         config_path = tmp_path / "config.yaml"
-        config_path.write_text(yaml.dump(config))
+        config_path.write_text(yaml.dump(config), encoding="utf-8")
         monkeypatch.setenv("HERMES_HOME", str(tmp_path))
         _ep_mod._config_passthrough = None
 
@@ -62,6 +66,42 @@ class TestConfigPassthrough:
         assert "SKILL_KEY" in all_pt
 
 
+class TestProfileScopedResolution:
+    def test_active_scope_overrides_process_fallback(self):
+        ss.set_multiplex_active(True)
+        token = ss.set_secret_scope({"SERVICE_TOKEN": "profile-b"})
+        try:
+            assert resolve_passthrough_value("SERVICE_TOKEN", "profile-a") == "profile-b"
+        finally:
+            ss.reset_secret_scope(token)
+
+    def test_active_scope_does_not_fall_back_to_another_profile(self):
+        ss.set_multiplex_active(True)
+        token = ss.set_secret_scope({})
+        try:
+            assert resolve_passthrough_value("SERVICE_TOKEN", "profile-a") is None
+        finally:
+            ss.reset_secret_scope(token)
+
+    def test_unscoped_multiplex_read_fails_closed(self):
+        ss.set_multiplex_active(True)
+        with pytest.raises(ss.UnscopedSecretError):
+            resolve_passthrough_value("SERVICE_TOKEN", "profile-a")
+
+    def test_single_profile_keeps_callers_fallback(self):
+        assert resolve_passthrough_value("SERVICE_TOKEN", "profile-a") == "profile-a"
+
+    def test_active_scope_keeps_explicit_global_override(self, monkeypatch):
+        """Global terminal settings still honor a caller-provided override."""
+        monkeypatch.setenv("TERMINAL_CWD", "/default")
+        ss.set_multiplex_active(True)
+        token = ss.set_secret_scope({})
+        try:
+            assert resolve_passthrough_value("TERMINAL_CWD", "/explicit") == "/explicit"
+        finally:
+            ss.reset_secret_scope(token)
+
+
 class TestExecuteCodeIntegration:
     """Verify that the passthrough is checked in execute_code's env filtering."""
 
@@ -114,10 +154,114 @@ class TestExecuteCodeIntegration:
         assert "TENOR_API_KEY" in child_env
         assert child_env["TENOR_API_KEY"] == "test123"
 
+    def test_execute_code_uses_active_profile_for_passthrough(self, monkeypatch):
+        """The execute_code child must receive the routed profile's value."""
+        from tools.code_execution_tool import _scrub_child_env
+
+        register_env_passthrough(["SERVICE_TOKEN"])
+        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
+        ss.set_multiplex_active(True)
+        token = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-routed-profile"})
+        try:
+            child_env = _scrub_child_env({"SERVICE_TOKEN": "token-for-default"})
+        finally:
+            ss.reset_secret_scope(token)
+            ss.set_multiplex_active(False)
+
+        assert child_env["SERVICE_TOKEN"] == "token-for-routed-profile"
+
+    def test_execute_code_omits_missing_scoped_passthrough(self, monkeypatch):
+        """A missing routed secret must not leak into the execute_code child."""
+        from tools.code_execution_tool import _scrub_child_env
+
+        register_env_passthrough(["SERVICE_TOKEN"])
+        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
+        ss.set_multiplex_active(True)
+        token = ss.set_secret_scope({})
+        try:
+            child_env = _scrub_child_env({"SERVICE_TOKEN": "token-for-default"})
+        finally:
+            ss.reset_secret_scope(token)
+            ss.set_multiplex_active(False)
+
+        assert "SERVICE_TOKEN" not in child_env
+
 
 class TestTerminalIntegration:
     """Verify that the passthrough is checked in terminal's env sanitizers."""
 
+    def test_background_terminal_uses_active_profile_for_passthrough(self, monkeypatch):
+        """Background/PTY terminal children must use the routed profile value."""
+        from tools.environments.local import _sanitize_subprocess_env
+
+        register_env_passthrough(["SERVICE_TOKEN"])
+        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
+        ss.set_multiplex_active(True)
+        token = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-routed-profile"})
+        try:
+            child_env = _sanitize_subprocess_env(
+                {"SERVICE_TOKEN": "token-for-default"},
+                {"SERVICE_TOKEN": "token-for-default"},
+            )
+        finally:
+            ss.reset_secret_scope(token)
+            ss.set_multiplex_active(False)
+
+        assert child_env["SERVICE_TOKEN"] == "token-for-routed-profile"
+
+    def test_background_terminal_omits_missing_scoped_passthrough(self, monkeypatch):
+        """A missing routed secret must not leak into background terminal work."""
+        from tools.environments.local import _sanitize_subprocess_env
+
+        register_env_passthrough(["SERVICE_TOKEN"])
+        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
+        ss.set_multiplex_active(True)
+        token = ss.set_secret_scope({})
+        try:
+            child_env = _sanitize_subprocess_env({"SERVICE_TOKEN": "token-for-default"})
+        finally:
+            ss.reset_secret_scope(token)
+            ss.set_multiplex_active(False)
+
+        assert "SERVICE_TOKEN" not in child_env
+
+    def test_shared_local_snapshot_re_resolves_current_profile(self, monkeypatch, tmp_path):
+        """A persistent shell snapshot must not retain the previous profile's value."""
+        from tools.environments.local import LocalEnvironment
+
+        register_env_passthrough(["SERVICE_TOKEN"])
+        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
+        ss.set_multiplex_active(True)
+        env = None
+        token_b = None
+        token_c = None
+        try:
+            token_a = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-profile-a"})
+            try:
+                env = LocalEnvironment(cwd=str(tmp_path))
+                assert env.execute("printf '%s' \"$SERVICE_TOKEN\"")["output"] == "token-for-profile-a"
+            finally:
+                ss.reset_secret_scope(token_a)
+
+            token_b = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-profile-b"})
+            result = env.execute("printf '%s' \"$SERVICE_TOKEN\"")
+            ss.reset_secret_scope(token_b)
+            token_b = None
+
+            token_c = ss.set_secret_scope({})
+            missing = env.execute("printf '%s' \"${SERVICE_TOKEN-unset}\"")
+        finally:
+            if token_b is not None:
+                ss.reset_secret_scope(token_b)
+            if token_c is not None:
+                ss.reset_secret_scope(token_c)
+            ss.set_multiplex_active(False)
+            if env is not None:
+                env.cleanup()
+
+        assert result["output"] == "token-for-profile-b"
+        assert missing["output"] == "unset"
+
     def test_blocklisted_var_blocked_by_default(self):
         from tools.environments.local import _sanitize_subprocess_env, _HERMES_PROVIDER_ENV_BLOCKLIST
 

tests/tools/test_local_env_blocklist.py

@@ -313,6 +313,48 @@ class TestActiveVenvMarkerStripping:
         assert "CONDA_PREFIX" in _ACTIVE_VENV_MARKER_VARS
 
 
+class TestProfileScopedPassthrough:
+    def test_make_run_env_uses_active_profile_for_passthrough(self, monkeypatch):
+        """Allowlisted values must come from the routed profile, not os.environ."""
+        from agent import secret_scope as ss
+        from tools.env_passthrough import clear_env_passthrough, register_env_passthrough
+        from tools.environments.local import _make_run_env
+
+        clear_env_passthrough()
+        register_env_passthrough(["SERVICE_TOKEN"])
+        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
+        ss.set_multiplex_active(True)
+        token = ss.set_secret_scope({"SERVICE_TOKEN": "token-for-routed-profile"})
+        try:
+            result = _make_run_env({})
+        finally:
+            ss.reset_secret_scope(token)
+            ss.set_multiplex_active(False)
+            clear_env_passthrough()
+
+        assert result["SERVICE_TOKEN"] == "token-for-routed-profile"
+
+    def test_make_run_env_omits_missing_scoped_passthrough(self, monkeypatch):
+        """A missing routed secret must not fall back to the default profile."""
+        from agent import secret_scope as ss
+        from tools.env_passthrough import clear_env_passthrough, register_env_passthrough
+        from tools.environments.local import _make_run_env
+
+        clear_env_passthrough()
+        register_env_passthrough(["SERVICE_TOKEN"])
+        monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
+        ss.set_multiplex_active(True)
+        token = ss.set_secret_scope({})
+        try:
+            result = _make_run_env({})
+        finally:
+            ss.reset_secret_scope(token)
+            ss.set_multiplex_active(False)
+            clear_env_passthrough()
+
+        assert "SERVICE_TOKEN" not in result
+
+
 class TestBlocklistCoverage:
     """Sanity checks that the blocklist covers all known providers."""
 

tools/code_execution_tool.py

@@ -208,7 +208,9 @@ def _scrub_child_env(source_env, is_passthrough=None, is_windows=None):
     """Produce the scrubbed child-process env for execute_code.
 
     Rules (order matters):
-      1. Passthrough vars (skill- or config-declared) always pass.
+      1. Passthrough vars (skill- or config-declared) pass through the active
+         profile secret scope; an absent scoped value is omitted and an
+         unscoped multiplex read fails closed.
       2. Secret-substring names (KEY/TOKEN/DSN/WEBHOOK/etc.) are blocked.
       3. Names matching a safe prefix pass.
       4. Operational HERMES_* vars (_HERMES_CHILD_ALLOWED) pass by exact name.
@@ -219,12 +221,22 @@ def _scrub_child_env(source_env, is_passthrough=None, is_windows=None):
     Extracted into a helper so tests can exercise the logic without
     spawning a subprocess.
     """
+    resolve_passthrough_value = None
     if is_passthrough is None:
         try:
-            from tools.env_passthrough import is_env_passthrough as _ep
+            from tools.env_passthrough import (
+                is_env_passthrough as _ep,
+                resolve_passthrough_value,
+            )
         except Exception:
             _ep = lambda _: False  # noqa: E731
+            resolve_passthrough_value = lambda _name, _fallback: None  # noqa: E731
         is_passthrough = _ep
+    else:
+        try:
+            from tools.env_passthrough import resolve_passthrough_value
+        except Exception:
+            resolve_passthrough_value = lambda _name, _fallback: None  # noqa: E731
     if is_windows is None:
         is_windows = _IS_WINDOWS
 
@@ -239,7 +251,9 @@ def _scrub_child_env(source_env, is_passthrough=None, is_windows=None):
     _dropped_hermes = []
     for k, v in source_env.items():
         if is_passthrough(k):
-            scrubbed[k] = v
+            resolved = resolve_passthrough_value(k, v)
+            if resolved is not None:
+                scrubbed[k] = resolved
             continue
         if any(s in k.upper() for s in _SECRET_SUBSTRINGS):
             continue

tools/env_passthrough.py

@@ -15,6 +15,8 @@ Two sources feed the allowlist:
 
 Both ``code_execution_tool.py`` and ``tools/environments/local.py`` consult
 :func:`is_env_passthrough` before stripping a variable.
+When profile multiplexing is active, their forwarded values are resolved
+through the current profile's secret scope rather than the process environment.
 """
 
 from __future__ import annotations
@@ -177,8 +179,45 @@ def get_all_passthrough() -> frozenset[str]:
     return frozenset(_get_allowed()) | _load_config_passthrough()
 
 
+def resolve_passthrough_value(
+    name: str,
+    fallback: str | None = None,
+) -> str | None:
+    """Resolve an allowlisted variable without crossing profile boundaries.
+
+    ``fallback`` is the value the caller would have forwarded before profile
+    secret scopes existed (typically a snapshot of ``os.environ`` or the
+    current profile's ``.env``).  An active multiplex scope is authoritative:
+    a missing key returns ``None`` and never falls back to the process-global
+    environment.  An unscoped read while multiplexing is active raises the
+    fail-closed ``UnscopedSecretError`` from :mod:`agent.secret_scope`.
+
+    Outside multiplexing, an installed scope keeps the existing overlay
+    semantics and an unscoped caller keeps its already-resolved fallback.
+    """
+    from agent.secret_scope import (
+        _is_global_env,
+        current_secret_scope,
+        get_secret,
+        is_multiplex_active,
+    )
+
+    # Global terminal/runtime settings are not profile secrets.  ``fallback``
+    # is already the caller's effective value (including an explicit per-call
+    # override), so preserve it instead of replacing it with the process-wide
+    # value while a multiplex scope is active.
+    if _is_global_env(name) and fallback is not None:
+        return fallback
+
+    scope = current_secret_scope()
+    multiplex_active = is_multiplex_active()
+    if scope is None:
+        if multiplex_active:
+            return get_secret(name)
+        return fallback
+    return get_secret(name, None if multiplex_active else fallback)
+
+
 def clear_env_passthrough() -> None:
     """Reset the skill-scoped allowlist (e.g. on session reset)."""
     _get_allowed().clear()
-
-

tools/environments/base.py

@@ -10,6 +10,7 @@ import codecs
 import json
 import logging
 import os
+import re
 import select
 import shlex
 import subprocess
@@ -19,7 +20,7 @@ import uuid
 from abc import ABC, abstractmethod
 from collections import deque
 from pathlib import Path
-from typing import IO, Callable, Protocol
+from typing import IO, Callable, Iterable, Protocol
 
 from hermes_constants import get_hermes_home
 from hermes_cli._subprocess_compat import windows_hide_flags
@@ -403,11 +404,16 @@ def _cwd_marker(session_id: str) -> str:
 _SNAPSHOT_EXCLUDED_ENV_REGEX = (
     "^declare -x (HERMES_SESSION_|HERMES_UI_SESSION_ID|HERMES_CRON_AUTO_DELIVER_)"
 )
+_SHELL_ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
 
 
-def _export_dump_excluding_session_vars(tmp_path: str) -> str:
+def _export_dump_excluding_session_vars(
+    tmp_path: str,
+    excluded_names: Iterable[str] = (),
+) -> str:
     """Return a shell snippet that dumps ``export -p`` to *tmp_path* minus the
-    per-session bridged vars (see ``_SNAPSHOT_EXCLUDED_ENV_REGEX``).
+    per-session bridged vars (see ``_SNAPSHOT_EXCLUDED_ENV_REGEX``) and any
+    additional names supplied by the caller.
 
     Unset the bridged vars in a subshell *before* ``export -p``. A line-based
     ``grep -vE`` filter is unsafe: bash 3.2 prints a value containing a newline
@@ -428,10 +434,19 @@ def _export_dump_excluding_session_vars(tmp_path: str) -> str:
     """
     # ${!PREFIX*} is bash 3.2+ name-prefix expansion; empty matches are fine
     # because ``unset`` with only missing names is ignored under 2>/dev/null.
+    # Quote caller-provided names so malformed configuration can never become
+    # shell syntax. Valid environment names remain unquoted by shlex.quote().
+    safe_names = {
+        name for name in excluded_names
+        if isinstance(name, str) and name
+    }
+    extra_unset = " ".join(shlex.quote(name) for name in sorted(safe_names))
+    if extra_unset:
+        extra_unset = f" {extra_unset}"
     return (
         "{ ( "
         "unset ${!HERMES_SESSION_*} ${!HERMES_CRON_AUTO_DELIVER_*} "
-        "HERMES_UI_SESSION_ID 2>/dev/null; "
+        f"HERMES_UI_SESSION_ID{extra_unset} 2>/dev/null; "
         "export -p; "
         ") || true; } "
         f"> {tmp_path}"
@@ -457,6 +472,11 @@ class BaseEnvironment(ABC):
     # Snapshot creation timeout (override for slow cold-starts).
     _snapshot_timeout: int = 30
 
+    # Local and Docker override this because they resolve allowlisted values
+    # through the active profile scope. Other backends keep their existing
+    # snapshot semantics until they implement the same resolver contract.
+    _profile_scoped_passthrough: bool = False
+
     def get_temp_dir(self) -> str:
         """Return the backend temp directory used for session artifacts.
 
@@ -477,6 +497,7 @@ class BaseEnvironment(ABC):
         self._cwd_file = f"{temp_dir}/hermes-cwd-{self._session_id}.txt"
         self._cwd_marker = _cwd_marker(self._session_id)
         self._snapshot_ready = False
+        self._snapshot_passthrough_names: set[str] = set()
         # When True, login bash is unusable (e.g. broken Git-for-Windows
         # ``Directory \\drivers\\etc`` startup) so execute() must not fall
         # back to ``bash -l`` per command — use non-login ``bash -c`` instead.
@@ -510,6 +531,32 @@ class BaseEnvironment(ABC):
     # Session snapshot (init_session)
     # ------------------------------------------------------------------
 
+    def _snapshot_excluded_passthrough_names(self) -> tuple[str, ...]:
+        """Return profile-scoped names that must not persist in the snapshot.
+
+        The set is monotonic for the environment lifetime. A skill/config
+        allowlist can be cleared after a value was captured; retaining the
+        exclusion prevents that old value from becoming visible to a later
+        profile through the shared snapshot.
+        """
+        if not self._profile_scoped_passthrough:
+            return ()
+        try:
+            from agent.secret_scope import is_multiplex_active
+            if is_multiplex_active():
+                from tools.env_passthrough import get_all_passthrough
+                self._snapshot_passthrough_names.update(
+                    name
+                    for name in get_all_passthrough()
+                    if isinstance(name, str) and _SHELL_ENV_NAME_RE.fullmatch(name)
+                )
+        except Exception:
+            logger.debug(
+                "Could not refresh profile-scoped snapshot exclusions",
+                exc_info=True,
+            )
+        return tuple(sorted(self._snapshot_passthrough_names))
+
     def init_session(self):
         """Capture login shell environment into a snapshot file.
 
@@ -552,9 +599,10 @@ class BaseEnvironment(ABC):
         # static path is shell-quoted (Windows/Git-Bash drive letters, spaces)
         # with ``$BASHPID`` left outside the quotes so it still expands.
         _snap_tmp = self._quote_shell_path(self._snapshot_path + ".tmp.") + "$BASHPID"
+        snapshot_excluded = self._snapshot_excluded_passthrough_names()
         bootstrap = (
             f"umask 077\n"
-            f"{_export_dump_excluding_session_vars(_snap_tmp)}\n"
+            f"{_export_dump_excluding_session_vars(_snap_tmp, snapshot_excluded)}\n"
             # Dump function definitions, filtering out private (``_``-prefixed)
             # helpers — mainly bash-completion internals (``_git``, ``_make``…)
             # — by NAME, not by line.  A naive ``declare -f | grep -vE '^_[^_]'``
@@ -669,6 +717,21 @@ class BaseEnvironment(ABC):
         _snap_tmp = self._quote_shell_path(self._snapshot_path + ".tmp.") + "$BASHPID"
 
         parts = []
+        passthrough_names = self._snapshot_excluded_passthrough_names()
+
+        # A shared snapshot may contain the previous profile's value. Save
+        # the current process environment before sourcing it, then restore the
+        # current profile's value (or unset the name) immediately afterwards.
+        # Values stay in environment memory and never enter the shell command
+        # string, so secrets are not exposed through process arguments/logs.
+        saved_names: list[tuple[str, str, str]] = []
+        for name in passthrough_names:
+            marker = f"_HERMES_RUNTIME_PASSTHROUGH_{name}"
+            present = f"{marker}_PRESENT"
+            value = f"{marker}_VALUE"
+            saved_names.append((name, present, value))
+            parts.append(f"{present}=${{{name}+x}}")
+            parts.append(f"{value}=${{{name}-}}")
 
         # Source snapshot (env vars from previous commands).
         # Redirect stdout to /dev/null: on macOS (bash 3.2 and certain
@@ -681,6 +744,13 @@ class BaseEnvironment(ABC):
                 f"source {_quoted_snap} >/dev/null 2>&1 || true"
             )
 
+        for name, present, value in saved_names:
+            parts.append(
+                f'if [ "${present}" = x ]; then export {name}="${value}"; '
+                f'else unset {name}; fi'
+            )
+            parts.append(f"unset {present} {value}")
+
         # Preserve bare ``~`` expansion, but rewrite ``~/...`` through
         # ``$HOME`` so suffixes with spaces remain a single shell word.
         quoted_cwd = self._quote_cwd_for_cd(cwd)
@@ -705,7 +775,7 @@ class BaseEnvironment(ABC):
         # _export_dump_excluding_session_vars.
         if self._snapshot_ready:
             parts.append(
-                f"{{ {_export_dump_excluding_session_vars(_snap_tmp)} "
+                f"{{ {_export_dump_excluding_session_vars(_snap_tmp, passthrough_names)} "
                 f"&& mv -f {_snap_tmp} {_quoted_snap}; }} "
                 f"2>/dev/null || rm -f {_snap_tmp} 2>/dev/null || true"
             )

tools/environments/docker.py

@@ -10,6 +10,7 @@ import json
 import logging
 import os
 import re
+import shlex
 import shutil
 import subprocess
 import sys
@@ -843,6 +844,8 @@ class DockerEnvironment(BaseEnvironment):
     across container restarts.
     """
 
+    _profile_scoped_passthrough = True
+
     def __init__(
         self,
         image: str,
@@ -872,6 +875,7 @@ class DockerEnvironment(BaseEnvironment):
         self._task_id = task_id
         self._forward_env = _normalize_forward_env_names(forward_env)
         self._env = _normalize_env_dict(env)
+        self._init_unset_passthrough_names: tuple[str, ...] = ()
         self._container_id: Optional[str] = None
         self._labels: dict[str, str] = {}
         self._image: str = ""
@@ -1488,9 +1492,7 @@ class DockerEnvironment(BaseEnvironment):
             self._container_id = result.stdout.strip()
             logger.info(f"Started container {container_name} ({self._container_id[:12]})")
 
-        # Build the init-time env forwarding args (used only by init_session
-        # to inject host env vars into the snapshot; subsequent commands get
-        # them from the snapshot file).
+        # Build the init-time env forwarding args used to seed the snapshot.
         self._init_env_args = self._build_init_env_args()
 
         # Initialize session snapshot inside the container
@@ -1499,15 +1501,41 @@ class DockerEnvironment(BaseEnvironment):
     def _build_init_env_args(self) -> list[str]:
         """Build -e KEY=VALUE args for injecting host env vars into init_session.
 
-        These are used once during init_session() so that export -p captures
-        them into the snapshot.  Subsequent execute() calls don't need -e flags.
+        These are used during init_session() so that export -p captures the
+        configured environment and the current profile's forwarded values.
         """
+        passthrough_env, unset_names = self._resolve_passthrough_env()
         exec_env: dict[str, str] = dict(self._env)
+        exec_env.update(passthrough_env)
+        for name in unset_names:
+            exec_env.pop(name, None)
+        self._init_unset_passthrough_names = tuple(sorted(unset_names))
 
+        args = []
+        for key in sorted(exec_env):
+            args.extend(["-e", f"{key}={exec_env[key]}"])
+        return args
+
+    def _build_passthrough_env(self) -> dict[str, str]:
+        """Resolve forwarded host variables through the active profile scope."""
+        return self._resolve_passthrough_env()[0]
+
+    def _resolve_passthrough_env(self) -> tuple[dict[str, str], set[str]]:
+        """Return forwarded values and scoped names that must be unset."""
+        exec_env: dict[str, str] = {}
         explicit_forward_keys = set(self._forward_env)
         passthrough_keys: set[str] = set()
+        resolve_passthrough_value = None
+        multiplex_active = False
+        is_global_env = lambda _name: False  # noqa: E731
         try:
-            from tools.env_passthrough import get_all_passthrough
+            from tools.env_passthrough import (
+                get_all_passthrough,
+                resolve_passthrough_value,
+            )
+            from agent.secret_scope import _is_global_env, is_multiplex_active as _is_multiplex_active
+            is_global_env = _is_global_env
+            multiplex_active = _is_multiplex_active()
             passthrough_keys = set(get_all_passthrough())
         except Exception:
             pass
@@ -1521,17 +1549,28 @@ class DockerEnvironment(BaseEnvironment):
         }
         forward_keys = explicit_forward_keys | (_implicit_forward - _HERMES_PROVIDER_ENV_BLOCKLIST)
         hermes_env = _load_hermes_env_vars() if forward_keys else {}
+        unset_names: set[str] = set()
         for key in sorted(forward_keys):
-            value = os.getenv(key)
-            if not value:
-                value = hermes_env.get(key)
-            if value:
+            value = os.getenv(key) or hermes_env.get(key)
+            if resolve_passthrough_value is not None:
+                value = resolve_passthrough_value(key, value)
+            if value is not None:
                 exec_env[key] = value
+            elif multiplex_active and not is_global_env(key) and _ENV_VAR_NAME_RE.fullmatch(key):
+                unset_names.add(key)
+        return exec_env, unset_names
 
+    def _build_runtime_env_args_with_unsets(self) -> tuple[list[str], tuple[str, ...]]:
+        """Build runtime forwarding args plus names absent from the active scope."""
+        passthrough_env, unset_names = self._resolve_passthrough_env()
         args = []
-        for key in sorted(exec_env):
-            args.extend(["-e", f"{key}={exec_env[key]}"])
-        return args
+        for key in sorted(passthrough_env):
+            args.extend(["-e", f"{key}={passthrough_env[key]}"])
+        return args, tuple(sorted(unset_names))
+
+    def _build_runtime_env_args(self) -> list[str]:
+        """Build only dynamic forwarded values for a non-login command."""
+        return self._build_runtime_env_args_with_unsets()[0]
 
     def _run_bash(self, cmd_string: str, *, login: bool = False,
                   timeout: int = 120,
@@ -1542,10 +1581,21 @@ class DockerEnvironment(BaseEnvironment):
         if stdin_data is not None:
             cmd.append("-i")
 
-        # Only inject -e env args during init_session (login=True).
-        # Subsequent commands get env vars from the snapshot.
+        # Init seeds the snapshot. Profile-scoped passthrough values are also
+        # injected on every later command because this container can be shared
+        # by multiple routed profiles in one gateway process.
+        unset_names: tuple[str, ...] = ()
         if login:
             cmd.extend(self._init_env_args)
+        elif self._profile_scoped_passthrough:
+            runtime_args, unset_names = self._build_runtime_env_args_with_unsets()
+            cmd.extend(runtime_args)
+
+        if login:
+            unset_names = getattr(self, "_init_unset_passthrough_names", ())
+        if unset_names:
+            quoted_names = " ".join(shlex.quote(name) for name in unset_names)
+            cmd_string = f"unset {quoted_names} 2>/dev/null || true\n{cmd_string}"
 
         cmd.extend([self._container_id])
 

tools/environments/local.py

@@ -456,9 +456,13 @@ def _inject_session_context_env(env: dict) -> None:
 def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = None) -> dict:
     """Filter Hermes-managed secrets from a subprocess environment."""
     try:
-        from tools.env_passthrough import is_env_passthrough as _is_passthrough
+        from tools.env_passthrough import (
+            is_env_passthrough as _is_passthrough,
+            resolve_passthrough_value as _resolve_passthrough_value,
+        )
     except Exception:
         _is_passthrough = lambda _: False  # noqa: E731
+        _resolve_passthrough_value = lambda _name, fallback: fallback  # noqa: E731
 
     sanitized: dict[str, str] = {}
 
@@ -467,8 +471,12 @@ def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = Non
             continue
         if _is_hermes_internal_secret(key):
             continue
-        if key not in _HERMES_PROVIDER_ENV_BLOCKLIST or _is_passthrough(key):
-            sanitized[key] = value
+        passthrough = _is_passthrough(key)
+        if key in _HERMES_PROVIDER_ENV_BLOCKLIST and not passthrough:
+            continue
+        resolved = _resolve_passthrough_value(key, value) if passthrough else value
+        if resolved is not None:
+            sanitized[key] = resolved
 
     for key, value in (extra_env or {}).items():
         if key.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX):
@@ -478,8 +486,13 @@ def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = Non
             sanitized[real_key] = value
         elif _is_hermes_internal_secret(key):
             continue
-        elif key not in _HERMES_PROVIDER_ENV_BLOCKLIST or _is_passthrough(key):
-            sanitized[key] = value
+        else:
+            passthrough = _is_passthrough(key)
+            if key in _HERMES_PROVIDER_ENV_BLOCKLIST and not passthrough:
+                continue
+            resolved = _resolve_passthrough_value(key, value) if passthrough else value
+            if resolved is not None:
+                sanitized[key] = resolved
 
     _inject_context_hermes_home(sanitized)
 
@@ -1255,9 +1268,13 @@ def _path_env_key(run_env: dict) -> str | None:
 def _make_run_env(env: dict) -> dict:
     """Build a run environment with a sane PATH and provider-var stripping."""
     try:
-        from tools.env_passthrough import is_env_passthrough as _is_passthrough
+        from tools.env_passthrough import (
+            is_env_passthrough as _is_passthrough,
+            resolve_passthrough_value as _resolve_passthrough_value,
+        )
     except Exception:
         _is_passthrough = lambda _: False  # noqa: E731
+        _resolve_passthrough_value = lambda _name, fallback: fallback  # noqa: E731
 
     merged = dict(os.environ | env)
     run_env = {}
@@ -1269,8 +1286,13 @@ def _make_run_env(env: dict) -> dict:
             run_env[real_key] = v
         elif _is_hermes_internal_secret(k):
             continue
-        elif k not in _HERMES_PROVIDER_ENV_BLOCKLIST or _is_passthrough(k):
-            run_env[k] = v
+        else:
+            passthrough = _is_passthrough(k)
+            if k in _HERMES_PROVIDER_ENV_BLOCKLIST and not passthrough:
+                continue
+            value = _resolve_passthrough_value(k, v) if passthrough else v
+            if value is not None:
+                run_env[k] = value
     path_key = _path_env_key(run_env)
     if path_key is not None:
         new_path = _append_missing_sane_path_entries(run_env.get(path_key, ""))
@@ -1397,6 +1419,8 @@ class LocalEnvironment(BaseEnvironment):
     CWD persists via file-based read after each command.
     """
 
+    _profile_scoped_passthrough = True
+
     def __init__(self, cwd: str = "", timeout: int = 60, env: dict = None):
         cwd = _resolve_local_initial_cwd(cwd)
         super().__init__(cwd=cwd, timeout=timeout, env=env)