fix(terminal): stop leaking provider env to local subprocesses

tests/tools/test_terminal_env_sanitization.py

@@ -0,0 +1,150 @@
+"""Regression tests for terminal subprocess environment scoping."""
+
+import os
+import shlex
+import sys
+import tempfile
+import time
+import types
+from pathlib import Path
+
+_IMPORT_HERMES_HOME = Path(tempfile.mkdtemp(prefix="hermes_test_import_"))
+for _subdir in ("sessions", "cron", "memories", "skills"):
+    (_IMPORT_HERMES_HOME / _subdir).mkdir()
+os.environ["HERMES_HOME"] = str(_IMPORT_HERMES_HOME)
+
+for _key in (
+    "OPENAI_API_KEY",
+    "OPENAI_BASE_URL",
+    "OPENROUTER_API_KEY",
+    "OPENROUTER_BASE_URL",
+    "ANTHROPIC_API_KEY",
+    "ALL_PROXY",
+    "all_proxy",
+    "HTTP_PROXY",
+    "http_proxy",
+    "HTTPS_PROXY",
+    "https_proxy",
+):
+    os.environ.pop(_key, None)
+
+from tools.environments.local import LocalEnvironment
+from tools.process_registry import ProcessRegistry
+
+
+_BLOCKED_VAR = "OPENAI_BASE_URL"
+_PARENT_VALUE = "http://parent.invalid/v1"
+_EXPLICIT_VALUE = "http://explicit.invalid/v1"
+
+
+def _print_env_command(var_name: str) -> str:
+    code = f'import os; print(os.getenv("{var_name}", ""))'
+    return f"{shlex.quote(sys.executable)} -c {shlex.quote(code)}"
+
+
+def _wait_for_exit(session, timeout: float = 5.0):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if session.exited:
+            return
+        time.sleep(0.05)
+    raise AssertionError(f"Timed out waiting for session {session.id} to exit")
+
+
+class TestLocalEnvironmentEnvScoping:
+    def test_execute_scrubs_provider_env_by_default(self, monkeypatch, tmp_path):
+        monkeypatch.setenv(_BLOCKED_VAR, _PARENT_VALUE)
+
+        env = LocalEnvironment(cwd=str(tmp_path), timeout=5)
+        result = env.execute(_print_env_command(_BLOCKED_VAR))
+
+        assert result["returncode"] == 0
+        assert result["output"].strip() == ""
+
+    def test_execute_allows_explicit_provider_env_override(self, monkeypatch, tmp_path):
+        monkeypatch.setenv(_BLOCKED_VAR, _PARENT_VALUE)
+
+        env = LocalEnvironment(
+            cwd=str(tmp_path),
+            timeout=5,
+            env={_BLOCKED_VAR: _EXPLICIT_VALUE},
+        )
+        result = env.execute(_print_env_command(_BLOCKED_VAR))
+
+        assert result["returncode"] == 0
+        assert result["output"].strip() == _EXPLICIT_VALUE
+
+
+class TestProcessRegistryEnvScoping:
+    def test_spawn_local_scrubs_provider_env_by_default(self, monkeypatch, tmp_path):
+        monkeypatch.setenv(_BLOCKED_VAR, _PARENT_VALUE)
+
+        registry = ProcessRegistry()
+        session = registry.spawn_local(
+            command=_print_env_command(_BLOCKED_VAR),
+            cwd=str(tmp_path),
+        )
+
+        _wait_for_exit(session)
+        assert session.exit_code == 0
+        assert session.output_buffer.strip() == ""
+
+    def test_spawn_local_allows_explicit_provider_env_override(self, monkeypatch, tmp_path):
+        monkeypatch.setenv(_BLOCKED_VAR, _PARENT_VALUE)
+
+        registry = ProcessRegistry()
+        session = registry.spawn_local(
+            command=_print_env_command(_BLOCKED_VAR),
+            cwd=str(tmp_path),
+            env_vars={_BLOCKED_VAR: _EXPLICIT_VALUE},
+        )
+
+        _wait_for_exit(session)
+        assert session.exit_code == 0
+        assert session.output_buffer.strip() == _EXPLICIT_VALUE
+
+    def test_spawn_local_pty_scrubs_provider_env_by_default(self, monkeypatch, tmp_path):
+        monkeypatch.setenv(_BLOCKED_VAR, _PARENT_VALUE)
+
+        captured = {}
+
+        class _FakePty:
+            pid = 123
+            exitstatus = 0
+
+            def isalive(self):
+                return False
+
+            def read(self, _size):
+                return b""
+
+            def wait(self):
+                return 0
+
+        class _FakePtyProcess:
+            @staticmethod
+            def spawn(argv, cwd, env, dimensions):
+                captured["argv"] = list(argv)
+                captured["cwd"] = cwd
+                captured["env"] = dict(env)
+                captured["dimensions"] = dimensions
+                return _FakePty()
+
+        monkeypatch.setitem(
+            sys.modules,
+            "ptyprocess",
+            types.SimpleNamespace(PtyProcess=_FakePtyProcess),
+        )
+
+        registry = ProcessRegistry()
+        session = registry.spawn_local(
+            command="true",
+            cwd=str(tmp_path),
+            use_pty=True,
+        )
+
+        _wait_for_exit(session)
+        assert session.exit_code == 0
+        assert captured["cwd"] == str(tmp_path)
+        assert captured["env"]["PYTHONUNBUFFERED"] == "1"
+        assert _BLOCKED_VAR not in captured["env"]

tools/environments/base.py

@@ -6,6 +6,27 @@
 from pathlib import Path
 
 
+_TERMINAL_BLOCKED_PROVIDER_ENV_KEYS = frozenset({
+    "ANTHROPIC_API_KEY",
+    "GLM_API_KEY",
+    "GLM_BASE_URL",
+    "KIMI_API_KEY",
+    "KIMI_BASE_URL",
+    "MINIMAX_API_KEY",
+    "MINIMAX_BASE_URL",
+    "MINIMAX_CN_API_KEY",
+    "MINIMAX_CN_BASE_URL",
+    "NOUS_API_KEY",
+    "NOUS_BASE_URL",
+    "OPENAI_API_KEY",
+    "OPENAI_BASE_URL",
+    "OPENROUTER_API_KEY",
+    "OPENROUTER_BASE_URL",
+    "ZAI_API_KEY",
+    "Z_AI_API_KEY",
+})
+
+
 def get_sandbox_dir() -> Path:
     """Return the host-side root for all sandbox storage (Docker workspaces,
     Singularity overlays/SIF cache, etc.).
@@ -21,6 +42,16 @@ def get_sandbox_dir() -> Path:
     return p
 
 
+def build_terminal_subprocess_env(extra_env: dict | None = None) -> dict:
+    """Build a subprocess env without leaking Hermes runtime provider state."""
+    env = dict(os.environ)
+    for key in _TERMINAL_BLOCKED_PROVIDER_ENV_KEYS:
+        env.pop(key, None)
+    if extra_env:
+        env.update(extra_env)
+    return env
+
+
 class BaseEnvironment(ABC):
     """Common interface for all Hermes execution backends.
 

tools/environments/local.py

@@ -10,7 +10,7 @@
 
 _IS_WINDOWS = platform.system() == "Windows"
 
-from tools.environments.base import BaseEnvironment
+from tools.environments.base import BaseEnvironment, build_terminal_subprocess_env
 
 # Unique marker to isolate real command output from shell init/exit noise.
 # printf (no trailing newline) keeps the boundaries clean for splitting.
@@ -192,7 +192,7 @@ def execute(self, command: str, cwd: str = "", *,
             # Ensure PATH always includes standard dirs — systemd services
             # and some terminal multiplexers inherit a minimal PATH.
             _SANE_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
-            run_env = dict(os.environ | self.env)
+            run_env = build_terminal_subprocess_env(self.env)
             existing_path = run_env.get("PATH", "")
             if "/usr/bin" not in existing_path.split(":"):
                 run_env["PATH"] = f"{existing_path}:{_SANE_PATH}" if existing_path else _SANE_PATH

tools/process_registry.py

@@ -42,6 +42,7 @@
 import uuid
 
 _IS_WINDOWS = platform.system() == "Windows"
+from tools.environments.base import build_terminal_subprocess_env
 from tools.environments.local import _find_shell
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -153,7 +154,7 @@ def spawn_local(
                 else:
                     from ptyprocess import PtyProcess as _PtyProcessCls
                 user_shell = _find_shell()
-                pty_env = os.environ | (env_vars or {})
+                pty_env = build_terminal_subprocess_env(env_vars)
                 pty_env["PYTHONUNBUFFERED"] = "1"
                 pty_proc = _PtyProcessCls.spawn(
                     [user_shell, "-lic", command],
@@ -194,7 +195,7 @@ def spawn_local(
         # Force unbuffered output for Python scripts so progress is visible
         # during background execution (libraries like tqdm/datasets buffer when
         # stdout is a pipe, hiding output from process(action="poll")).
-        bg_env = os.environ | (env_vars or {})
+        bg_env = build_terminal_subprocess_env(env_vars)
         bg_env["PYTHONUNBUFFERED"] = "1"
         proc = subprocess.Popen(
             [user_shell, "-lic", command],