fix(gateway): hydrate cold profile secret sources

agent/secret_sources/base.py

@@ -39,17 +39,36 @@ from __future__ import annotations
 import os
 import re
 import subprocess
+from contextvars import ContextVar, Token
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from enum import Enum
 from pathlib import Path
-from typing import Dict, FrozenSet, List, Optional, Sequence
+from typing import Dict, FrozenSet, List, MutableMapping, Optional, Sequence
 
 # Bump ONLY for breaking changes to the required contract surface
 # (abstract-method signatures, FetchResult required fields).  Additive
 # optional hooks must ship with defaults and must NOT bump this.
 SECRET_SOURCE_API_VERSION = 1
 
+_SOURCE_ENVIRONMENT: ContextVar[Optional[MutableMapping[str, str]]]
+_SOURCE_ENVIRONMENT = ContextVar("hermes_secret_source_environment", default=None)
+
+
+def set_source_environment(environ: MutableMapping[str, str]) -> Token:
+    """Install a per-fetch environment view without changing ``os.environ``."""
+    return _SOURCE_ENVIRONMENT.set(environ)
+
+
+def reset_source_environment(token: Token) -> None:
+    _SOURCE_ENVIRONMENT.reset(token)
+
+
+def get_source_environment() -> MutableMapping[str, str]:
+    """Return the active per-fetch environment, or the process environment."""
+    environ = _SOURCE_ENVIRONMENT.get()
+    return environ if environ is not None else os.environ
+
 # Timeout the orchestrator enforces around fetch() when the source's
 # config section doesn't override it.  Generous because a first run may
 # include a one-time CLI binary auto-install (e.g. bws download+verify).

agent/secret_sources/bitwarden.py

@@ -58,6 +58,7 @@ from agent.secret_sources._cache import (
     is_valid_env_name as _is_valid_env_name,
 )
 from agent.secret_sources.base import ErrorKind, SecretSource
+from agent.secret_sources.base import get_source_environment
 
 logger = logging.getLogger(__name__)
 
@@ -667,10 +668,15 @@ def _run_bws_list(
     bws: Path, access_token: str, project_id: str, server_url: str = ""
 ) -> Tuple[Dict[str, str], List[str]]:
     cmd = [str(bws), "secret", "list", project_id, "--output", "json"]
-    # bws child intentionally receives the access token; exact preservation
-    # (BWS_SERVER_URL manual overrides etc. must survive untouched).
-    from tools.environments.local import build_subprocess_env
-    env = build_subprocess_env(scrub_secrets=False, inherit_profile_home=False)
+    # bws child intentionally receives the access token.  Under a profile-local
+    # fetch it must not inherit sibling credentials from process-global env.
+    source_env = get_source_environment()
+    if source_env is os.environ:
+        from tools.environments.local import build_subprocess_env
+
+        env = build_subprocess_env(scrub_secrets=False, inherit_profile_home=False)
+    else:
+        env = dict(source_env)
     env["BWS_ACCESS_TOKEN"] = access_token
     # Make sure we're not echoing telemetry / colour codes into json.
     env.setdefault("NO_COLOR", "1")
@@ -908,7 +914,7 @@ class BitwardenSource(SecretSource):
         result = FetchResult()
 
         access_token_env = str(cfg.get("access_token_env") or "BWS_ACCESS_TOKEN")
-        access_token = os.environ.get(access_token_env, "").strip()
+        access_token = get_source_environment().get(access_token_env, "").strip()
         if not access_token:
             result.error = (
                 f"secrets.bitwarden.enabled is true but {access_token_env} is "

agent/secret_sources/command.py

@@ -44,6 +44,7 @@ from typing import Dict, Optional
 # Reuse the exact result shape the bitwarden source returns so
 # hermes_cli.env_loader can consume both providers identically.
 from agent.secret_sources.base import ErrorKind, SecretSource
+from agent.secret_sources.base import get_source_environment
 from agent.secret_sources.bitwarden import FetchResult
 
 __all__ = [
@@ -181,8 +182,17 @@ def _run_helper(
 
     # User-configured secret-helper command: runs with the user's full shell
     # env by design (it may need any credential to resolve the secret).
-    from tools.environments.local import build_subprocess_env
-    env = build_subprocess_env(scrub_secrets=False, inherit_profile_home=False)
+    source_env = get_source_environment()
+    if source_env is os.environ:
+        # Legacy single-profile startup intentionally preserves the existing
+        # helper contract, which may rely on the user's full environment.
+        from tools.environments.local import build_subprocess_env
+        env = build_subprocess_env(scrub_secrets=False, inherit_profile_home=False)
+    else:
+        # A multiplex profile must never inherit sibling secrets from the
+        # process-global environment.  hydrate_profile_secret_sources seeds
+        # only global-safe values plus this profile's own .env.
+        env = dict(source_env)
     env["HERMES_SECRET_KEY"] = secret_key
 
     try:

agent/secret_sources/onepassword.py

@@ -55,6 +55,7 @@ from agent.secret_sources._cache import (
     is_valid_env_name,
 )
 from agent.secret_sources.base import ErrorKind, SecretSource
+from agent.secret_sources.base import get_source_environment
 
 logger = logging.getLogger(__name__)
 
@@ -182,15 +183,16 @@ def _auth_fingerprint(token_env: str) -> str:
     previous identity is never served under a new one.  Never logged or
     displayed; the raw token never leaves this hash.
     """
+    source_env = get_source_environment()
     parts: List[str] = [
-        f"token={os.environ.get(token_env, '')}",
-        f"account={os.environ.get('OP_ACCOUNT', '')}",
-        f"connect_host={os.environ.get('OP_CONNECT_HOST', '')}",
-        f"connect_token={os.environ.get('OP_CONNECT_TOKEN', '')}",
+        f"token={source_env.get(token_env, '')}",
+        f"account={source_env.get('OP_ACCOUNT', '')}",
+        f"connect_host={source_env.get('OP_CONNECT_HOST', '')}",
+        f"connect_token={source_env.get('OP_CONNECT_TOKEN', '')}",
     ]
-    for key in sorted(os.environ):
+    for key in sorted(source_env):
         if key.startswith("OP_SESSION_"):
-            parts.append(f"{key}={os.environ[key]}")
+            parts.append(f"{key}={source_env[key]}")
     material = "\n".join(parts)
     return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]
 
@@ -238,13 +240,14 @@ def _scrub(text: str) -> str:
 
 def _op_child_env(token_value: str) -> Dict[str, str]:
     """Build a minimal allowlisted environment for the ``op`` child process."""
+    source_env = get_source_environment()
     env: Dict[str, str] = {}
     for key in _OP_ENV_ALLOWLIST:
-        val = os.environ.get(key)
+        val = source_env.get(key)
         if val is not None:
             env[key] = val
     # Desktop / interactive session credentials.
-    for key, val in os.environ.items():
+    for key, val in source_env.items():
         if key.startswith("OP_SESSION_"):
             env[key] = val
     # `op` reads OP_SERVICE_ACCOUNT_TOKEN regardless of which env var the user
@@ -340,7 +343,7 @@ def fetch_onepassword_secrets(
     if not valid:
         return {}, warnings
 
-    token_value = os.environ.get(token_env, "").strip()
+    token_value = get_source_environment().get(token_env, "").strip()
     cache_key: _CacheKey = (
         _auth_fingerprint(token_env),
         account or "",

agent/secret_sources/registry.py

@@ -32,7 +32,7 @@ import logging
 import os
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Dict, List, MutableMapping, Optional
 
 from agent.secret_sources.base import (
     SECRET_SOURCE_API_VERSION,
@@ -40,6 +40,8 @@ from agent.secret_sources.base import (
     FetchResult,
     SecretSource,
     is_valid_env_name,
+    reset_source_environment,
+    set_source_environment,
 )
 
 logger = logging.getLogger(__name__)
@@ -196,7 +198,8 @@ def _reset_registry_for_tests() -> None:
 
 
 def _fetch_with_timeout(
-    source: SecretSource, cfg: dict, home_path: Path
+    source: SecretSource, cfg: dict, home_path: Path,
+    environ: MutableMapping[str, str],
 ) -> FetchResult:
     """Run source.fetch() under a wall-clock budget; never raises.
 
@@ -211,7 +214,14 @@ def _fetch_with_timeout(
         max_workers=1, thread_name_prefix=f"secret-src-{source.name}"
     )
     try:
-        future = executor.submit(source.fetch, cfg, home_path)
+        def _fetch() -> FetchResult:
+            token = set_source_environment(environ)
+            try:
+                return source.fetch(cfg, home_path)
+            finally:
+                reset_source_environment(token)
+
+        future = executor.submit(_fetch)
         try:
             result = future.result(timeout=timeout)
         except concurrent.futures.TimeoutError:
@@ -321,7 +331,7 @@ def _profile_alias_target(var: str, profile: str) -> Optional[str]:
 
 
 def apply_all(secrets_cfg: dict, home_path: Path,
-              environ: Optional[Dict[str, str]] = None) -> ApplyReport:
+              environ: Optional[MutableMapping[str, str]] = None) -> ApplyReport:
     """Fetch from every enabled source and apply the merged result to env.
 
     ``environ`` defaults to ``os.environ``; injectable for tests.
@@ -376,7 +386,7 @@ def apply_all(secrets_cfg: dict, home_path: Path,
     for source in ordered:
         cfg = secrets_cfg.get(source.name)
         cfg = cfg if isinstance(cfg, dict) else {}
-        result = _fetch_with_timeout(source, cfg, home_path)
+        result = _fetch_with_timeout(source, cfg, home_path, env)
         fetches.append((source, cfg, result))
         try:
             for var in source.protected_env_vars(cfg):

gateway/run.py

@@ -1822,8 +1822,10 @@ def _profile_runtime_scope(profile_home: "Path"):
         set_secret_scope,
         reset_secret_scope,
     )
+    from hermes_cli.env_loader import hydrate_profile_secret_sources
 
     home_token = set_hermes_home_override(str(profile_home))
+    hydrate_profile_secret_sources(Path(profile_home))
     secret_token = set_secret_scope(build_profile_secret_scope(Path(profile_home)))
     try:
         yield

hermes_cli/env_loader.py

@@ -6,6 +6,7 @@ import codecs
 import io
 import os
 import sys
+import threading
 from pathlib import Path
 
 from dotenv import load_dotenv
@@ -48,6 +49,7 @@ _SECRET_SOURCE_VALUES_BY_HOME: dict[str, dict[str, str]] = {}
 # in-process cache prevents redundant network calls, but the print, the
 # config re-parse, and the ASCII sanitization sweep still ran every time.
 _APPLIED_HOMES: set[str] = set()
+_SECRET_SOURCE_CACHE_LOCK = threading.RLock()
 
 
 def _known_hermes_env_keys() -> set[str]:
@@ -164,6 +166,69 @@ def get_secret_source_values(
     return dict(_SECRET_SOURCE_VALUES_BY_HOME.get(home_key, {}))
 
 
+def hydrate_profile_secret_sources(
+    hermes_home: str | os.PathLike,
+) -> dict[str, str]:
+    """Resolve one profile's configured sources without mutating ``os.environ``.
+
+    Multiplex gateways can route a first turn to a secondary profile that has
+    never run the process-global dotenv startup path.  Resolve that profile's
+    sources against a private mapping seeded from its own ``.env`` and record
+    the usual per-home snapshot for ``build_profile_secret_scope()``.
+
+    Fail-open and once-per-home semantics intentionally mirror
+    ``_apply_external_secret_sources``.  The returned mapping contains only
+    values actually contributed by external sources, never the profile's
+    plaintext ``.env`` entries.
+    """
+    with _SECRET_SOURCE_CACHE_LOCK:
+        return _hydrate_profile_secret_sources(Path(hermes_home))
+
+
+def _hydrate_profile_secret_sources(home: Path) -> dict[str, str]:
+    """Locked implementation for :func:`hydrate_profile_secret_sources`."""
+    home_key = str(home.resolve())
+    if home_key in _APPLIED_HOMES:
+        return get_secret_source_values(home)
+
+    try:
+        cfg = _load_secrets_config(home)
+    except Exception:  # noqa: BLE001 — external sources must not block routing
+        return {}
+    if not cfg:
+        return {}
+
+    try:
+        from agent.secret_scope import _is_global_env, load_env_file
+        from agent.secret_sources.registry import apply_all
+
+        local_env = {
+            name: value
+            for name, value in os.environ.items()
+            if _is_global_env(name)
+        }
+        local_env.update(load_env_file(home / ".env"))
+        local_env["HERMES_HOME"] = str(home)
+        report = apply_all(cfg, home, environ=local_env)
+    except Exception:  # noqa: BLE001 — preserve fail-open startup behavior
+        return {}
+
+    if not report.sources:
+        return {}
+
+    _APPLIED_HOMES.add(home_key)
+    values: dict[str, str] = {}
+    for name, applied in report.provenance.items():
+        value = local_env.get(name)
+        if value is None:
+            continue
+        _SECRET_SOURCES[name] = applied.source
+        values[name] = value
+    if values:
+        _SECRET_SOURCE_VALUES_BY_HOME[home_key] = values
+    return dict(values)
+
+
 def reset_secret_source_cache() -> None:
     """Forget which HERMES_HOME paths have already had external secrets applied.
 

tests/gateway/test_multiplex_credential_isolation.py

@@ -88,3 +88,81 @@ class TestProfilePathResolutionUnderMultiplexScope:
         assert b_seen == prof_b / "skills"
 
 
+def test_cold_profile_hydrates_external_source_without_global_env(
+    tmp_path, monkeypatch
+):
+    """The first routed secondary turn must resolve its own source locally."""
+    import os
+
+    from agent.secret_sources.base import FetchResult
+    from agent.secret_sources.registry import AppliedVar, ApplyReport, SourceReport
+    from agent.secret_sources import registry
+    from agent.secret_scope import get_secret
+    from hermes_cli import env_loader
+    from gateway.run import _profile_runtime_scope
+
+    profile = tmp_path / "profiles" / "secondary"
+    sibling = tmp_path / "profiles" / "sibling"
+    profile.mkdir(parents=True)
+    sibling.mkdir(parents=True)
+    (profile / ".env").write_text(
+        "EXPLICIT_API_KEY=dotenv-wins\n", encoding="utf-8"
+    )
+    monkeypatch.delenv("TEST_PROVIDER_API_KEY", raising=False)
+    monkeypatch.delenv("EXPLICIT_API_KEY", raising=False)
+    monkeypatch.setattr(
+        env_loader,
+        "_load_secrets_config",
+        lambda home: (
+            {"fake-source": {"enabled": True}}
+            if Path(home).resolve() == profile.resolve()
+            else {}
+        ),
+    )
+
+    calls = {"count": 0}
+
+    def _fake_apply_all(_cfg, _home, *, environ=None):
+        calls["count"] += 1
+        assert environ is not os.environ
+        assert environ is not None
+        assert environ["EXPLICIT_API_KEY"] == "dotenv-wins"
+        environ["TEST_PROVIDER_API_KEY"] = "profile-only"
+        return ApplyReport(
+            sources=[
+                SourceReport(
+                    name="fake-source",
+                    label="Fake Source",
+                    result=FetchResult(),
+                    applied=["TEST_PROVIDER_API_KEY"],
+                )
+            ],
+            provenance={
+                "TEST_PROVIDER_API_KEY": AppliedVar(
+                    name="TEST_PROVIDER_API_KEY",
+                    source="fake-source",
+                    shape="mapped",
+                    overrode_env=False,
+                )
+            },
+        )
+
+    monkeypatch.setattr(registry, "apply_all", _fake_apply_all)
+    env_loader.reset_secret_source_cache()
+
+    with _profile_runtime_scope(profile):
+        assert get_secret("TEST_PROVIDER_API_KEY") == "profile-only"
+        assert get_secret("EXPLICIT_API_KEY") == "dotenv-wins"
+        assert env_loader.get_secret_source_values(profile) == {
+            "TEST_PROVIDER_API_KEY": "profile-only"
+        }
+    with _profile_runtime_scope(profile):
+        assert get_secret("TEST_PROVIDER_API_KEY") == "profile-only"
+    with _profile_runtime_scope(sibling):
+        assert get_secret("TEST_PROVIDER_API_KEY") is None
+
+    assert calls["count"] == 1
+    assert "TEST_PROVIDER_API_KEY" not in os.environ
+    assert "EXPLICIT_API_KEY" not in os.environ
+
+

tests/secret_sources/test_profile_secrets.py

@@ -123,5 +123,55 @@ def test_hyphenated_profile_name_matches_underscore_suffix():
     assert env["SLACK_APP_TOKEN"] == "xapp-1"
 
 
+def test_source_fetch_reads_injected_environment_without_global_mutation(
+    monkeypatch, tmp_path
+):
+    """Cold-profile bootstrap values reach sources through the local mapping."""
+    from agent.secret_sources.base import get_source_environment
+
+    class _BootstrapSource(SecretSource):
+        name = "bootstrap"
+        shape = "mapped"
+
+        def fetch(self, cfg, home_path):
+            result = FetchResult()
+            result.secrets = {
+                "RESOLVED_API_KEY": get_source_environment()["BOOTSTRAP_TOKEN"]
+            }
+            return result
+
+    registry.register_source(_BootstrapSource())
+    monkeypatch.delenv("BOOTSTRAP_TOKEN", raising=False)
+    env = {"BOOTSTRAP_TOKEN": "profile-token"}
+    _, applied = _apply(
+        {},
+        cfg_extra={"bootstrap": {"enabled": True}},
+        home=tmp_path,
+        env=env,
+    )
+
+    assert applied["RESOLVED_API_KEY"] == "profile-token"
+    assert "BOOTSTRAP_TOKEN" not in __import__("os").environ
+
+
+def test_empty_injected_environment_does_not_fall_back_to_process(monkeypatch, tmp_path):
+    from agent.secret_sources.base import get_source_environment
+
+    class _CanarySource(SecretSource):
+        name = "canary"
+        shape = "mapped"
+
+        def fetch(self, cfg, home_path):
+            result = FetchResult()
+            assert get_source_environment().get("LEAK_CANARY") is None
+            return result
+
+    registry.register_source(_CanarySource())
+    monkeypatch.setenv("LEAK_CANARY", "global-secret")
+    registry.apply_all(
+        {"canary": {"enabled": True}}, tmp_path, environ={}
+    )
+
+
 
 

tests/test_command_secret_source.py

@@ -35,12 +35,17 @@ if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
 from agent.secret_sources.command import (  # noqa: E402
+    _run_helper,
     apply_command_secrets,
     get_command_secret,
     list_command_secrets,
     parse_secret_output,
     unquote_dotenv_value,
 )
+from agent.secret_sources.base import (  # noqa: E402
+    reset_source_environment,
+    set_source_environment,
+)
 from hermes_cli import env_loader  # noqa: E402
 
 
@@ -57,6 +62,22 @@ def _write_helper(tmp_path: Path, body: str, name: str = "helper.sh") -> Path:
     return script
 
 
+def test_profile_helper_does_not_inherit_process_secret(monkeypatch):
+    monkeypatch.setenv("LEAK_CANARY", "global-secret")
+    token = set_source_environment({"PROFILE_ONLY": "profile-value"})
+    try:
+        output = _run_helper(
+            'printf "%s|%s" "${LEAK_CANARY-unset}" "$PROFILE_ONLY"',
+            "",
+            1.0,
+            1024,
+        )
+    finally:
+        reset_source_environment(token)
+
+    assert output == "unset|profile-value"
+
+
 @pytest.fixture(autouse=True)
 def _clean_env(monkeypatch):
     """Each test starts with a clean source map, applied-home guard, and no

tests/test_env_loader_secret_sources.py

@@ -135,6 +135,45 @@ def test_apply_external_secret_sources_records_bitwarden_origin(tmp_path, monkey
     )
 
 
+def test_cold_profile_bitwarden_uses_profile_bootstrap_without_global_env(
+    tmp_path, monkeypatch
+):
+    """Real Bitwarden adapter reads its token from the profile-local view."""
+    monkeypatch.delenv("BWS_ACCESS_TOKEN", raising=False)
+    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+    (tmp_path / ".env").write_text(
+        "BWS_ACCESS_TOKEN=profile-bootstrap\n", encoding="utf-8"
+    )
+    (tmp_path / "config.yaml").write_text(
+        "secrets:\n"
+        "  bitwarden:\n"
+        "    enabled: true\n"
+        "    project_id: test-project\n"
+        "    access_token_env: BWS_ACCESS_TOKEN\n",
+        encoding="utf-8",
+    )
+
+    import agent.secret_sources.bitwarden as bw_module
+    from agent.secret_sources import registry as reg_module
+
+    captured = {}
+    monkeypatch.setattr(bw_module, "find_bws", lambda **_kw: Path("/fake/bws"))
+
+    def _fake_fetch(**kwargs):
+        captured.update(kwargs)
+        return {"ANTHROPIC_API_KEY": "profile-provider-key"}, []
+
+    monkeypatch.setattr(bw_module, "fetch_bitwarden_secrets", _fake_fetch)
+    reg_module._reset_registry_for_tests()
+
+    assert env_loader.hydrate_profile_secret_sources(tmp_path) == {
+        "ANTHROPIC_API_KEY": "profile-provider-key"
+    }
+    assert captured["access_token"] == "profile-bootstrap"
+    assert os.environ.get("BWS_ACCESS_TOKEN") is None
+    assert os.environ.get("ANTHROPIC_API_KEY") is None
+
+
 def test_apply_external_secret_sources_noop_when_disabled(tmp_path, monkeypatch):
     """Disabled Bitwarden config must not touch the source map."""