fix(terminal): isolate ambient bridge and bind launch scope under multiplexing (#107422)

tests/tools/test_terminal_env_bridge.py

@@ -152,3 +152,57 @@ def test_bridge_config_failure_does_not_crash(monkeypatch):
 
     assert config["env_type"] == "ssh"
     assert config["ssh_host"] == "example.test"
+
+
+def test_secondary_home_override_does_not_latch_ambient_env(tmp_path, monkeypatch):
+    """#107422: first bridge under a secondary profile must not poison os.environ.
+
+    Multiplexed dashboard sets ``set_hermes_home_override`` for profile B. If
+    ``_ensure_terminal_env_bridged`` ran there (no terminal scope yet), the
+    one-shot latch used to write B's docker policy into process-global env and
+    every later unscoped launch-profile tool call inherited it.
+    """
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+    launch_home = tmp_path / "launch"
+    secondary_home = tmp_path / "profiles" / "docker-bee"
+    launch_home.mkdir(parents=True)
+    secondary_home.mkdir(parents=True)
+    (launch_home / "config.yaml").write_text(
+        "terminal:\n  backend: local\n", encoding="utf-8"
+    )
+    (secondary_home / "config.yaml").write_text(
+        "terminal:\n"
+        "  backend: docker\n"
+        "  docker_image: bee/local:1\n"
+        '  docker_volumes:\n'
+        '    - /bee/vol:/data\n',
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_HOME", str(launch_home))
+    # Clean ambient — the dashboard process starts without TERMINAL_ENV.
+    for name in (
+        "TERMINAL_ENV",
+        "TERMINAL_DOCKER_IMAGE",
+        "TERMINAL_DOCKER_VOLUMES",
+    ):
+        monkeypatch.delenv(name, raising=False)
+
+    token = set_hermes_home_override(str(secondary_home))
+    try:
+        # Unscoped call under secondary home (the residual path).
+        terminal_tool._ensure_terminal_env_bridged()
+    finally:
+        reset_hermes_home_override(token)
+
+    assert "TERMINAL_ENV" not in os.environ
+    assert "TERMINAL_DOCKER_IMAGE" not in os.environ
+    assert "TERMINAL_DOCKER_VOLUMES" not in os.environ
+    # Bridge must still be available for the real launch profile afterwards.
+    assert terminal_tool._terminal_config_bridge_attempted is False
+
+    config = terminal_tool._get_env_config()
+    assert config["env_type"] == "local"
+    assert os.environ["TERMINAL_ENV"] == "local"
+    assert "bee/local:1" not in os.environ.get("TERMINAL_DOCKER_IMAGE", "")
+

tests/tools/test_terminal_scope_multiplex.py

@@ -240,3 +240,161 @@ def test_dotenv_json_strings_stay_json_strings(tmp_path):
     scope = build_profile_terminal_scope(home)
     assert json.loads(scope["TERMINAL_DOCKER_FORWARD_ENV"]) == ["EMAIL_HOME_ADDRESS"]
     assert json.loads(scope["TERMINAL_DOCKER_VOLUMES"]) == ["/tmp/a:/data"]
+
+
+def test_launch_terminal_scope_preserves_ambient_env(tmp_path, monkeypatch):
+    """#107422 / Finding 3: launch scope preserves ambient process environment
+    (e.g. TERMINAL_ENV=ssh, TERMINAL_SSH_HOST) when launch config.yaml has no terminal
+    section, while secondary profile scopes never inherit ambient env."""
+    from tools.terminal_scope import (
+        build_launch_terminal_scope,
+        build_profile_terminal_scope,
+    )
+
+    launch_home = tmp_path / "launch-profile"
+    secondary_home = tmp_path / "profiles" / "sec"
+    launch_home.mkdir(parents=True, exist_ok=True)
+    secondary_home.mkdir(parents=True, exist_ok=True)
+    (launch_home / "config.yaml").write_text("{}\n", encoding="utf-8")
+    (secondary_home / "config.yaml").write_text("{}\n", encoding="utf-8")
+
+    monkeypatch.setenv("HERMES_HOME", str(launch_home))
+    monkeypatch.setenv("TERMINAL_ENV", "ssh")
+    monkeypatch.setenv("TERMINAL_SSH_HOST", "example.test")
+
+    launch_scope = build_launch_terminal_scope(launch_home)
+    assert launch_scope["TERMINAL_ENV"] == "ssh"
+    assert launch_scope["TERMINAL_SSH_HOST"] == "example.test"
+
+    # Secondary profile must NOT inherit ambient TERMINAL_ENV=ssh
+    sec_scope = build_profile_terminal_scope(secondary_home)
+    assert sec_scope["TERMINAL_ENV"] == "local"
+    assert sec_scope.get("TERMINAL_SSH_HOST", "") == ""
+
+    # Explicit launch config backend overrides ambient env
+    (launch_home / "config.yaml").write_text(
+        "terminal:\n  backend: docker\n", encoding="utf-8"
+    )
+    launch_scope_docker = build_launch_terminal_scope(launch_home)
+    assert launch_scope_docker["TERMINAL_ENV"] == "docker"
+    assert launch_scope_docker["TERMINAL_SSH_HOST"] == "example.test"
+
+
+def test_prepare_turn_input_binds_launch_scope_once_multiplexing_is_active(
+    tmp_path, monkeypatch
+):
+    """#107422 / Finding 4: real prompt_turn._prepare_turn_input path binds the
+    launch profile's terminal scope once _served_profile_homes is non-empty."""
+    from tools.terminal_scope import get_terminal_scope, reset_terminal_scope, terminal_env
+    import tui_gateway.server as srv
+    from tui_gateway.prompt_turn import _TurnRun
+
+    launch_home = tmp_path / "launch-turn-home"
+    launch_home.mkdir(parents=True, exist_ok=True)
+    (launch_home / "config.yaml").write_text(
+        "terminal:\n  backend: local\n", encoding="utf-8"
+    )
+    monkeypatch.setenv("HERMES_HOME", str(launch_home))
+    monkeypatch.setattr(srv, "_hermes_home", str(launch_home), raising=False)
+
+    # Single profile mode: _served_profile_homes is empty
+    monkeypatch.setattr(srv, "_served_profile_homes", set(), raising=False)
+
+    import threading
+    session = {
+        "session_key": "test_launch_sess",
+        "profile_home": None,
+        "history": [],
+        "history_lock": threading.Lock(),
+        "cols": 80,
+        "agent": None,
+    }
+    st_single = _TurnRun(
+        agent=None, one_turn_restore=True, terminal_callback=None, receipt_committed=False
+    )
+    # Monkeypatch helper dependencies that _prepare_turn_input runs
+    monkeypatch.setattr(srv, "_wire_callbacks", lambda sid: None, raising=False)
+    monkeypatch.setattr(srv, "_sync_bot_capabilities", lambda sid, s: None, raising=False)
+    monkeypatch.setattr(srv, "_session_cwd", lambda s: str(launch_home), raising=False)
+    monkeypatch.setattr(srv, "_register_session_cwd", lambda s: None, raising=False)
+    monkeypatch.setattr(srv, "make_stream_renderer", lambda cols: None, raising=False)
+    monkeypatch.setattr(srv, "_start_turn_voice", lambda: (None, None), raising=False)
+    monkeypatch.setattr(srv, "_turn_notes", lambda sid, s: [], raising=False)
+
+    srv._prepare_turn_input("sid1", session, st_single, "hello", [])
+    assert st_single.scopes.terminal is None
+    assert get_terminal_scope() is None
+
+    # Now activate multiplexing by registering a secondary served profile
+    sec_home = tmp_path / "profiles" / "sec"
+    sec_home.mkdir(parents=True, exist_ok=True)
+    served = {sec_home}
+    monkeypatch.setattr(srv, "_served_profile_homes", served, raising=False)
+
+    st_multi = _TurnRun(
+        agent=None, one_turn_restore=True, terminal_callback=None, receipt_committed=False
+    )
+    srv._prepare_turn_input("sid1", session, st_multi, "hello", [])
+    try:
+        assert st_multi.scopes.terminal is not None
+        assert get_terminal_scope() is not None
+        assert terminal_env("TERMINAL_ENV") == "local"
+    finally:
+        reset_terminal_scope(st_multi.scopes.terminal)
+    assert get_terminal_scope() is None
+
+
+def test_profile_build_scope_binds_terminal_scope(tmp_path):
+    """#107422 / Finding 2: _profile_build_scope binds the terminal scope for the
+    profile so eager resume and branch builds resolve routed terminal policy."""
+    from tui_gateway.methods_session import _profile_build_scope
+    from tools.terminal_scope import get_terminal_scope, terminal_env
+
+    sec_home = tmp_path / "profiles" / "docker-resume"
+    sec_home.mkdir(parents=True, exist_ok=True)
+    (sec_home / "config.yaml").write_text(
+        "terminal:\n  backend: docker\n  docker_image: test/eager:1\n",
+        encoding="utf-8",
+    )
+
+    assert get_terminal_scope() is None
+    with _profile_build_scope(sec_home):
+        assert get_terminal_scope() is not None
+        assert terminal_env("TERMINAL_ENV") == "docker"
+        assert terminal_env("TERMINAL_DOCKER_IMAGE") == "test/eager:1"
+    assert get_terminal_scope() is None
+
+
+def test_spawn_side_agent_binds_terminal_scope_on_worker(tmp_path, monkeypatch):
+    """#107422 / Finding 1: _spawn_side_agent binds the secondary profile's terminal
+    scope for the full worker lifetime on daemon thread."""
+    import threading
+    import tui_gateway.server as srv
+    from tools.terminal_scope import get_terminal_scope, terminal_env
+
+    sec_home = tmp_path / "profiles" / "docker-worker"
+    sec_home.mkdir(parents=True, exist_ok=True)
+    (sec_home / "config.yaml").write_text(
+        "terminal:\n  backend: docker\n", encoding="utf-8"
+    )
+
+    observed = {}
+    event_done = threading.Event()
+
+    def body():
+        observed["terminal_env"] = terminal_env("TERMINAL_ENV")
+        observed["scope_present"] = get_terminal_scope() is not None
+        event_done.set()
+        return "ok"
+
+    monkeypatch.setattr(srv, "_emit", lambda ev, parent, payload: None, raising=False)
+    monkeypatch.setattr(srv, "_session_cwd", lambda s: str(tmp_path), raising=False)
+
+    session = {"profile_home": str(sec_home), "session_key": "s-side"}
+    srv._spawn_side_agent("rid", session, "task_1", "parent_1", "event_1", body)
+
+    assert event_done.wait(timeout=5.0), "side agent worker timed out"
+    assert observed["scope_present"] is True
+    assert observed["terminal_env"] == "docker"
+
+

tools/terminal_scope.py

@@ -86,12 +86,15 @@ def terminal_env(name: str, default: str = "") -> str:
     return default if value is None else str(value)
 
 
-def build_profile_terminal_scope(hermes_home: "Any") -> Dict[str, str]:
+def build_profile_terminal_scope(
+    hermes_home: "Any", *, ambient_env: Optional[Dict[str, str]] = None
+) -> Dict[str, str]:
     """Build the COMPLETE effective ``TERMINAL_*`` policy for a profile home.
 
-    Projection: ``DEFAULT_CONFIG['terminal']`` <- profile ``.env`` TERMINAL_* <- profile
-    ``config.yaml`` ``terminal:``. Total by construction, so a bound scope never widens back to
-    ambient authority. Raises :class:`TerminalPolicyUnavailable` if a present file is unreadable.
+    Projection: ``DEFAULT_CONFIG['terminal']`` <- optional ambient_env (for launch profile)
+    <- profile ``.env`` TERMINAL_* <- profile ``config.yaml`` ``terminal:``. Total by
+    construction, so a bound scope never widens back to ambient authority. Raises
+    :class:`TerminalPolicyUnavailable` if a present file is unreadable.
     """
     from hermes_cli.config import TERMINAL_CONFIG_ENV_MAP, _terminal_env_value
     from hermes_cli.config_defaults import DEFAULT_CONFIG
@@ -112,6 +115,8 @@ def _apply(mapping: Dict[str, Any]) -> None:
                 scope[env_var] = _terminal_env_value(value)
 
     _apply({**_TOOL_LEVEL_DEFAULTS, **(DEFAULT_CONFIG.get("terminal") or {})})
+    if ambient_env:
+        scope.update((k, str(v)) for k, v in ambient_env.items() if k.startswith("TERMINAL_"))
     env_path = home / ".env"
     if env_path.exists():
         # load_env_file swallows OSError by design (secret scope fails soft); an unreadable
@@ -145,6 +150,28 @@ def _apply(mapping: Dict[str, Any]) -> None:
     return scope
 
 
+def build_launch_terminal_scope(launch_home: "Any" = None) -> Dict[str, str]:
+    """Build the effective ``TERMINAL_*`` policy for the launch profile, preserving ambient env.
+
+    Under multiplexing, launch-profile turns bind an authoritative scope so they never fall back
+    to an unmanaged environment. Ambient process env (e.g. TERMINAL_ENV=ssh set at startup) is
+    preserved unless explicitly overridden by launch_home's config.yaml (#107422).
+    """
+    from hermes_constants import get_process_hermes_home
+
+    home = Path(launch_home) if launch_home is not None else get_process_hermes_home()
+    return build_profile_terminal_scope(home, ambient_env=dict(os.environ))
+
+
+def install_launch_terminal_scope(launch_home: "Any" = None) -> Token:
+    """Build and install launch profile's policy; on failure install the refusal scope. Never raises."""
+    try:
+        return set_terminal_scope(build_launch_terminal_scope(launch_home))
+    except TerminalPolicyUnavailable as exc:
+        logger.warning("launch terminal policy unavailable: %s", exc)
+        return _terminal_scope_var.set(TerminalPolicyRefusal(str(exc)))
+
+
 def install_profile_terminal_scope(hermes_home: "Any") -> Token:
     """Build AND install a profile's policy; on failure install the refusal scope. Never raises."""
     try:

tools/terminal_tool.py

@@ -546,6 +546,11 @@ def _ensure_terminal_env_bridged() -> None:
 
     if get_terminal_scope() is not None:
         return
+    # Never write a secondary profile's terminal.* into process-global env.
+    from hermes_constants import get_hermes_home_override
+
+    if get_hermes_home_override() is not None:
+        return
     global _terminal_config_bridge_attempted
     if _terminal_config_bridge_attempted:
         return

tui_gateway/methods_prompt.py

@@ -5,6 +5,8 @@
 """
 
 import contextlib
+from pathlib import Path
+import threading
 
 from .method_ctx import HandlerRegistry, bind_module
 
@@ -934,10 +936,33 @@ def run():
         # process for the task_id and tear down the very server the restart just started.
         profile_home = session.get("profile_home")
         home_token = set_hermes_home_override(profile_home) if profile_home else None
+        secret_token = None
+        terminal_token = None
+        if profile_home:
+            with contextlib.suppress(Exception):
+                from agent.secret_scope import build_profile_secret_scope, set_secret_scope
+                secret_token = set_secret_scope(build_profile_secret_scope(Path(profile_home)))
+            with contextlib.suppress(Exception):
+                from tools.terminal_scope import install_profile_terminal_scope
+                terminal_token = install_profile_terminal_scope(Path(profile_home))
+        else:
+            with contextlib.suppress(Exception):
+                from tui_gateway.server import _hermes_home, _served_profile_homes
+                if _served_profile_homes:
+                    from tools.terminal_scope import install_launch_terminal_scope
+                    terminal_token = install_launch_terminal_scope(Path(_hermes_home))
         try:
             try:
                 text = body()
             finally:
+                if terminal_token is not None:
+                    with contextlib.suppress(Exception):
+                        from tools.terminal_scope import reset_terminal_scope
+                        reset_terminal_scope(terminal_token)
+                if secret_token is not None:
+                    with contextlib.suppress(Exception):
+                        from agent.secret_scope import reset_secret_scope
+                        reset_secret_scope(secret_token)
                 if home_token is not None:
                     reset_hermes_home_override(home_token)
             _emit(event, parent, {"task_id": task_id, **extra, "text": text})

tui_gateway/methods_session.py

@@ -5,7 +5,11 @@
 server.py the same way (tests monkeypatching ``server.X`` still intercept)."""
 
 import contextlib
+from pathlib import Path
 
+from agent.secret_scope import build_profile_secret_scope, reset_secret_scope, set_secret_scope
+from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+from tools.terminal_scope import install_profile_terminal_scope, reset_terminal_scope
 from .method_ctx import HandlerRegistry, bind_module
 
 _registry = HandlerRegistry()
@@ -69,17 +73,31 @@ def _new_runtime_ids(params: dict) -> tuple[str, str]:
 
 @contextlib.contextmanager
 def _profile_build_scope(profile_home):
-    """Bind HERMES_HOME + secret scope for an agent build (home alone leaves get_secret() on the LAUNCH .env)."""
+    """Bind HERMES_HOME + secret + terminal scope for an agent build.
+
+    Home alone leaves get_secret() and terminal policy on the launch profile. Eager-resume and branch builds
+    run inside this scope, so tool availability discovery and prompt probing resolve the routed profile (#107422).
+    """
     if not profile_home:
         yield
         return
     home_token = set_hermes_home_override(str(profile_home))
-    secret_token = set_secret_scope(build_profile_secret_scope(Path(str(profile_home))))
+    secret_token = None
+    with contextlib.suppress(Exception):
+        secret_token = set_secret_scope(build_profile_secret_scope(Path(str(profile_home))))
+    terminal_token = None
+    with contextlib.suppress(Exception):
+        terminal_token = install_profile_terminal_scope(Path(str(profile_home)))
     try:
         yield
     finally:
+        if terminal_token is not None:
+            with contextlib.suppress(Exception):
+                reset_terminal_scope(terminal_token)
+        if secret_token is not None:
+            with contextlib.suppress(Exception):
+                reset_secret_scope(secret_token)
         reset_hermes_home_override(home_token)
-        reset_secret_scope(secret_token)
 
 
 def _make_agent_in_context(sid: str, key: str, **kwargs):

tui_gateway/prompt_turn.py

@@ -450,6 +450,11 @@ def _prepare_turn_input(sid: str, session: dict, st: _TurnRun, text: Any, images
         scopes.secret = set_secret_scope(build_profile_secret_scope(Path(profile_home)))
         from tools.terminal_scope import install_profile_terminal_scope
         scopes.terminal = install_profile_terminal_scope(Path(profile_home))
+    else:
+        from tui_gateway.server import _hermes_home, _served_profile_homes
+        if _served_profile_homes:
+            from tools.terminal_scope import install_launch_terminal_scope
+            scopes.terminal = install_launch_terminal_scope(Path(_hermes_home))
     # The sudo password callback is thread-local: without re-wiring here, sudo prompts
     # fall through to /dev/tty and hang the headless gateway (re-run is a no-op).
     _wire_callbacks(sid)