ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

simplify(compat): code_execution_tool/environments.local — drop 36 re-exports, repoint 5 callers + 14 test files

Browse Source
This commit is contained in:
Teknium
2026-09-03 13:24:04 -07:00
parent 5c3db6750f
commit 98c140bc4b
75 changed files with 375 additions and 381 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/chat_completion_helpers.py
+1 -1
View File
@@ -39,7 +39,7 @@ from agent.message_metadata import append_message, stamp_message_timestamp
from agent.message_sanitization import (_sanitize_surrogates, _repair_tool_call_arguments)
from agent.reasoning_summaries import separate_glued_reasoning_blocks
from agent.stream_single_writer import claim_stream_writer, stream_writer_is_current
from tools.terminal_tool import is_persistent_env
from tools.terminal_tool_lifecycle import is_persistent_env
from utils import base_url_host_matches, base_url_hostname, env_float, env_int
logger = logging.getLogger(__name__)
agent/prompt_builder.py
+4 -3
View File
@@ -857,11 +857,12 @@ _BACKEND_PROBE_CMD = (
def _run_backend_probe(env_type: str, terminal_tool) -> str:
"""Execute the probe command inside a freshly built backend; "" when it yields nothing."""
from tools.terminal_tool_backends import _ssh_config_from_config
from tools.terminal_tool_backends import _create_environment, _ssh_config_from_config
from tools.terminal_tool_lifecycle import _cleanup_env
config = terminal_tool._get_env_config()
# Mirrors tools/terminal_tool.py's live-command assembly (`_create_environment` is the factory).
env = terminal_tool._create_environment(
env = _create_environment(
env_type=env_type, image=config.get(_BACKEND_IMAGE_KEYS[env_type], "") if env_type in _BACKEND_IMAGE_KEYS else "", cwd=config.get("cwd", ""),
timeout=config.get("timeout", 180),
ssh_config=_ssh_config_from_config(config) if env_type == "ssh" else None,
@@ -878,7 +879,7 @@ def _run_backend_probe(env_type: str, terminal_tool) -> str:
# (keyed by user@host:port) shared with the agent's real environment; ControlPersist expires it.
if env_type != "ssh":
try:
terminal_tool._cleanup_env(env, force_remove=True)
_cleanup_env(env, force_remove=True)
except Exception:
logger.debug("Backend probe cleanup failed", exc_info=True)
if result.get("returncode") != 0:
agent/terminal_env_provider.py
+1 -1
View File
@@ -2,7 +2,7 @@
Pluggable-backend interface for terminal execution environments (cloud sandboxes, remote
runners). Providers register via :meth:`PluginContext.register_terminal_environment_provider`;
:func:`tools.terminal_tool._create_environment` consults the registry for any ``TERMINAL_ENV``
:func:`tools.terminal_tool_backends._create_environment` consults the registry for any ``TERMINAL_ENV``
/ ``terminal.backend`` value that is not a built-in (built-ins stay in ``tools/environments/``;
third-party sandbox vendors do NOT have to live in core). :meth:`create_environment` returns
any ``BaseEnvironment`` duck type (``execute()``, ``cleanup()`` …); the factory stamps
agent/terminal_env_registry.py
+1 -1
View File
@@ -2,7 +2,7 @@
Central map of registered pluggable terminal backends, populated by plugins via
:meth:`PluginContext.register_terminal_environment_provider` and consumed by
:func:`tools.terminal_tool._create_environment` plus the classification helpers across the
:func:`tools.terminal_tool_backends._create_environment` plus the classification helpers across the
terminal/file/approval/prompt surfaces. Unlike the image/video/web/browser registries there
is **no active-provider resolution**: the active backend is whatever ``TERMINAL_ENV`` /
``terminal.backend`` names. Built-in names are reserved (registration raises) so a plugin can
gateway/run_shutdown.py
+1 -1
View File
@@ -1397,7 +1397,7 @@ class GatewayShutdownMixin:
_marked_cron_jobs = _step("mark_running_jobs_interrupted", _mark_cron_interrupted) or []
_step("async interrupt_all", _interrupt_delegations)
def _cleanup_environments() -> None:
from tools.terminal_tool import cleanup_all_environments
from tools.terminal_tool_lifecycle import cleanup_all_environments
cleanup_all_environments()
def _cleanup_browsers() -> None:
hermes_cli/doctor_tools.py
+1 -1
View File
@@ -180,7 +180,7 @@ def _check_daytona_backend(issues: list[str]) -> None:
def _check_vercel_backend(issues: list[str]) -> None:
from tools.terminal_tool import _SUPPORTED_VERCEL_RUNTIMES
from tools.terminal_tool_backends import _SUPPORTED_VERCEL_RUNTIMES
runtime = os.getenv("TERMINAL_VERCEL_RUNTIME", "node24").strip() or "node24"
supported = ", ".join(_SUPPORTED_VERCEL_RUNTIMES)
_require(runtime in _SUPPORTED_VERCEL_RUNTIMES, ("Vercel runtime", f"({runtime})"),
hermes_cli/setup_terminal.py
+3 -3
View File
@@ -20,7 +20,7 @@ def _prompt_vercel_sandbox_settings(config: dict):
terminal = config.setdefault("terminal", {})
_setup._info(None, "Vercel Sandbox settings:", " Filesystem persistence uses Vercel snapshots.",
" Snapshots restore files only; live processes do not continue after sandbox recreation.")
from tools.terminal_tool import _SUPPORTED_VERCEL_RUNTIMES
from tools.terminal_tool_backends import _SUPPORTED_VERCEL_RUNTIMES
current_runtime = terminal.get("vercel_runtime") or "node24"
supported_label = ", ".join(_SUPPORTED_VERCEL_RUNTIMES)
runtime = _setup.prompt(f" Runtime ({supported_label})", current_runtime).strip() or current_runtime
@@ -176,8 +176,8 @@ def _setup_backend_modal(config: dict) -> None:
from tools.managed_tool_gateway import is_managed_tool_gateway_ready
from tools.tool_backend_helpers import normalize_modal_mode
managed_modal_available = bool(
_setup.managed_nous_tools_enabled()
and _setup.get_nous_subscription_features(config).nous_auth_present
tool_backend_helpers.managed_nous_tools_enabled()
and nous_subscription.get_nous_subscription_features(config).nous_auth_present
and is_managed_tool_gateway_ready("modal"))
modal_mode = normalize_modal_mode(_setup.cfg_get(config, "terminal", "modal_mode"))
use_managed_modal = False
tests/acp/test_approval_isolation.py
+1 -1
View File
@@ -120,7 +120,7 @@ class TestThreadLocalApprovalCallback:
def test_sudo_password_cache_does_not_leak_across_threads(self):
"""Interactive sudo cache must not bleed into another executor thread."""
from tools.terminal_tool import (
from tools.terminal_tool_sudo import (
_get_cached_sudo_password,
_reset_cached_sudo_passwords,
_set_cached_sudo_password,
tests/agent/test_prompt_builder.py
+6 -6
View File
@@ -816,9 +816,9 @@ class TestEnvironmentHints:
created["env_type"] = env_type
return _FakeEnv()
# Patch the REAL factory in tools.terminal_tool — the probe imports it
# Patch the REAL factory in tools.terminal_tool_backends — the probe imports it
# locally, so the import itself must succeed (the bug was here).
import tools.terminal_tool as _tt
import tools.terminal_tool_backends as _tt
monkeypatch.setattr(_tt, "_create_environment", _fake_create_environment)
line = _pb._probe_remote_backend("docker")
@@ -858,7 +858,7 @@ class TestEnvironmentHints:
def cleanup(self, *, force_remove=False):
cleaned["force_remove"] = force_remove
import tools.terminal_tool as _tt
import tools.terminal_tool_backends as _tt
monkeypatch.setattr(_tt, "_create_environment", lambda **kw: _FakeEnv())
assert _pb._probe_remote_backend("docker") is not None
@@ -882,7 +882,7 @@ class TestEnvironmentHints:
def cleanup(self, *, force_remove=False):
cleaned.append(force_remove)
import tools.terminal_tool as _tt
import tools.terminal_tool_backends as _tt
monkeypatch.setattr(_tt, "_create_environment", lambda **kw: _ExplodingEnv())
assert _pb._probe_remote_backend("docker") is None
@@ -911,7 +911,7 @@ class TestEnvironmentHints:
def cleanup(self):
calls.append("bare")
import tools.terminal_tool as _tt
import tools.terminal_tool_backends as _tt
monkeypatch.setattr(_tt, "_create_environment", lambda **kw: _LegacyEnv())
assert _pb._probe_remote_backend("singularity") is not None
@@ -941,7 +941,7 @@ class TestEnvironmentHints:
def cleanup(self):
calls.append("cleanup")
import tools.terminal_tool as _tt
import tools.terminal_tool_backends as _tt
monkeypatch.setattr(_tt, "_create_environment", lambda **kw: _SharedSshEnv())
assert _pb._probe_remote_backend("ssh") is not None
tests/cron/test_cron_script.py
+39 -39
View File
@@ -88,7 +88,7 @@ class TestRunJobScript:
"""Test the _run_job_script() function."""
def test_successful_script(self, cron_env):
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
script = cron_env / "scripts" / "test.py"
script.write_text('print("hello from script")\n')
@@ -98,7 +98,7 @@ class TestRunJobScript:
assert output == "hello from script"
def test_script_relative_path(self, cron_env):
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
script = cron_env / "scripts" / "relative.py"
script.write_text('print("relative works")\n')
@@ -110,8 +110,8 @@ class TestRunJobScript:
def test_script_subprocess_env_sanitized(self, cron_env, monkeypatch):
"""Cron scripts must not inherit Hermes provider env (SECURITY.md §2.3)."""
from tools.environments.local import _HERMES_PROVIDER_ENV_BLOCKLIST
from cron.scheduler import _run_job_script
from tools.environments.local_env_policy import _HERMES_PROVIDER_ENV_BLOCKLIST
from cron.scheduler_script import _run_job_script
# sorted() so the probed var is deterministic across runs
# (frozenset iteration order varies with PYTHONHASHSEED).
@@ -139,7 +139,8 @@ class TestRunJobScript:
# ``Scripts/python.exe`` launcher layout or the CREATE_NO_WINDOW
# creationflags this branch exists for.
from cron import scheduler as sched_mod
from cron.scheduler import _run_job_script
from cron import scheduler_script as sched_script
from cron.scheduler_script import _run_job_script
script = cron_env / "scripts" / "probe.py"
script.write_text('print("ok")\n')
@@ -177,7 +178,7 @@ class TestRunJobScript:
fake_run = FakeProc
monkeypatch.setattr(sched_mod.sys, "executable", str(venv_python))
monkeypatch.setattr(sched_mod, "windows_hide_flags", lambda: 0x08000000)
monkeypatch.setattr(sched_script, "windows_hide_flags", lambda: 0x08000000)
monkeypatch.setattr(sched_mod.subprocess, "Popen", fake_run)
success, output = _run_job_script("probe.py")
@@ -196,7 +197,7 @@ class TestRunJobScript:
# The script runner always adds CREATE_NEW_PROCESS_GROUP on win32 so a
# cancel can taskkill the whole tree; on POSIX the getattr default is
# 0 and the flag set is exactly windows_hide_flags().
expected_flags = sched_mod.windows_hide_flags() | getattr(
expected_flags = sched_script.windows_hide_flags() | getattr(
sched_mod.subprocess, "CREATE_NEW_PROCESS_GROUP", 0
)
assert captured["kwargs"]["creationflags"] == expected_flags
@@ -210,7 +211,7 @@ class TestRunJobScript:
installs would raise ModuleNotFoundError in cron scripts)."""
import subprocess
from cron.scheduler import _windows_cron_bootstrap_argv
from cron.scheduler_script import _windows_cron_bootstrap_argv
venv = tmp_path / "venv"
site_packages = venv / "Lib" / "site-packages"
@@ -241,7 +242,7 @@ class TestRunJobScript:
(runpy.run_path alone does not add it)."""
import subprocess
from cron.scheduler import _windows_cron_bootstrap_argv
from cron.scheduler_script import _windows_cron_bootstrap_argv
venv = tmp_path / "venv"
site_packages = venv / "Lib" / "site-packages"
@@ -266,7 +267,7 @@ class TestRunJobScript:
def test_bootstrap_argv_falls_back_without_site_packages(self, cron_env, tmp_path):
"""Unresolvable venv layout must not break the run — fall back to a
plain invocation (pre-existing PYTHONPATH behaviour)."""
from cron.scheduler import _windows_cron_bootstrap_argv
from cron.scheduler_script import _windows_cron_bootstrap_argv
script = cron_env / "scripts" / "probe.py"
script.write_text('print("ok")\n', encoding="utf-8")
@@ -284,7 +285,8 @@ class TestRunJobScript:
def test_non_windows_script_preserves_default_text_decoding(self, cron_env, monkeypatch):
# No platform patching: the Linux CI host already takes this branch.
from cron import scheduler as sched_mod
from cron.scheduler import _run_job_script
from cron import scheduler_script as sched_script
from cron.scheduler_script import _run_job_script
script = cron_env / "scripts" / "probe.py"
script.write_text('print("ok")\n')
@@ -326,7 +328,8 @@ class TestRunJobScript:
stay a plain `python script.py` — the bootstrap is overlay-only.
Cross-platform: forces the non-overlay branch explicitly."""
from cron import scheduler as sched_mod
from cron.scheduler import _run_job_script
from cron import scheduler_script as sched_script
from cron.scheduler_script import _run_job_script
script = cron_env / "scripts" / "probe.py"
script.write_text('print("ok")\n', encoding="utf-8")
@@ -344,9 +347,7 @@ class TestRunJobScript:
def communicate(self, timeout=None):
return ("ok\n", "")
monkeypatch.setattr(
sched_mod,
"_windows_cron_python_invocation",
monkeypatch.setattr(sched_script, "_windows_cron_python_invocation",
lambda python_exe: (python_exe, {}),
)
monkeypatch.setattr(sched_mod.subprocess, "Popen", FakeProc)
@@ -365,7 +366,7 @@ class TestRunJobScript:
carry emoji through. Either way the delivery content is the real
text, never an exception.
"""
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
script = cron_env / "scripts" / "emoji.py"
script.write_text(
@@ -385,7 +386,7 @@ class TestRunJobScript:
silently drop the whole delivery (#42384). The run may fail, but it
must fail as a (False, message) result the scheduler can deliver.
"""
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
script = cron_env / "scripts" / "bad_bytes.py"
# b'\xe6\x97' is the first two bytes of a three-byte CJK sequence —
@@ -493,7 +494,7 @@ class TestScriptPathContainment:
def test_absolute_path_outside_scripts_dir_blocked(self, cron_env):
"""Absolute paths outside ~/.hermes/scripts/ must be rejected."""
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
# Create a script outside the scripts dir
outside_script = cron_env / "outside.py"
@@ -506,7 +507,7 @@ class TestScriptPathContainment:
def test_tilde_path_blocked(self, cron_env):
"""~ prefixed paths must be rejected (expanduser bypasses check)."""
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
success, output = _run_job_script("~/evil.py")
assert success is False
@@ -514,7 +515,7 @@ class TestScriptPathContainment:
def test_tilde_traversal_blocked(self, cron_env):
"""~/../../../tmp/evil.py must be rejected."""
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
success, output = _run_job_script("~/../../../tmp/evil.py")
assert success is False
@@ -522,7 +523,7 @@ class TestScriptPathContainment:
def test_relative_traversal_still_blocked(self, cron_env):
"""../../etc/passwd style traversal must still be blocked."""
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
success, output = _run_job_script("../../etc/passwd")
assert success is False
@@ -530,7 +531,7 @@ class TestScriptPathContainment:
def test_relative_path_inside_scripts_dir_allowed(self, cron_env):
"""Relative paths within the scripts dir should still work."""
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
script = cron_env / "scripts" / "good.py"
script.write_text('print("ok")\n')
@@ -541,7 +542,7 @@ class TestScriptPathContainment:
def test_subdirectory_inside_scripts_dir_allowed(self, cron_env):
"""Relative paths to subdirectories within scripts/ should work."""
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
subdir = cron_env / "scripts" / "monitors"
subdir.mkdir()
@@ -559,7 +560,7 @@ class TestScriptPathContainment:
)
def test_symlink_escape_blocked(self, cron_env, tmp_path):
"""Symlinks pointing outside scripts/ must be rejected."""
from cron.scheduler import _run_job_script
from cron.scheduler_script import _run_job_script
# Create a script outside the scripts dir
outside = tmp_path / "outside_evil.py"
@@ -639,18 +640,17 @@ class TestScriptTimeoutTreeKill:
def test_unified_tree_kill_failure_falls_back(self, monkeypatch, caplog):
from agent import deadline
from cron import scheduler as sched
from cron import scheduler_script as sched_script
proc = SimpleNamespace(pid=12345, poll=lambda: None)
fallback_calls = []
monkeypatch.setattr(deadline, "kill_process_tree", lambda _pid: False)
monkeypatch.setattr(
sched,
"_terminate_cron_script_process",
monkeypatch.setattr(sched_script, "_terminate_cron_script_process",
lambda candidate: fallback_calls.append(candidate),
)
with caplog.at_level("WARNING", logger=sched.__name__):
sched._terminate_cron_script_tree(cast("subprocess.Popen", proc))
sched_script._terminate_cron_script_tree(cast("subprocess.Popen", proc))
assert fallback_calls == [proc]
assert "falling back to process-group termination" in caplog.text
@@ -658,6 +658,7 @@ class TestScriptTimeoutTreeKill:
def test_invalid_pid_never_reaches_unified_tree_kill(self, monkeypatch, caplog):
from agent import deadline
from cron import scheduler as sched
from cron import scheduler_script as sched_script
proc = SimpleNamespace(pid=0, poll=lambda: None)
tree_kill_calls = []
@@ -667,14 +668,12 @@ class TestScriptTimeoutTreeKill:
"kill_process_tree",
lambda pid: tree_kill_calls.append(pid),
)
monkeypatch.setattr(
sched,
"_terminate_cron_script_process",
monkeypatch.setattr(sched_script, "_terminate_cron_script_process",
lambda candidate: fallback_calls.append(candidate),
)
with caplog.at_level("WARNING", logger=sched.__name__):
sched._terminate_cron_script_tree(cast("subprocess.Popen", proc))
sched_script._terminate_cron_script_tree(cast("subprocess.Popen", proc))
assert tree_kill_calls == []
assert fallback_calls == [proc]
@@ -685,6 +684,7 @@ class TestScriptTimeoutTreeKill:
and must not produce a spurious "no signal" warning."""
from agent import deadline
from cron import scheduler as sched
from cron import scheduler_script as sched_script
proc = SimpleNamespace(pid=12345, poll=lambda: 0)
tree_kill_calls = []
@@ -694,13 +694,11 @@ class TestScriptTimeoutTreeKill:
"kill_process_tree",
lambda pid: tree_kill_calls.append(pid) or True,
)
monkeypatch.setattr(
sched,
"_terminate_cron_script_process",
monkeypatch.setattr(sched_script, "_terminate_cron_script_process",
lambda candidate: fallback_calls.append(candidate),
)
sched._terminate_cron_script_tree(cast("subprocess.Popen", proc))
sched_script._terminate_cron_script_tree(cast("subprocess.Popen", proc))
assert tree_kill_calls == []
assert fallback_calls == []
@@ -709,6 +707,7 @@ class TestScriptTimeoutTreeKill:
"""The ownership-lost/cancel kill site is the timeout site's sibling:
it must go through the same tree-kill (#71148 class)."""
from cron import scheduler as sched
from cron import scheduler_script as sched_script
tree_calls = []
@@ -718,7 +717,7 @@ class TestScriptTimeoutTreeKill:
tree_calls.append(proc.pid)
proc.kill()
monkeypatch.setattr(sched, "_terminate_cron_script_tree", _record_and_kill)
monkeypatch.setattr(sched_script, "_terminate_cron_script_tree", _record_and_kill)
class _Cancelled:
def is_set(self):
@@ -731,7 +730,7 @@ class TestScriptTimeoutTreeKill:
(scripts_dir / "long.py").write_text(
"import time; time.sleep(30)\n", encoding="utf-8"
)
ok, out = sched._run_job_script(
ok, out = sched_script._run_job_script(
str(scripts_dir / "long.py"),
workdir=str(cron_env),
cancel_event=_Cancelled(),
@@ -753,6 +752,7 @@ class TestScriptTimeoutTreeKill:
)
from cron import scheduler as sched
from cron import scheduler_script as sched_script
def is_live(pid):
try:
@@ -779,7 +779,7 @@ class TestScriptTimeoutTreeKill:
monkeypatch.setenv("HERMES_CRON_SCRIPT_TIMEOUT", "2")
monkeypatch.setattr(sched, "_SCRIPT_TIMEOUT", sched._DEFAULT_SCRIPT_TIMEOUT)
ok, out = sched._run_job_script(
ok, out = sched_script._run_job_script(
str(scripts_dir / "spawner.py"), workdir=str(cron_env)
)
assert not ok, f"script should have timed out, got {out!r}"
tests/cron/test_stale_module_leaf_imports.py
+1 -1
View File
@@ -57,7 +57,7 @@ def test_docker_import_ignores_stale_base_environment(monkeypatch):
# Model base.py cached before the shared sanitizer existed. A Docker module
# imported later by tool discovery must get the helper from the leaf module.
monkeypatch.delattr(base, "sanitize_task_id_for_path")
monkeypatch.delattr(base, "sanitize_task_id_for_path", raising=False)
previous = sys.modules.pop("tools.environments.docker", None)
try:
docker = importlib.import_module("tools.environments.docker")
tests/gateway/test_agent_cache.py
+2 -2
View File
@@ -650,7 +650,7 @@ class TestAgentCacheIdleResume:
def test_release_clients_does_not_touch_terminal_or_browser(self, monkeypatch):
"""release_clients must not call cleanup_vm or cleanup_browser."""
from run_agent import AIAgent
from tools import terminal_tool as _tt
from tools import terminal_tool_lifecycle as _tt
from tools import browser_tool as _bt
agent = AIAgent(
@@ -716,7 +716,7 @@ class TestAgentCacheIdleResume:
vm_calls: list = []
# AIAgent.close() calls the ``cleanup_vm`` name bound into
# ``run_agent`` at import time, not ``tools.terminal_tool.cleanup_vm``
# ``run_agent`` at import time, not ``tools.terminal_tool_lifecycle.cleanup_vm``
# directly — so patch the ``run_agent`` reference.
original_vm = _ra.cleanup_vm
_ra.cleanup_vm = lambda tid: vm_calls.append(tid)
tests/gateway/test_platform_base.py
+2 -1
View File
@@ -1336,7 +1336,8 @@ class TestDockerProfileSandboxMediaTranslation:
@staticmethod
def _sandbox_dir(task_id: str = "default"):
from tools.environments.base import get_sandbox_dir, sanitize_task_id_for_path
from tools.environments.base import get_sandbox_dir
from tools.environments.path_utils import sanitize_task_id_for_path
name = task_id if task_id == "default" else sanitize_task_id_for_path(task_id)
return get_sandbox_dir() / "docker" / name
tests/test_tui_gateway_server.py
+3 -3
View File
@@ -3999,11 +3999,11 @@ def test_session_cwd_set_profile_session_updates_profile_db(monkeypatch, tmp_pat
profile_db = ProfileDB()
import tools.terminal_tool as terminal_tool
import tools.terminal_tool_lifecycle as terminal_tool_lifecycle
monkeypatch.setattr("hermes_state.get_shared_session_db", lambda db_path=None: profile_db)
monkeypatch.setattr("hermes_state_registry.acquire", lambda db_path=None: profile_db)
monkeypatch.setattr(server, "_get_db", lambda: LaunchDB())
monkeypatch.setattr(terminal_tool, "cleanup_vm", lambda _key: None)
monkeypatch.setattr(terminal_tool_lifecycle, "cleanup_vm", lambda _key: None)
monkeypatch.setattr(server, "_register_session_cwd", lambda _session: None)
session = {"session_key": target, "profile_home": str(profile_home)}
tests/tools/test_approval_outcome_parity.py
+12 -10
View File
@@ -12,16 +12,18 @@ from __future__ import annotations
import pytest
import tools.approval as approval_mod
from tools import approval_human_wait
import tools.terminal_tool as terminal_tool
import tools.terminal_tool_sudo as terminal_tool_sudo
@pytest.fixture(autouse=True)
def _clean_human_wait_state():
with approval_mod._human_wait_lock:
approval_mod._human_wait_states.clear()
with approval_human_wait._human_wait_lock:
approval_human_wait._human_wait_states.clear()
yield
with approval_mod._human_wait_lock:
approval_mod._human_wait_states.clear()
with approval_human_wait._human_wait_lock:
approval_human_wait._human_wait_states.clear()
class TestSudoWaitExcludedFromDeadlines:
@@ -43,11 +45,11 @@ class TestSudoWaitExcludedFromDeadlines:
monkeypatch.setattr(
terminal_tool, "_get_sudo_password_callback", lambda: _slow_cb
)
before = approval_mod.human_wait_seconds(session)
pw = terminal_tool._prompt_for_sudo_password(timeout_seconds=5)
before = approval_human_wait.human_wait_seconds(session)
pw = terminal_tool_sudo._prompt_for_sudo_password(timeout_seconds=5)
assert pw == "pw"
after = approval_mod.human_wait_seconds(session)
after = approval_human_wait.human_wait_seconds(session)
assert after > before, (
f"sudo wait did not accrue human-wait time ({before} -> {after}); "
"the wait still counts against tool deadlines"
@@ -65,12 +67,12 @@ class TestSudoWaitExcludedFromDeadlines:
# read_password_thread writes into `result` via closure in the real
# code; stub the thread target by making join return quickly and the
# result dict empty -> returns "" but the wait must still be wrapped.
before = approval_mod.human_wait_seconds(session)
pw = terminal_tool._prompt_for_sudo_password(timeout_seconds=1)
before = approval_human_wait.human_wait_seconds(session)
pw = terminal_tool_sudo._prompt_for_sudo_password(timeout_seconds=1)
assert pw == ""
# The wrap is structural; a zero-length join may not move the clock,
# so assert only that no exception escaped and state stays consistent.
assert approval_mod.human_wait_seconds(session) >= before
assert approval_human_wait.human_wait_seconds(session) >= before
if __name__ == "__main__":
tests/tools/test_base_environment.py
+8 -7
View File
@@ -6,7 +6,8 @@ init_session() failure handling, and the CWD marker contract.
from unittest.mock import MagicMock
from tools.environments.base import BaseEnvironment, _BoundedOutputCollector
from tools.environments.base import BaseEnvironment
from tools.environments.base_output import _BoundedOutputCollector
class _TestableEnv(BaseEnvironment):
@@ -418,7 +419,7 @@ class TestSanitizeTaskIdForPath:
"""
def test_docker_unsafe_characters_are_replaced(self):
from tools.environments.base import sanitize_task_id_for_path
from tools.environments.path_utils import sanitize_task_id_for_path
out = sanitize_task_id_for_path("session:agent:main:telegram:dm:12345")
assert ":" not in out
@@ -426,13 +427,13 @@ class TestSanitizeTaskIdForPath:
def test_safe_ids_pass_through_verbatim(self):
"""Existing sandboxes keep resolving to their current directory."""
from tools.environments.base import sanitize_task_id_for_path
from tools.environments.path_utils import sanitize_task_id_for_path
for value in ("default", "task-01.abc_def", "astropy__astropy-12907"):
assert sanitize_task_id_for_path(value) == value
def test_deterministic_and_collision_free_for_distinct_inputs(self):
from tools.environments.base import sanitize_task_id_for_path
from tools.environments.path_utils import sanitize_task_id_for_path
assert sanitize_task_id_for_path("a:b") == sanitize_task_id_for_path("a:b")
# substitution alone is not injective — the digest must disambiguate
@@ -440,7 +441,7 @@ class TestSanitizeTaskIdForPath:
assert sanitize_task_id_for_path("!!!") != sanitize_task_id_for_path("@@@")
def test_empty_and_traversal_inputs_are_neutralized(self):
from tools.environments.base import sanitize_task_id_for_path
from tools.environments.path_utils import sanitize_task_id_for_path
assert sanitize_task_id_for_path("") == "default"
for value in (".", "..", "../../etc", "..\\..\\escape"):
@@ -449,7 +450,7 @@ class TestSanitizeTaskIdForPath:
assert "/" not in out and "\\" not in out
def test_oversized_input_truncates_with_unique_digest(self):
from tools.environments.base import (
from tools.environments.path_utils import (
_SANDBOX_DIR_MAX_LEN,
sanitize_task_id_for_path,
)
@@ -463,7 +464,7 @@ class TestSanitizeTaskIdForPath:
assert out_a != out_b
def test_sanitized_dir_is_creatable(self, tmp_path):
from tools.environments.base import sanitize_task_id_for_path
from tools.environments.path_utils import sanitize_task_id_for_path
target = tmp_path / "docker" / sanitize_task_id_for_path(
"session:agent:main:telegram:dm:12345"
tests/tools/test_code_execution.py
+2 -2
View File
@@ -248,7 +248,7 @@ class TestExecuteCode(unittest.TestCase):
def _run(self, code, enabled_tools=None):
"""Helper: run code with mocked handle_function_call."""
with patch("tools.code_execution_tool._rpc_server_loop") as mock_rpc:
with patch("tools.code_execution_rpc._rpc_server_loop") as mock_rpc:
# Use real execution but mock the tool dispatcher
pass
# Actually run with full integration, mocking at the model_tools level
@@ -837,7 +837,7 @@ class TestRpcTokenAuthorization(unittest.TestCase):
Sends each dict in *requests* as a newline-delimited JSON message
and returns the list of decoded JSON responses.
"""
from tools.code_execution_tool import _rpc_server_loop
from tools.code_execution_rpc import _rpc_server_loop
# socketpair gives us a connected client end and a "server" end we
# can hand to accept() by wrapping it in a tiny listener shim.
tests/tools/test_code_execution_modes.py
+9 -7
View File
@@ -44,18 +44,20 @@ def _fresh_kernel_registry():
shutdown_all_kernels()
from tools.code_execution_env import (
_is_usable_python,
_python_environment_prefix,
_python_prefix_cache,
_resolve_child_cwd,
_resolve_child_python,
_usable_python_cache,
_uses_hermes_python_environment,
)
from tools.code_execution_tool import (
SANDBOX_ALLOWED_TOOLS,
DEFAULT_EXECUTION_MODE,
EXECUTION_MODES,
_get_execution_mode,
_is_usable_python,
_python_environment_prefix,
_python_prefix_cache,
_usable_python_cache,
_resolve_child_cwd,
_resolve_child_python,
_uses_hermes_python_environment,
build_execute_code_schema,
execute_code,
)
tests/tools/test_code_execution_windows_env.py
+1 -1
View File
@@ -27,7 +27,7 @@ import textwrap
import pytest
from tools.code_execution_tool import (
from tools.code_execution_env import (
_SECRET_SUBSTRINGS,
_WINDOWS_ESSENTIAL_ENV_VARS,
_scrub_child_env,
tests/tools/test_container_cwd_sanitize.py
+4 -3
View File
@@ -32,7 +32,8 @@ class TestIsUnusableContainerCwd:
def test_container_backends_set(self):
assert tt._CONTAINER_BACKENDS == frozenset(
from tools.terminal_tool_config import _CONTAINER_BACKENDS
assert _CONTAINER_BACKENDS == frozenset(
{"docker", "singularity", "modal", "daytona", "vercel_sandbox"}
)
@@ -84,7 +85,7 @@ class TestOverrideCwdSanitizedAtCallSite:
monkeypatch.setattr(tt, "_get_env_config", lambda: config)
monkeypatch.setattr(tt, "_start_cleanup_thread", lambda: None)
monkeypatch.setattr(tt, "_check_all_guards", lambda *a, **k: {"approved": True})
monkeypatch.setattr(tt, "_create_environment", fake_create_environment)
monkeypatch.setattr("tools.terminal_tool_backends._create_environment", fake_create_environment)
# Force a fresh environment build so _create_environment is invoked.
monkeypatch.setattr(tt, "_active_environments", {})
monkeypatch.setattr(tt, "_last_activity", {})
@@ -174,7 +175,7 @@ class TestFileOpsCwdSanitizedAtCallSite:
monkeypatch.setattr(tt, "_get_env_config", lambda: config)
monkeypatch.setattr(tt, "_start_cleanup_thread", lambda: None)
monkeypatch.setattr(tt, "_create_environment", fake_create_environment)
monkeypatch.setattr("tools.terminal_tool_backends._create_environment", fake_create_environment)
# Force a fresh environment build.
monkeypatch.setattr(tt, "_active_environments", {})
monkeypatch.setattr(tt, "_last_activity", {})
tests/tools/test_delegate_kanban_isolation.py
+7 -4
View File
@@ -37,10 +37,11 @@ def _make_running_kanban_task(monkeypatch, tmp_path):
monkeypatch.setenv("HERMES_KANBAN_ATTACHMENTS_ROOT", str(attachments_root))
from hermes_cli import kanban_db as kb
from hermes_cli import kanban_db_connect as kbc
kb._INITIALIZED_PATHS.clear()
kb.init_db()
conn = kb.connect()
conn = kbc.connect()
try:
tid = kb.create_task(
conn,
@@ -151,7 +152,7 @@ def test_delegate_child_execute_code_env_bridges_contextvar_and_scrubs_kanban(
monkeypatch.delenv("HERMES_DELEGATED_CHILD_CONTEXT", raising=False)
from agent.delegation_context import delegated_child_context
from tools.code_execution_tool import _scrub_child_env
from tools.code_execution_env import _scrub_child_env
with delegated_child_context():
env = _scrub_child_env(
@@ -211,6 +212,7 @@ def test_delegate_child_kanban_cli_cannot_delete_parent_board(
def test_delegate_child_attach_url_guard_leaves_no_row_or_file(monkeypatch, tmp_path):
kb, tid, _workspace, attachments_root = _make_running_kanban_task(monkeypatch, tmp_path)
from hermes_cli import kanban_db_connect as kbc
from agent.delegation_context import delegated_child_context
from tools import kanban_tools
@@ -230,7 +232,7 @@ def test_delegate_child_attach_url_guard_leaves_no_row_or_file(monkeypatch, tmp_
assert payload["error"]
assert "delegate_task child" in payload["error"]
conn = kb.connect()
conn = kbc.connect()
try:
assert kb.list_attachments(conn, tid) == []
finally:
@@ -245,6 +247,7 @@ def test_child_attempting_default_complete_does_not_finish_parent_or_delete_work
):
"""Deterministic E2E: a delegated child cannot complete its parent task."""
kb, tid, workspace, _attachments_root = _make_running_kanban_task(monkeypatch, tmp_path)
from hermes_cli import kanban_db_connect as kbc
from tools import delegate_tool
from tools import kanban_tools
@@ -284,7 +287,7 @@ def test_child_attempting_default_complete_does_not_finish_parent_or_delete_work
result = delegate_tool._run_single_child(0, "try to complete parent", Child(), Parent())
conn = kb.connect()
conn = kbc.connect()
try:
task = kb.get_task(conn, tid)
run = kb.latest_run(conn, tid)
tests/tools/test_docker_environment.py
+2 -1
View File
@@ -1149,6 +1149,7 @@ def test_cleanup_vm_default_honors_persist_mode(monkeypatch):
_install_fake_thread(monkeypatch)
from tools import terminal_tool
from tools.terminal_tool_lifecycle import cleanup_vm
env = _make_dummy_env(task_id="session-close-test")
container_id = env._container_id
@@ -1164,7 +1165,7 @@ def test_cleanup_vm_default_honors_persist_mode(monkeypatch):
monkeypatch.setattr(docker_env.subprocess, "run", _capturing_run)
try:
terminal_tool.cleanup_vm("session-close-test")
cleanup_vm("session-close-test")
finally:
terminal_tool._active_environments.pop("session-close-test", None)
tests/tools/test_docker_session_isolation.py
+7 -6
View File
@@ -30,7 +30,8 @@ import os
import pytest
from tools import terminal_tool
from tools import terminal_tool, terminal_tool_backends
from tools.terminal_tool_lifecycle import is_persistent_env
@pytest.fixture(autouse=True)
@@ -282,7 +283,7 @@ class TestSessionScopedContainerLifecycle:
terminal_tool._active_environments, "tui:sess-1", _FakeEnv()
)
try:
assert terminal_tool.is_persistent_env("tui:sess-1") is True
assert is_persistent_env("tui:sess-1") is True
finally:
terminal_tool._active_environments.pop("tui:sess-1", None)
@@ -296,10 +297,10 @@ class TestSessionScopedContainerLifecycle:
def __init__(self, **kwargs):
captured.update(kwargs)
monkeypatch.setattr(terminal_tool, "_DockerEnvironment", _FakeDockerEnv)
monkeypatch.setattr(terminal_tool_backends, "_DockerEnvironment", _FakeDockerEnv)
monkeypatch.setattr(terminal_tool, "_maybe_reap_docker_orphans", lambda cc: None)
env = terminal_tool._create_environment(
env = terminal_tool_backends._create_environment(
env_type="docker", image="img:1", cwd="/workspace", timeout=60,
container_config={"docker_persist_across_processes": True},
task_id="tui:sess-1",
@@ -315,10 +316,10 @@ class TestSessionScopedContainerLifecycle:
def __init__(self, **kwargs):
captured.update(kwargs)
monkeypatch.setattr(terminal_tool, "_DockerEnvironment", _FakeDockerEnv)
monkeypatch.setattr(terminal_tool_backends, "_DockerEnvironment", _FakeDockerEnv)
monkeypatch.setattr(terminal_tool, "_maybe_reap_docker_orphans", lambda cc: None)
env = terminal_tool._create_environment(
env = terminal_tool_backends._create_environment(
env_type="docker", image="img:1", cwd="/workspace", timeout=60,
container_config={"docker_persist_across_processes": True},
task_id="default",
tests/tools/test_ensure_task_env.py
+7 -5
View File
@@ -6,6 +6,8 @@ from types import SimpleNamespace
from unittest.mock import patch
import tools.terminal_tool as tt
import tools.terminal_tool_backends as ttb
from tools.terminal_tool_lifecycle import get_active_env
def _clear(*task_ids):
@@ -17,7 +19,7 @@ def _clear(*task_ids):
def test_local_backend_is_noop(monkeypatch):
"""Local backend reads images host-side — no sandbox is created."""
monkeypatch.setenv("TERMINAL_ENV", "local")
with patch.object(tt, "_create_environment") as create:
with patch.object(ttb, "_create_environment") as create:
assert tt.ensure_task_env("t-local") is None
create.assert_not_called()
@@ -31,10 +33,10 @@ def test_non_local_creates_and_reuses(monkeypatch):
_clear(eff, task_id)
fake = SimpleNamespace(execute=lambda *a, **k: {"returncode": 0, "output": ""})
try:
with patch.object(tt, "_create_environment", return_value=fake) as create:
with patch.object(ttb, "_create_environment", return_value=fake) as create:
assert tt.ensure_task_env(task_id) is fake
create.assert_called_once()
assert tt.get_active_env(task_id) is fake
assert get_active_env(task_id) is fake
# Already active -> no second creation.
assert tt.ensure_task_env(task_id) is fake
@@ -51,8 +53,8 @@ def test_creation_failure_returns_none_and_caches_nothing(monkeypatch):
eff = tt._resolve_container_task_id(task_id)
_clear(eff, task_id)
try:
with patch.object(tt, "_create_environment", side_effect=RuntimeError("boom")):
with patch.object(ttb, "_create_environment", side_effect=RuntimeError("boom")):
assert tt.ensure_task_env(task_id) is None
assert tt.get_active_env(task_id) is None
assert get_active_env(task_id) is None
finally:
_clear(eff, task_id)
tests/tools/test_env_passthrough.py
+11 -14
View File
@@ -156,7 +156,7 @@ class TestExecuteCodeIntegration:
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
from tools.code_execution_env import _scrub_child_env
register_env_passthrough(["SERVICE_TOKEN"])
monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
@@ -172,7 +172,7 @@ class TestExecuteCodeIntegration:
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
from tools.code_execution_env import _scrub_child_env
register_env_passthrough(["SERVICE_TOKEN"])
monkeypatch.setenv("SERVICE_TOKEN", "token-for-default")
@@ -196,7 +196,7 @@ class TestExecuteCodeIntegration:
- BUZZ_RELAY_URL matches no secret substring but is not on the safe
prefix allowlist, so it is dropped too.
"""
from tools.code_execution_tool import _scrub_child_env
from tools.code_execution_env import _scrub_child_env
buzz_vars = {
"BUZZ_PRIVATE_KEY": "nsec1fake",
@@ -290,7 +290,8 @@ class TestTerminalIntegration:
assert missing["output"] == "unset"
def test_blocklisted_var_blocked_by_default(self):
from tools.environments.local import _sanitize_subprocess_env, _HERMES_PROVIDER_ENV_BLOCKLIST
from tools.environments.local import _sanitize_subprocess_env
from tools.environments.local_env_policy import _HERMES_PROVIDER_ENV_BLOCKLIST
# Pick a var we know is in the blocklist
blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
@@ -304,10 +305,8 @@ class TestTerminalIntegration:
Hermes provider credentials — that was the bypass where a skill
could declare ANTHROPIC_TOKEN / OPENAI_API_KEY as passthrough and
defeat the execute_code sandbox scrubbing."""
from tools.environments.local import (
_sanitize_subprocess_env,
_HERMES_PROVIDER_ENV_BLOCKLIST,
)
from tools.environments.local import _sanitize_subprocess_env
from tools.environments.local_env_policy import _HERMES_PROVIDER_ENV_BLOCKLIST
blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
# Attempt to register — must be silently refused (logged warning).
@@ -368,7 +367,7 @@ class TestTerminalIntegration:
result = _sanitize_subprocess_env({var: "value", "PATH": "/usr/bin"})
assert result.get(var) == "value"
# ...but the execute_code child never sees them.
from tools.code_execution_tool import _scrub_child_env
from tools.code_execution_env import _scrub_child_env
child_env = _scrub_child_env({var: "value", "PATH": "/usr/bin"})
assert var not in child_env
@@ -388,10 +387,8 @@ class TestTerminalIntegration:
def test_make_run_env_blocklist_override_rejected(self):
"""_make_run_env must NOT expose a blocklisted var to subprocess env
even after a skill attempts to register it via passthrough."""
from tools.environments.local import (
_make_run_env,
_HERMES_PROVIDER_ENV_BLOCKLIST,
)
from tools.environments.local import _make_run_env
from tools.environments.local_env_policy import _HERMES_PROVIDER_ENV_BLOCKLIST
blocked_var = next(iter(_HERMES_PROVIDER_ENV_BLOCKLIST))
os.environ[blocked_var] = "secret_value"
@@ -433,7 +430,7 @@ class TestTerminalIntegration:
"""
import builtins
from tools.code_execution_tool import _scrub_child_env
from tools.code_execution_env import _scrub_child_env
real_import = builtins.__import__
tests/tools/test_execute_code_approval_cluster.py
+31 -28
View File
@@ -9,7 +9,7 @@ Covers the canonical fix for issues #4146, #27303, #30882, #33057:
3. tools.approval.check_execute_code_guard — the entry-point guard decision
matrix (isolated backends, yolo/off, cron-deny, headless-local,
gateway approve/deny/timeout/missing-notify, smart mode).
4. tools.code_execution_tool._scrub_child_env — broad HERMES_ prefix dropped,
4. tools.code_execution_env._scrub_child_env — broad HERMES_ prefix dropped,
operational allowlist kept, DSN/WEBHOOK blocked, passthrough precedence.
"""
@@ -23,6 +23,9 @@ import threading
import pytest
from tools import approval as A
from tools import approval_context
from tools import approval_context
from tools import approval_smart
from tools.thread_context import propagate_context_to_thread
from gateway.session_context import clear_session_vars, reset_session_vars, set_session_vars
@@ -119,11 +122,11 @@ def gw_session(monkeypatch):
monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
# Force manual mode regardless of host config and disable any process-level
# yolo inherited from the developer's live environment.
monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False)
session_key = "cluster-test-session"
token = A.set_current_session_key(session_key)
token = approval_context.set_current_session_key(session_key)
with A._lock:
A._gateway_queues.pop(session_key, None)
A._gateway_notify_cbs.pop(session_key, None)
@@ -132,7 +135,7 @@ def gw_session(monkeypatch):
try:
yield session_key
finally:
A.reset_current_session_key(token)
approval_context.reset_current_session_key(token)
with A._lock:
A._gateway_queues.pop(session_key, None)
A._gateway_notify_cbs.pop(session_key, None)
@@ -180,7 +183,7 @@ def test_guard_headless_local_approved(monkeypatch):
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual")
assert A.check_execute_code_guard("import os", "local")["approved"] is True
@@ -188,8 +191,8 @@ def test_guard_cron_deny_blocks(monkeypatch):
monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False)
monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(A, "_get_cron_approval_mode", lambda: "deny")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(approval_context, "_get_cron_approval_mode", lambda: "deny")
tokens = set_session_vars(cron_session="1")
try:
res = A.check_execute_code_guard("import os", "local")
@@ -205,8 +208,8 @@ def test_guard_explicit_non_cron_masks_leaked_env(monkeypatch):
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(A, "_get_cron_approval_mode", lambda: "deny")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(approval_context, "_get_cron_approval_mode", lambda: "deny")
tokens = set_session_vars(cron_session="")
try:
res = A.check_execute_code_guard("import os", "local")
@@ -221,8 +224,8 @@ def test_guard_legacy_env_cron_still_blocks(monkeypatch):
monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False)
monkeypatch.setenv("HERMES_CRON_SESSION", "1")
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(A, "_get_cron_approval_mode", lambda: "deny")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(approval_context, "_get_cron_approval_mode", lambda: "deny")
res = A.check_execute_code_guard("import os", "local")
assert res["approved"] is False
assert res["outcome"] == "blocked"
@@ -260,20 +263,20 @@ def test_guard_gateway_missing_notify_is_pending(gw_session):
def test_guard_smart_mode(gw_session, monkeypatch):
monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(A, "_smart_approve", lambda c, d: "approve")
monkeypatch.setattr(approval_smart, "_smart_approve", lambda c, d: "approve")
res = A.check_execute_code_guard("import os", "local")
assert res["approved"] is True and res.get("smart_approved") is True
# Smart DENY on an interactive surface now asks the owner. With no bound
# notifier it remains pending rather than being hard-denied.
monkeypatch.setattr(A, "_smart_approve", lambda c, d: "deny")
monkeypatch.setattr(approval_smart, "_smart_approve", lambda c, d: "deny")
res = A.check_execute_code_guard("import os", "local")
assert res["approved"] is False and res["status"] == "pending_approval"
# escalate → falls through to manual gateway approval
monkeypatch.setattr(A, "_smart_approve", lambda c, d: "escalate")
monkeypatch.setattr(approval_smart, "_smart_approve", lambda c, d: "escalate")
_register_resolver(gw_session, "once")
res = A.check_execute_code_guard("import os", "local")
assert res["approved"] is True
@@ -284,8 +287,8 @@ def test_terminal_smart_deny_owner_override_is_one_operation(gw_session, monkeyp
with A._lock:
A._permanent_approved.discard("owner-override-test-danger")
A._session_approved.get(gw_session, set()).discard("owner-override-test-danger")
monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(A, "_smart_approve", lambda _command, _description: "deny")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(approval_smart, "_smart_approve", lambda _command, _description: "deny")
monkeypatch.setattr(
A,
"detect_dangerous_command",
@@ -317,8 +320,8 @@ def test_execute_code_smart_deny_owner_override_is_one_operation(gw_session, mon
with A._lock:
A._permanent_approved.discard("execute_code")
A._session_approved.get(gw_session, set()).discard("execute_code")
monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(A, "_smart_approve", lambda _command, _description: "deny")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(approval_smart, "_smart_approve", lambda _command, _description: "deny")
shown = _register_capturing_resolver(gw_session, "session")
result = A.check_execute_code_guard("print('first')", "local")
@@ -340,8 +343,8 @@ def test_smart_escalate_still_persists_session_choice(gw_session, monkeypatch):
key = "smart-escalate-persistence"
with A._lock:
A._session_approved.get(gw_session, set()).discard(key)
monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(A, "_smart_approve", lambda _command, _description: "escalate")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(approval_smart, "_smart_approve", lambda _command, _description: "escalate")
monkeypatch.setattr(
A, "detect_dangerous_command",
lambda command: (True, key, f"risk:{command}"),
@@ -362,8 +365,8 @@ def test_smart_escalate_still_persists_session_choice(gw_session, monkeypatch):
def test_terminal_smart_deny_pending_payload_is_one_operation(gw_session, monkeypatch):
monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(A, "_smart_approve", lambda _command, _description: "deny")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(approval_smart, "_smart_approve", lambda _command, _description: "deny")
monkeypatch.setattr(
A, "detect_dangerous_command",
lambda command: (True, "pending-smart-deny", f"risk:{command}"),
@@ -386,8 +389,8 @@ def test_terminal_smart_deny_pending_payload_is_one_operation(gw_session, monkey
def test_execute_code_smart_deny_pending_payload_is_one_operation(gw_session, monkeypatch):
monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(A, "_smart_approve", lambda _command, _description: "deny")
monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart")
monkeypatch.setattr(approval_smart, "_smart_approve", lambda _command, _description: "deny")
result = A.check_execute_code_guard("print('pending')", "local")
@@ -438,7 +441,7 @@ def test_guard_session_yolo_bypasses(gw_session):
# ---------------------------------------------------------------------------
def test_env_scrub_hermes_allowlist_and_secret_blocks():
from tools.code_execution_tool import _scrub_child_env
from tools.code_execution_env import _scrub_child_env
env = {
# operational allowlist → kept
@@ -472,7 +475,7 @@ def test_env_scrub_hermes_allowlist_and_secret_blocks():
def test_env_scrub_passthrough_overrides_secret_block():
"""A skill/config-declared passthrough var is an explicit user opt-in and
passes even if it matches a secret substring (precedence is intentional)."""
from tools.code_execution_tool import _scrub_child_env
from tools.code_execution_env import _scrub_child_env
env = {"MY_SERVICE_DSN": "value"}
out = _scrub_child_env(env, is_passthrough=lambda k: k == "MY_SERVICE_DSN",
@@ -494,7 +497,7 @@ def test_env_scrub_no_log_when_nothing_dropped(caplog):
"""No diagnostic noise when there are no dropped HERMES_* vars."""
import logging
from tools.code_execution_tool import _scrub_child_env
from tools.code_execution_env import _scrub_child_env
with caplog.at_level(logging.DEBUG, logger="tools.code_execution_tool"):
_scrub_child_env(
tests/tools/test_file_tools.py
+2 -2
View File
@@ -547,7 +547,7 @@ class TestSessionCwdSurvivesEnvRecreation:
@patch("tools.terminal_tool._active_environments", new_callable=dict)
@patch("tools.file_tools._file_ops_cache", new_callable=dict)
@patch("tools.terminal_tool._get_env_config")
@patch("tools.terminal_tool._create_environment")
@patch("tools.terminal_tool_backends._create_environment")
def test_recorded_cwd_used_for_recreated_env(
self, mock_create_env, mock_config, mock_cache, mock_active
):
@@ -588,7 +588,7 @@ class TestSessionCwdSurvivesEnvRecreation:
@patch("tools.terminal_tool._active_environments", new_callable=dict)
@patch("tools.file_tools._file_ops_cache", new_callable=dict)
@patch("tools.terminal_tool._get_env_config")
@patch("tools.terminal_tool._create_environment")
@patch("tools.terminal_tool_backends._create_environment")
def test_stale_cache_cwd_rescued_into_record_on_cleanup_detection(
self, mock_create_env, mock_config, mock_cache, mock_active
):
tests/tools/test_file_tools_container_config.py
+1 -1
View File
@@ -40,7 +40,7 @@ class TestFileToolsContainerConfig:
patch("tools.terminal_tool._active_environments", {}), \
patch("tools.terminal_tool._creation_locks", {}), \
patch("tools.terminal_tool._creation_locks_lock", __import__("threading").Lock()), \
patch("tools.terminal_tool._create_environment", side_effect=fake_create_env), \
patch("tools.terminal_tool_backends._create_environment", side_effect=fake_create_env), \
patch("tools.terminal_tool._start_cleanup_thread"), \
patch("tools.terminal_tool._check_disk_usage_warning"), \
patch("tools.file_tools._file_ops_cache", {}), \
tests/tools/test_file_write_surrogate_roundtrip.py
+1 -1
View File
@@ -12,7 +12,7 @@ from unittest.mock import MagicMock
import pytest
from tools.environments.base import _pipe_stdin
from tools.environments.base_output import _pipe_stdin
from tools.environments.local import LocalEnvironment
from tools.file_operations import ShellFileOperations
tests/tools/test_find_shell.py
+10 -7
View File
@@ -111,8 +111,9 @@ class TestFindBashSkipsBrokenCustomPath:
%LOCALAPPDATA%\\hermes\\git → Program Files) only exists in
``_find_bash``'s Windows branch."""
import tools.environments.local as local_mod
from tools.environments import local_gitbash_probe as gitbash_probe
local_mod._bash_starts_cache.clear()
gitbash_probe._bash_starts_cache.clear()
broken = tmp_path / "broken" / "bash.exe"
broken.parent.mkdir()
@@ -140,9 +141,10 @@ class TestGitBashExternalProgramProbe:
every host, so this stays on the Linux runner with ``subprocess.run``
mocked — no platform faking needed."""
import tools.environments.local as local_mod
from tools.environments import local_gitbash_probe as gitbash_probe
local_mod._bash_starts_cache.clear()
local_mod._bash_probe_details_cache.clear()
gitbash_probe._bash_starts_cache.clear()
gitbash_probe._bash_probe_details_cache.clear()
calls = []
def fake_run(argv, **kwargs):
@@ -151,7 +153,7 @@ class TestGitBashExternalProgramProbe:
monkeypatch.setattr(local_mod.subprocess, "run", fake_run)
assert local_mod._bash_starts(r"C:\Git\bin\bash.exe") is True
assert gitbash_probe._bash_starts(r"C:\Git\bin\bash.exe") is True
assert calls[0][0][-1] == "/usr/bin/true; /usr/bin/cat --version >/dev/null"
@pytest.mark.windows_only
@@ -162,9 +164,10 @@ class TestGitBashExternalProgramProbe:
``_find_bash``'s Windows candidate ladder and names PowerShell's
``Set-ProcessMitigation`` — unreachable off Windows."""
import tools.environments.local as local_mod
from tools.environments import local_gitbash_probe as gitbash_probe
local_mod._bash_starts_cache.clear()
local_mod._bash_probe_details_cache.clear()
gitbash_probe._bash_starts_cache.clear()
gitbash_probe._bash_probe_details_cache.clear()
portable = tmp_path / "hermes" / "git" / "bin" / "bash.exe"
portable.parent.mkdir(parents=True)
portable.write_text("", encoding="utf-8")
@@ -177,7 +180,7 @@ class TestGitBashExternalProgramProbe:
monkeypatch.setattr(local_mod, "_mandatory_aslr_enabled", lambda: True)
def failed_probe(path: str) -> bool:
local_mod._bash_probe_details_cache[path] = (
gitbash_probe._bash_probe_details_cache[path] = (
"dofork: child -1 - forked process died unexpectedly"
)
return False
tests/tools/test_hermes_subprocess_env.py
+2 -5
View File
@@ -14,11 +14,8 @@ full credential environment. Two tiers:
import os
from unittest.mock import patch
from tools.environments.local import (
hermes_subprocess_env,
_ALWAYS_STRIP_KEYS,
_HERMES_PROVIDER_ENV_FORCE_PREFIX,
)
from tools.environments.local import hermes_subprocess_env
from tools.environments.local_env_policy import _ALWAYS_STRIP_KEYS, _HERMES_PROVIDER_ENV_FORCE_PREFIX
_TIER1_SAMPLE = {
tests/tools/test_local_env_blocklist.py
+50 -42
View File
@@ -17,8 +17,8 @@ from unittest.mock import MagicMock, patch
import pytest
from tools.environments.local import (
LocalEnvironment,
from tools.environments.local import LocalEnvironment
from tools.environments.local_env_policy import (
_HERMES_PROVIDER_ENV_BLOCKLIST,
_HERMES_PROVIDER_ENV_FORCE_PREFIX,
)
@@ -62,7 +62,7 @@ def _run_with_env(extra_os_env=None, self_env=None):
with patch("tools.environments.local._find_bash", return_value="/bin/bash"), \
patch("subprocess.Popen", side_effect=_make_fake_popen(captured)), \
patch("tools.terminal_tool._interrupt_event", fake_interrupt), \
patch("tools.interrupt._interrupt_event", fake_interrupt), \
patch.dict(os.environ, test_environ, clear=True):
env.execute("echo hello")
@@ -595,7 +595,7 @@ class TestActiveVenvMarkerStripping:
assert result.get("HOME") == "/home/user"
def test_markers_constant_contents(self):
from tools.environments.local import _ACTIVE_VENV_MARKER_VARS
from tools.environments.local_env_policy import _ACTIVE_VENV_MARKER_VARS
assert "VIRTUAL_ENV" in _ACTIVE_VENV_MARKER_VARS
assert "CONDA_PREFIX" in _ACTIVE_VENV_MARKER_VARS
@@ -657,7 +657,7 @@ class TestPythonpathSelectiveStrip:
(PYTHONPATH key removed), and mixed user/Hermes ordering with an
empty component preserved.
"""
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
venv_sp = str(_running_venv_site_packages())
local_file = Path(__import__("tools.environments.local", fromlist=["__file__"]).__file__).resolve()
@@ -697,7 +697,7 @@ class TestPythonpathSelectiveStrip:
the same contract -- ownership is decided by provenance, never by
path shape or version (P1/P2, #74817 follow-ups).
"""
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
env = {"PYTHONPATH": user_pp}
_strip_hermes_owned_pythonpath(env)
assert env.get("PYTHONPATH") == user_pp
@@ -711,7 +711,7 @@ class TestPythonpathSelectiveStrip:
injects a direct child as a standalone entry, so such paths are user
paths by contract.
"""
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
import sys
running_minor = sys.version_info[1]
@@ -745,7 +745,7 @@ class TestPythonpathSelectiveStrip:
by the same Hermes-owned check (covered by the Windows-only test
below).
"""
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
import sys
pyver = f"python{sys.version_info[0]}.{sys.version_info[1]}"
@@ -771,7 +771,7 @@ class TestPythonpathSelectiveStrip:
user Windows path is preserved. Windows-only: POSIX ``Path`` does
not split on backslashes, so this cannot be meaningfully simulated
on a POSIX host."""
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
venv_sp = str(_running_venv_site_packages())
# Windows form: C:\...\venv\Lib\site-packages (backslashes)
@@ -787,7 +787,7 @@ class TestPythonpathSelectiveStrip:
def test_empty_pythonpath_unchanged(self):
"""An empty PYTHONPATH is a no-op (falsy -> early return)."""
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
env = {"PYTHONPATH": ""}
_strip_hermes_owned_pythonpath(env)
# Empty string is falsy, so the function returns early without
@@ -796,7 +796,7 @@ class TestPythonpathSelectiveStrip:
def test_empty_component_preserved(self):
"""An empty component means cwd and must survive unchanged."""
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
user_pp = os.pathsep.join(["/foo", "", "/bar"])
env = {"PYTHONPATH": user_pp}
@@ -807,7 +807,7 @@ class TestPythonpathSelectiveStrip:
def test_raw_user_spelling_preserved(self):
"""The sanitizer does not trim, normalize, or deduplicate user entries."""
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
user_pp = os.pathsep.join([
" /opt/user-lib ",
@@ -831,6 +831,7 @@ class TestPythonpathSelectiveStrip:
identifies ``<repo>/venv`` as the Hermes runtime producer contract.
"""
import tools.environments.local as local
from tools.environments import local_pythonpath
repo_root = tmp_path / "hermes-agent"
runtime_venv = repo_root / "venv"
@@ -860,6 +861,7 @@ class TestPythonpathSelectiveStrip:
def test_unrelated_virtual_env_is_not_runtime_provenance(self, tmp_path, monkeypatch):
"""An arbitrary inherited VIRTUAL_ENV cannot claim PYTHONPATH ownership."""
import tools.environments.local as local
from tools.environments import local_pythonpath
repo_root = tmp_path / "hermes-agent"
repo_root.mkdir()
@@ -876,14 +878,14 @@ class TestPythonpathSelectiveStrip:
"VIRTUAL_ENV": str(unrelated_venv),
"PYTHONPATH": str(unrelated_sp),
}
local._strip_hermes_owned_pythonpath(env)
local_pythonpath._strip_hermes_owned_pythonpath(env)
assert env["PYTHONPATH"] == str(unrelated_sp)
def test_no_pythonpath_key(self):
"""Missing PYTHONPATH key is a no-op."""
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
env = {"PATH": "/usr/bin"}
_strip_hermes_owned_pythonpath(env)
assert "PYTHONPATH" not in env
@@ -924,8 +926,8 @@ class TestPythonpathSelectiveStrip:
_strip_hermes_owned_pythonpath is applied (as the spawn path does),
while user entries (even for another Python version) are preserved.
"""
from tools.code_execution_tool import _scrub_child_env
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.code_execution_env import _scrub_child_env
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
venv_sp = str(_running_venv_site_packages())
other_sp = "/opt/other-venv/lib/python3.99/site-packages"
@@ -1043,7 +1045,7 @@ class TestPythonpathSelectiveStrip:
PYTHONPATH entry. A user path that merely happens to live under
the repo directory must therefore be preserved.
"""
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
local_file = Path(__import__("tools.environments.local", fromlist=["__file__"]).__file__).resolve()
real_repo_root = local_file.parents[2]
@@ -1061,6 +1063,7 @@ class TestPythonpathSelectiveStrip:
def test_configured_home_alias_matches_launcher_output(self, tmp_path, monkeypatch):
"""The real producer spelling is derived and consumed end to end."""
import tools.environments.local as local
from tools.environments import local_pythonpath
from hermes_cli.gateway_windows import _preserve_hermes_home_path
physical_home = tmp_path / "physical-home"
@@ -1073,7 +1076,7 @@ class TestPythonpathSelectiveStrip:
monkeypatch.setenv("HERMES_HOME", str(configured_home))
launcher_entry = Path(_preserve_hermes_home_path(physical_root))
aliases = local._build_hermes_repo_root_aliases(
aliases = local_pythonpath._build_hermes_repo_root_aliases(
physical_root.resolve(),
physical_root,
configured_home,
@@ -1091,7 +1094,7 @@ class TestPythonpathSelectiveStrip:
"/home/user/my-lib",
])
}
local._strip_hermes_owned_pythonpath(env)
local_pythonpath._strip_hermes_owned_pythonpath(env)
assert env["PYTHONPATH"].split(os.pathsep) == [
str(nested_user_path),
@@ -1109,6 +1112,7 @@ class TestPythonpathSelectiveStrip:
repo-root entry is stripped.
"""
import tools.environments.local as local
from tools.environments import local_pythonpath
from hermes_cli.profiles import resolve_profile_env
physical_home = tmp_path / "physical-home"
@@ -1131,16 +1135,16 @@ class TestPythonpathSelectiveStrip:
assert Path(resolve_profile_env("coder")) == configured_home / "profiles" / "coder"
# The sanitizer now runs under the re-homed (profile) HERMES_HOME.
aliases = local._build_hermes_repo_root_aliases(
aliases = local_pythonpath._build_hermes_repo_root_aliases(
physical_root.resolve(),
physical_root,
configured_home / "profiles" / "coder",
)
assert any(local._same_path(a, lexical_root) for a in aliases)
assert any(local_pythonpath._same_path(a, lexical_root) for a in aliases)
monkeypatch.setattr(local, "_hermes_repo_root_aliases", aliases)
env = {"PYTHONPATH": os.pathsep.join([str(lexical_root), "/home/user/my-lib"])}
local._strip_hermes_owned_pythonpath(env)
local_pythonpath._strip_hermes_owned_pythonpath(env)
assert env["PYTHONPATH"].split(os.pathsep) == ["/home/user/my-lib"]
@@ -1152,6 +1156,7 @@ class TestPythonpathSelectiveStrip:
proof (strict resolve), not a name-based guess.
"""
import tools.environments.local as local
from tools.environments import local_pythonpath
physical_root = _physical_repo_root(tmp_path)
configured_home = tmp_path / "configured-home"
@@ -1163,16 +1168,16 @@ class TestPythonpathSelectiveStrip:
pytest.skip(f"directory link unavailable on this host: {exc}")
lexical_root = configured_home / "hermes-agent"
aliases = local._build_hermes_repo_root_aliases(
aliases = local_pythonpath._build_hermes_repo_root_aliases(
physical_root.resolve(),
physical_root,
configured_home,
)
assert any(local._same_path(a, lexical_root) for a in aliases)
assert any(local_pythonpath._same_path(a, lexical_root) for a in aliases)
monkeypatch.setattr(local, "_hermes_repo_root_aliases", aliases)
env = {"PYTHONPATH": os.pathsep.join([str(lexical_root), "/home/user/my-lib"])}
local._strip_hermes_owned_pythonpath(env)
local_pythonpath._strip_hermes_owned_pythonpath(env)
assert env["PYTHONPATH"].split(os.pathsep) == ["/home/user/my-lib"]
def test_same_named_non_owned_directories_preserved(self, tmp_path, monkeypatch):
@@ -1182,6 +1187,7 @@ class TestPythonpathSelectiveStrip:
not the name; no ownership provenance means no strip.
"""
import tools.environments.local as local
from tools.environments import local_pythonpath
physical_root = _physical_repo_root(tmp_path)
configured_home = tmp_path / "configured-home"
@@ -1189,18 +1195,18 @@ class TestPythonpathSelectiveStrip:
unrelated = tmp_path / "user-tools" / "hermes-agent"
unrelated.mkdir(parents=True)
aliases = local._build_hermes_repo_root_aliases(
aliases = local_pythonpath._build_hermes_repo_root_aliases(
physical_root.resolve(),
physical_root,
configured_home,
)
for lookalike in (configured_home / "hermes-agent", unrelated):
assert not any(local._same_path(a, lookalike) for a in aliases)
assert not any(local_pythonpath._same_path(a, lookalike) for a in aliases)
monkeypatch.setattr(local, "_hermes_repo_root_aliases", aliases)
for lookalike in (configured_home / "hermes-agent", unrelated):
env = {"PYTHONPATH": os.pathsep.join([str(lookalike), "/home/user/my-lib"])}
local._strip_hermes_owned_pythonpath(env)
local_pythonpath._strip_hermes_owned_pythonpath(env)
assert env["PYTHONPATH"].split(os.pathsep) == [str(lookalike), "/home/user/my-lib"]
def test_profile_home_with_repo_level_junction(self, tmp_path, monkeypatch):
@@ -1210,6 +1216,7 @@ class TestPythonpathSelectiveStrip:
the lexical repo alias recovered from it.
"""
import tools.environments.local as local
from tools.environments import local_pythonpath
physical_root = _physical_repo_root(tmp_path)
configured_root = tmp_path / "configured-root"
@@ -1221,17 +1228,17 @@ class TestPythonpathSelectiveStrip:
configured_home = configured_root / "profiles" / "coder"
lexical_root = configured_root / "hermes-agent"
aliases = local._build_hermes_repo_root_aliases(
aliases = local_pythonpath._build_hermes_repo_root_aliases(
physical_root.resolve(),
physical_root,
configured_home,
)
assert any(local._same_path(a, lexical_root) for a in aliases)
assert not any(local._same_path(a, configured_home / "hermes-agent") for a in aliases)
assert any(local_pythonpath._same_path(a, lexical_root) for a in aliases)
assert not any(local_pythonpath._same_path(a, configured_home / "hermes-agent") for a in aliases)
monkeypatch.setattr(local, "_hermes_repo_root_aliases", aliases)
env = {"PYTHONPATH": os.pathsep.join([str(lexical_root), "/home/user/my-lib"])}
local._strip_hermes_owned_pythonpath(env)
local_pythonpath._strip_hermes_owned_pythonpath(env)
assert env["PYTHONPATH"].split(os.pathsep) == ["/home/user/my-lib"]
def test_validated_runtime_venv_lexical_after_repo_recovery(self, tmp_path, monkeypatch):
@@ -1240,6 +1247,7 @@ class TestPythonpathSelectiveStrip:
stripped together with the repo root, while user entries survive.
"""
import tools.environments.local as local
from tools.environments import local_pythonpath
physical_root = _physical_repo_root(tmp_path)
venv_dir = physical_root / "venv"
@@ -1253,18 +1261,18 @@ class TestPythonpathSelectiveStrip:
pytest.skip(f"directory link unavailable on this host: {exc}")
lexical_root = configured_home / "hermes-agent"
aliases = local._build_hermes_repo_root_aliases(
aliases = local_pythonpath._build_hermes_repo_root_aliases(
physical_root.resolve(),
physical_root,
configured_home,
)
assert any(local._same_path(a, lexical_root) for a in aliases)
assert any(local_pythonpath._same_path(a, lexical_root) for a in aliases)
monkeypatch.setattr(local, "_hermes_repo_root_aliases", aliases)
lexical_venv = lexical_root / "venv"
validated = local._validated_runtime_venv({"VIRTUAL_ENV": str(lexical_venv)})
validated = local_pythonpath._validated_runtime_venv({"VIRTUAL_ENV": str(lexical_venv)})
assert validated is not None
assert local._same_path(validated, lexical_venv)
assert local_pythonpath._same_path(validated, lexical_venv)
local._hermes_site_packages = None
env = {"PYTHONPATH": os.pathsep.join([
@@ -1272,7 +1280,7 @@ class TestPythonpathSelectiveStrip:
str(lexical_venv / "Lib" / "site-packages"),
"/home/user/my-lib",
]), "VIRTUAL_ENV": str(lexical_venv)}
local._strip_hermes_owned_pythonpath(env)
local_pythonpath._strip_hermes_owned_pythonpath(env)
assert env["PYTHONPATH"].split(os.pathsep) == ["/home/user/my-lib"]
@@ -1320,7 +1328,7 @@ class TestPythonhomeSanitized:
def test_pythonhome_removed_from_active_venv_markers(self):
"""PYTHONHOME is part of _ACTIVE_VENV_MARKER_VARS so all builders
that iterate it drop the variable."""
from tools.environments.local import _ACTIVE_VENV_MARKER_VARS
from tools.environments.local_env_policy import _ACTIVE_VENV_MARKER_VARS
assert "PYTHONHOME" in _ACTIVE_VENV_MARKER_VARS
def test_build_subprocess_env_no_scrub_preserves_pythonhome(self):
@@ -1695,7 +1703,7 @@ class TestHermesInternalDynamicSecrets:
"""
def test_predicate_matches_auxiliary_api_key(self):
from tools.environments.local import _is_hermes_internal_secret
from tools.environments.local_env_policy import _is_hermes_internal_secret
assert _is_hermes_internal_secret("AUXILIARY_VISION_API_KEY")
assert _is_hermes_internal_secret("AUXILIARY_WEB_EXTRACT_API_KEY")
assert _is_hermes_internal_secret("AUXILIARY_APPROVAL_API_KEY")
@@ -1703,12 +1711,12 @@ class TestHermesInternalDynamicSecrets:
assert _is_hermes_internal_secret("AUXILIARY_MY_PLUGIN_TASK_API_KEY")
def test_predicate_matches_auxiliary_base_url(self):
from tools.environments.local import _is_hermes_internal_secret
from tools.environments.local_env_policy import _is_hermes_internal_secret
assert _is_hermes_internal_secret("AUXILIARY_VISION_BASE_URL")
assert _is_hermes_internal_secret("AUXILIARY_COMPRESSION_BASE_URL")
def test_predicate_matches_gateway_relay_auth(self):
from tools.environments.local import _is_hermes_internal_secret
from tools.environments.local_env_policy import _is_hermes_internal_secret
assert _is_hermes_internal_secret("GATEWAY_RELAY_SECRET")
assert _is_hermes_internal_secret("GATEWAY_RELAY_DELIVERY_KEY")
assert _is_hermes_internal_secret("GATEWAY_RELAY_SESSION_TOKEN")
@@ -1716,7 +1724,7 @@ class TestHermesInternalDynamicSecrets:
def test_predicate_allows_auxiliary_non_secrets(self):
"""AUXILIARY_*_PROVIDER / _MODEL and GATEWAY_RELAY_* routing hints are
NOT secrets and must remain visible so tooling that reads them works."""
from tools.environments.local import _is_hermes_internal_secret
from tools.environments.local_env_policy import _is_hermes_internal_secret
assert not _is_hermes_internal_secret("AUXILIARY_VISION_PROVIDER")
assert not _is_hermes_internal_secret("AUXILIARY_VISION_MODEL")
assert not _is_hermes_internal_secret("GATEWAY_RELAY_URL")
tests/tools/test_local_env_cwd_recovery.py
+2 -2
View File
@@ -99,7 +99,7 @@ class TestRunBashCwdRecovery:
try:
with patch("tools.environments.local._find_bash", return_value="/bin/bash"), \
patch("subprocess.Popen", side_effect=_make_fake_popen(captured, fds)), \
patch("tools.terminal_tool._interrupt_event", _fake_interrupt()), \
patch("tools.interrupt._interrupt_event", _fake_interrupt()), \
caplog.at_level("WARNING", logger="tools.environments.local"):
env.execute("echo hello")
finally:
@@ -124,7 +124,7 @@ class TestRunBashCwdRecovery:
try:
with patch("tools.environments.local._find_bash", return_value="/bin/bash"), \
patch("subprocess.Popen", side_effect=_make_fake_popen(captured, fds)), \
patch("tools.terminal_tool._interrupt_event", _fake_interrupt()), \
patch("tools.interrupt._interrupt_event", _fake_interrupt()), \
caplog.at_level("WARNING", logger="tools.environments.local"):
env.execute("echo hello")
finally:
tests/tools/test_managed_browserbase_and_modal.py
+4 -3
View File
@@ -264,14 +264,15 @@ def test_terminal_tool_respects_direct_modal_mode_without_falling_back_to_manage
env.pop("MODAL_TOKEN_SECRET", None)
with patch.dict(os.environ, env, clear=True):
terminal_tool = _load_tool_module("tools.terminal_tool", "terminal_tool.py")
_load_tool_module("tools.terminal_tool", "terminal_tool.py")
terminal_tool_backends = sys.modules["tools.terminal_tool_backends"]
with (
patch.object(terminal_tool, "is_managed_tool_gateway_ready", return_value=True),
patch.object(terminal_tool_backends, "is_managed_tool_gateway_ready", return_value=True),
patch.object(Path, "exists", return_value=False),
):
with pytest.raises(ValueError, match="direct Modal credentials"):
terminal_tool._create_environment(
terminal_tool_backends._create_environment(
env_type="modal",
image="python:3.11",
cwd="/root",
tests/tools/test_modal_sandbox_fixes.py
+3 -2
View File
@@ -168,9 +168,10 @@ class TestCwdHandling:
captured.update(kwargs)
return sentinel
monkeypatch.setattr(_tt_mod, "_DockerEnvironment", _fake_docker_environment)
from tools.terminal_tool_backends import _create_environment
monkeypatch.setattr("tools.terminal_tool_backends._DockerEnvironment", _fake_docker_environment)
env = _tt_mod._create_environment(
env = _create_environment(
env_type="docker",
image="python:3.11",
cwd="/workspace",
tests/tools/test_notify_on_complete.py
+1 -1
View File
@@ -199,7 +199,7 @@ class TestTerminalSchema:
class TestCodeExecutionBlocked:
def test_notify_on_complete_blocked_in_sandbox(self):
from tools.code_execution_tool import _TERMINAL_BLOCKED_PARAMS
from tools.code_execution_rpc import _TERMINAL_BLOCKED_PARAMS
assert "notify_on_complete" in _TERMINAL_BLOCKED_PARAMS
tests/tools/test_process_registry.py
+1 -1
View File
@@ -12,7 +12,7 @@ import time
import pytest
from unittest.mock import MagicMock, patch
from tools.environments.local import _HERMES_PROVIDER_ENV_FORCE_PREFIX
from tools.environments.local_env_policy import _HERMES_PROVIDER_ENV_FORCE_PREFIX
from tools.process_registry import (
ProcessRegistry,
ProcessSession,
tests/tools/test_singularity_persistent_overlay.py
+1 -1
View File
@@ -7,7 +7,7 @@ components must go through the shared sanitizer in tools.environments.base.
"""
from tools.environments import singularity as singularity_env
from tools.environments.base import sanitize_task_id_for_path
from tools.environments.path_utils import sanitize_task_id_for_path
def _stub_singularity(monkeypatch, tmp_path):
tests/tools/test_snapshot_multiline_session_env_injection.py
+1 -1
View File
@@ -18,7 +18,7 @@ from pathlib import Path
import pytest
from tools.environments.base import _export_dump_excluding_session_vars
from tools.environments.base_session_env import _export_dump_excluding_session_vars
def _bash() -> str:
tests/tools/test_snapshot_session_id_leak.py
+1 -1
View File
@@ -21,7 +21,7 @@ import sys
import pytest
from tools.environments.base import (
from tools.environments.base_session_env import (
_SNAPSHOT_EXCLUDED_ENV_REGEX,
_export_dump_excluding_session_vars,
)
tests/tools/test_ssh_environment.py
+1 -1
View File
@@ -29,7 +29,7 @@ def _run(command, task_id="ssh_test", **kwargs):
def _cleanup(task_id="ssh_test"):
from tools.terminal_tool import cleanup_vm
from tools.terminal_tool_lifecycle import cleanup_vm
cleanup_vm(task_id)
tests/tools/test_subagent_sudo_prompt.py
+12 -11
View File
@@ -21,6 +21,7 @@ import pytest
from agent.delegation_context import delegated_child_context
from tools import terminal_tool as tt
from tools import terminal_tool_sudo as tts
@pytest.fixture(autouse=True)
@@ -29,12 +30,12 @@ def _clean_sudo_state(monkeypatch):
monkeypatch.delenv("SUDO_PASSWORD", raising=False)
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
# Host sudoers NOPASSWD must not short-circuit the path under test.
monkeypatch.setattr(tt, "_sudo_nopasswd_works", lambda: False)
tt._reset_cached_sudo_passwords()
monkeypatch.setattr(tts, "_sudo_nopasswd_works", lambda: False)
tts._reset_cached_sudo_passwords()
tt.set_sudo_password_callback(None)
yield
tt.set_sudo_password_callback(None)
tt._reset_cached_sudo_passwords()
tts._reset_cached_sudo_passwords()
def _transform_in_child(command: str):
@@ -50,7 +51,7 @@ def _transform_in_child(command: str):
ctx = contextvars.copy_context()
def _worker():
result["value"] = ctx.run(tt._transform_sudo_command, command)
result["value"] = ctx.run(tts._transform_sudo_command, command)
t = threading.Thread(target=_worker)
t.start()
@@ -66,7 +67,7 @@ class TestDelegatedChildNeverPrompts:
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
calls = []
monkeypatch.setattr(
tt,
tts,
"_prompt_for_sudo_password",
lambda timeout_seconds=45: calls.append(1) or "hunter2",
)
@@ -88,7 +89,7 @@ class TestDelegatedChildNeverPrompts:
tt.set_sudo_password_callback(lambda: calls.append(1) or "pw")
try:
with delegated_child_context("child-session"):
return tt._transform_sudo_command("sudo systemctl restart foo")
return tts._transform_sudo_command("sudo systemctl restart foo")
finally:
tt.set_sudo_password_callback(None)
@@ -101,10 +102,10 @@ class TestDelegatedChildNeverPrompts:
"""The fix must not break interactive prompting outside children."""
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
monkeypatch.setattr(
tt, "_prompt_for_sudo_password", lambda timeout_seconds=45: "hunter2"
tts, "_prompt_for_sudo_password", lambda timeout_seconds=45: "hunter2"
)
transformed, sudo_stdin = tt._transform_sudo_command("sudo whoami")
transformed, sudo_stdin = tts._transform_sudo_command("sudo whoami")
assert sudo_stdin == "hunter2\n"
assert "sudo -S -p ''" in transformed
@@ -122,7 +123,7 @@ class TestDelegatedChildNeverPrompts:
class TestDelegatedChildFailureMessaging:
def test_child_gets_headless_sudo_tip(self):
with delegated_child_context("child-session"):
out = tt._handle_sudo_failure(
out = tts._handle_sudo_failure(
"sudo: a password is required", env_type="local"
)
assert "Subagents cannot prompt" in out
@@ -130,10 +131,10 @@ class TestDelegatedChildFailureMessaging:
def test_parent_output_unchanged(self, monkeypatch):
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
out = tt._handle_sudo_failure("sudo: a password is required", env_type="local")
out = tts._handle_sudo_failure("sudo: a password is required", env_type="local")
assert out == "sudo: a password is required"
def test_gateway_tip_preserved(self, monkeypatch):
monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
out = tt._handle_sudo_failure("sudo: a password is required", env_type="local")
out = tts._handle_sudo_failure("sudo: a password is required", env_type="local")
assert "To enable sudo over messaging" in out
tests/tools/test_terminal_compound_background.py
+1 -1
View File
@@ -17,7 +17,7 @@ import subprocess
import pytest
from tools.terminal_tool import _rewrite_compound_background as rewrite
from tools.terminal_tool_sudo import _rewrite_compound_background as rewrite
class TestRewrites:
tests/tools/test_terminal_error_redaction.py
+1 -1
View File
@@ -107,7 +107,7 @@ def test_environment_creation_import_error_redacts_exception_text(monkeypatch):
def fail_create_environment(**kwargs):
raise ImportError(f"backend import failed with {SECRET}")
monkeypatch.setattr(terminal_tool, "_create_environment", fail_create_environment)
monkeypatch.setattr("tools.terminal_tool_backends._create_environment", fail_create_environment)
result = json.loads(terminal_tool.terminal_tool(command="echo ok"))
tests/tools/test_terminal_exit_semantics.py
+1 -1
View File
@@ -2,7 +2,7 @@
import pytest
from tools.terminal_tool import _interpret_exit_code
from tools.terminal_tool_result import _interpret_exit_code
class TestInterpretExitCode:
tests/tools/test_terminal_heredoc_background_guard.py
+1 -1
View File
@@ -23,9 +23,9 @@ real background operator is not.
"""
from tools.shell_heredoc import strip_inert_heredoc_bodies
from tools.terminal_tool_guards import _strip_quotes
from tools.terminal_tool import (
_foreground_background_guidance as guidance,
_strip_quotes,
)
# Build commands without a literal '&' in this source where convenient, so the
tests/tools/test_terminal_hints.py
+4 -4
View File
@@ -107,9 +107,9 @@ class TestTerminalIntegration:
def test_hint_field_wired(self):
# Exercise the wiring path shape without a live environment: the
# result assembly guards on returncode != 0 and no exit_note.
from tools import terminal_tool
from tools import terminal_tool_result
# simulate: interpret gives None, hints give a value
note = terminal_tool._interpret_exit_code("python x.py", 127)
note = terminal_tool_result._interpret_exit_code("python x.py", 127)
assert note is None
hint = annotate_failure("python x.py", 127, "bash: python: command not found")
assert hint and "python3" in hint
@@ -118,8 +118,8 @@ class TestTerminalIntegration:
# grep exit 1 is informational; annotate_failure must not be reached
# for it in the wiring (exit_note wins). Just verify the semantics
# table still covers it.
from tools import terminal_tool
assert terminal_tool._interpret_exit_code("grep foo bar.txt", 1) is not None
from tools import terminal_tool_result
assert terminal_tool_result._interpret_exit_code("grep foo bar.txt", 1) is not None
class TestMaskedSuccess:
tests/tools/test_terminal_none_command_guard.py
+2 -1
View File
@@ -2,7 +2,8 @@
import json
from tools.terminal_tool import _transform_sudo_command, terminal_tool
from tools.terminal_tool import terminal_tool
from tools.terminal_tool_sudo import _transform_sudo_command
def test_transform_sudo_command_none_returns_cleanly():
tests/tools/test_terminal_requirements.py
+2 -1
View File
@@ -39,6 +39,7 @@ def _clear_terminal_env(monkeypatch):
monkeypatch.setattr(terminal_tool_module, "managed_nous_tools_enabled", lambda: False)
import tools.tool_backend_helpers as _tbh
monkeypatch.setattr(_tbh, "managed_nous_tools_enabled", lambda: False)
monkeypatch.setattr("tools.terminal_tool_backends.managed_nous_tools_enabled", lambda: False)
def test_local_terminal_requirements(monkeypatch, caplog):
@@ -73,7 +74,7 @@ def test_modal_backend_managed_mode_without_feature_flag_logs_clear_error(monkey
monkeypatch.setenv("TERMINAL_MODAL_MODE", "managed")
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.setattr(terminal_tool_module, "is_managed_tool_gateway_ready", lambda _vendor: False)
monkeypatch.setattr("tools.terminal_tool_backends.is_managed_tool_gateway_ready", lambda _vendor: False)
with caplog.at_level(logging.ERROR):
ok = terminal_tool_module.check_terminal_requirements()
tests/tools/test_terminal_signal_exit.py
+1 -1
View File
@@ -8,7 +8,7 @@ mis-diagnosing.
import pytest
from tools.terminal_tool import _interpret_exit_code, _interpret_signal_exit
from tools.terminal_tool_result import _interpret_exit_code, _interpret_signal_exit
class TestInterpretSignalExit:
tests/tools/test_terminal_tool.py
+11 -10
View File
@@ -1,14 +1,15 @@
"""Regression tests for sudo detection and sudo password handling."""
import tools.terminal_tool as terminal_tool
import tools.terminal_tool_sudo as terminal_tool_sudo
def setup_function():
terminal_tool._reset_cached_sudo_passwords()
terminal_tool_sudo._reset_cached_sudo_passwords()
def teardown_function():
terminal_tool._reset_cached_sudo_passwords()
terminal_tool_sudo._reset_cached_sudo_passwords()
def test_searching_for_sudo_does_not_trigger_rewrite(monkeypatch):
@@ -16,7 +17,7 @@ def test_searching_for_sudo_does_not_trigger_rewrite(monkeypatch):
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
command = "rg --line-number --no-heading --with-filename 'sudo' . | head -n 20"
transformed, sudo_stdin = terminal_tool._transform_sudo_command(command)
transformed, sudo_stdin = terminal_tool_sudo._transform_sudo_command(command)
assert transformed == command
assert sudo_stdin is None
@@ -35,7 +36,7 @@ def test_printf_literal_sudo_does_not_trigger_rewrite(monkeypatch):
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
command = "printf '%s\\n' sudo"
transformed, sudo_stdin = terminal_tool._transform_sudo_command(command)
transformed, sudo_stdin = terminal_tool_sudo._transform_sudo_command(command)
assert transformed == command
assert sudo_stdin is None
@@ -46,7 +47,7 @@ def test_non_command_argument_named_sudo_does_not_trigger_rewrite(monkeypatch):
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
command = "grep -n sudo README.md"
transformed, sudo_stdin = terminal_tool._transform_sudo_command(command)
transformed, sudo_stdin = terminal_tool_sudo._transform_sudo_command(command)
assert transformed == command
assert sudo_stdin is None
@@ -56,7 +57,7 @@ def test_actual_sudo_command_uses_configured_password(monkeypatch):
monkeypatch.setenv("SUDO_PASSWORD", "testpass")
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
transformed, sudo_stdin = terminal_tool._transform_sudo_command("sudo apt install -y ripgrep")
transformed, sudo_stdin = terminal_tool_sudo._transform_sudo_command("sudo apt install -y ripgrep")
assert transformed == "sudo -S -p '' apt install -y ripgrep"
assert sudo_stdin == "testpass\n"
@@ -69,9 +70,9 @@ def test_explicit_empty_sudo_password_tries_empty_without_prompt(monkeypatch):
def _fail_prompt(*_args, **_kwargs):
raise AssertionError("interactive sudo prompt should not run for explicit empty password")
monkeypatch.setattr(terminal_tool, "_prompt_for_sudo_password", _fail_prompt)
monkeypatch.setattr(terminal_tool_sudo, "_prompt_for_sudo_password", _fail_prompt)
transformed, sudo_stdin = terminal_tool._transform_sudo_command("sudo true")
transformed, sudo_stdin = terminal_tool_sudo._transform_sudo_command("sudo true")
assert transformed == "sudo -S -p '' true"
assert sudo_stdin == "\n"
@@ -104,5 +105,5 @@ def test_validate_workdir_still_blocks_metachars_in_unicode_paths():
def test_count_real_sudo_invocations_ignores_mentions(monkeypatch):
assert terminal_tool._count_real_sudo_invocations("grep sudo README.md") == 0
assert terminal_tool._count_real_sudo_invocations("sudo a; sudo b") == 2
assert terminal_tool_sudo._count_real_sudo_invocations("grep sudo README.md") == 0
assert terminal_tool_sudo._count_real_sudo_invocations("sudo a; sudo b") == 2
tests/tools/test_terminal_tool_requirements.py
+2 -3
View File
@@ -34,7 +34,7 @@ class TestTerminalRequirements:
def test_terminal_and_execute_code_tools_resolve_for_managed_modal(self, monkeypatch, tmp_path):
monkeypatch.setattr("tools.tool_backend_helpers.managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr(terminal_tool_module, "managed_nous_tools_enabled", lambda: True)
monkeypatch.setattr("tools.terminal_tool_backends.managed_nous_tools_enabled", lambda: True)
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.delenv("MODAL_TOKEN_ID", raising=False)
@@ -45,8 +45,7 @@ class TestTerminalRequirements:
lambda: {"env_type": "modal", "modal_mode": "managed"},
)
monkeypatch.setattr(
terminal_tool_module,
"is_managed_tool_gateway_ready",
"tools.terminal_tool_backends.is_managed_tool_gateway_ready",
lambda _vendor: True,
)
tools = get_tool_definitions(enabled_toolsets=["terminal", "code_execution"], quiet_mode=True)
tests/tools/test_threaded_process_handle.py
+1 -1
View File
@@ -3,7 +3,7 @@
import threading
import time
from tools.environments.base import _ThreadedProcessHandle
from tools.environments.base_output import _ThreadedProcessHandle
class TestBasicExecution:
tests/tools/test_watch_patterns.py
+1 -1
View File
@@ -312,7 +312,7 @@ class TestTerminalToolSchema:
class TestCodeExecutionBlocked:
def test_watch_patterns_blocked(self):
from tools.code_execution_tool import _TERMINAL_BLOCKED_PARAMS
from tools.code_execution_rpc import _TERMINAL_BLOCKED_PARAMS
assert "watch_patterns" in _TERMINAL_BLOCKED_PARAMS
tools/code_execution_env.py
+1 -1
View File
@@ -131,7 +131,7 @@ def _build_child_env(*, rpc_endpoint: str, rpc_token: str, tmpdir: str,
# (PYTHONPATH is in _SAFE_ENV_PREFIXES so it passes the scrub). They are redundant for same-Hermes-
# environment children and may be incompatible with external interpreters (project mode can select a
# different venv), so they must not shadow or poison the child's sys.path (#74817).
from tools.environments.local import _strip_hermes_owned_pythonpath
from tools.environments.local_pythonpath import _strip_hermes_owned_pythonpath
_strip_hermes_owned_pythonpath(child_env)
_existing_pp = child_env.get("PYTHONPATH", "")
_pp_parts = [tmpdir]
tools/code_execution_tool.py
+6 -10
View File
@@ -28,14 +28,8 @@ from typing import Any, Dict, List, Optional, Tuple
from tools.thread_context import propagate_context_to_thread
from tools.registry import registry, tool_error
# Sibling-module symbols re-exported so `from tools.code_execution_tool import X` / patch() keep working.
from tools.code_execution_env import ( # noqa: F401
_SAFE_ENV_PREFIXES, _SECRET_SUBSTRINGS, _HERMES_CHILD_ALLOWED, _WINDOWS_ESSENTIAL_ENV_VARS,
_scrub_child_env, _build_child_env, _PROBE_CACHE_MAX, _usable_python_cache, _python_prefix_cache,
_external_env_logged, _cache_probe_result, _is_usable_python, _probe_python,
_python_environment_prefix, _uses_hermes_python_environment, _resolve_child_python, _resolve_child_cwd,
)
from tools.code_execution_rpc import _TERMINAL_BLOCKED_PARAMS, _rpc_server_loop, _rpc_poll_loop # noqa: F401
from tools.code_execution_env import _resolve_child_cwd, _resolve_child_python
from tools.code_execution_rpc import _rpc_poll_loop
logger = logging.getLogger(__name__)
@@ -110,7 +104,8 @@ def check_sandbox_requirements() -> bool:
if not SANDBOX_AVAILABLE:
return False
try:
from tools.terminal_tool import _check_vercel_sandbox_requirements, _get_env_config
from tools.terminal_tool import _get_env_config
from tools.terminal_tool_backends import _check_vercel_sandbox_requirements
config = _get_env_config()
except Exception:
logger.debug("Could not resolve terminal config for execute_code availability", exc_info=True)
@@ -409,8 +404,9 @@ _CONTAINER_CONFIG_DEFAULTS = (
def _get_or_create_env(task_id: str):
"""``(env, env_type)`` — the environment the terminal/file tools share for *task_id*, created on
first use (same double-checked per-task lock pattern as file_tools._get_file_ops)."""
from tools.terminal_tool_backends import _create_environment
from tools.terminal_tool import (
_active_environments, _env_lock, _create_environment, _get_env_config, _last_activity,
_active_environments, _env_lock, _get_env_config, _last_activity,
_start_cleanup_thread, _creation_locks, _creation_locks_lock, _task_env_overrides,
_resolve_container_task_id, _resolve_task_host_cwd, _is_container_backend, _select_image,
_ssh_config_from_config,
tools/code_kernel.py
+6 -4
View File
@@ -256,7 +256,7 @@ class CellAuthority:
def dispatch(self, tool_name: str, tool_args: dict) -> str:
"""Run one tool call under THIS cell's context and callbacks."""
from tools.code_execution_tool import tool_error
from tools.registry import tool_error
if not self.active:
return tool_error("No active execute_code cell: the cell this kernel call "
"belonged to has settled, so its tool authority is retired.")
@@ -415,7 +415,7 @@ def _resolve_owner(task_id: str) -> str:
(verified live, both directions). Children get their own kernels keyed by delegation session id.
"""
try:
from tools.approval import get_current_session_key
from tools.approval_context import get_current_session_key
session_key = get_current_session_key(default="")
except Exception:
session_key = ""
@@ -455,7 +455,8 @@ def _rpc_forever(kernel: SessionKernel, max_tool_calls: int,
its 300s idle timeout, and a kernel idles longer between cells, so re-accept until teardown
(the client stub reconnects: HERMES_RPC_PERSISTENT). The serving thread carries NO frozen
authority — every dispatch routes through the CURRENT cell's ``CellAuthority``."""
from tools.code_execution_tool import _rpc_server_loop, tool_error
from tools.code_execution_rpc import _rpc_server_loop
from tools.registry import tool_error
def _dispatch(tool_name: str, tool_args: dict) -> str:
authority = kernel.cell_authority
if authority is None:
@@ -580,7 +581,8 @@ def _parent_process_handle(child_env: Dict[str, str]):
def _spawn(kernel: SessionKernel, *, child_python: str, child_cwd: str,
sandbox_tools: frozenset, max_tool_calls: int, task_id: str = "") -> None:
from tools.code_execution_tool import _build_child_env, generate_hermes_tools_module
from tools.code_execution_env import _build_child_env
from tools.code_execution_tool import generate_hermes_tools_module
kernel.tmpdir = tempfile.mkdtemp(prefix="hermes_kernel_")
kernel.rpc_token = secrets.token_urlsafe(32)
kernel.sentinel = "@@HERMES-KERNEL-" + secrets.token_urlsafe(16) + "@@"
tools/env_passthrough.py
+1 -1
View File
@@ -41,7 +41,7 @@ def _is_hermes_provider_credential(name: str) -> bool:
in the ``execute_code`` child); non-Hermes keys (TENOR_API_KEY, …) stay
registerable. Fails closed when the blocklist cannot be imported."""
try:
from tools.environments.local import (
from tools.environments.local_env_policy import (
_HERMES_PROVIDER_ENV_BLOCKLIST, _is_hermes_internal_secret)
except Exception as e:
logger.warning(
tools/environments/base.py
+7 -17
View File
@@ -4,8 +4,7 @@ Unified spawn-per-call model: every command spawns a fresh ``bash -c`` process.
A session snapshot (env vars, functions, aliases) is captured once at init and
re-sourced before each command. CWD persists via in-band stdout markers (remote)
or a temp file (local). Cohesive pieces live in sibling modules (``base_output``,
``base_session_env``, ``base_wait``, ``path_utils``) and are re-exported here so
``from tools.environments.base import X`` / ``patch("tools.environments.base.X")`` keep working.
``base_session_env``, ``base_wait``, ``path_utils``).
"""
import json
@@ -21,19 +20,14 @@ from typing import Callable, Iterable
from hermes_constants import get_hermes_home
from tools.interrupt import is_interrupted, is_thread_interrupted
from tools.environments.base_output import ( # noqa: F401
ProcessHandle, _BoundedOutputCollector, _ThreadedProcessHandle, _UNBOUNDED_CAPTURE_CHARS,
_finalize_wait_result, _new_output_collector, _pipe_stdin, _popen_bash, _start_drain_thread,
from tools.environments.base_output import (
ProcessHandle, _finalize_wait_result, _new_output_collector, _start_drain_thread,
)
from tools.environments.base_session_env import ( # noqa: F401
_SHELL_ENV_NAME_RE, _SNAP_TMP, _SNAP_TMP_SUFFIX, _SNAPSHOT_EXCLUDED_ENV_REGEX, _cwd_marker,
_export_dump_excluding_session_vars, _snapshot_bootstrap_script, _split_cwd_marker,
from tools.environments.base_session_env import (
_SHELL_ENV_NAME_RE, _SNAP_TMP_SUFFIX, _cwd_marker, _snapshot_bootstrap_script, _split_cwd_marker,
_wrap_command_script,
)
from tools.environments.base_wait import _WaitTrace
from tools.environments.path_utils import ( # noqa: F401
_SANDBOX_DIR_HASH_LEN, _SANDBOX_DIR_MAX_LEN, _SANDBOX_DIR_UNSAFE_RE, sanitize_task_id_for_path,
)
logger = logging.getLogger(__name__)
@@ -440,7 +434,7 @@ class BaseEnvironment(ABC):
# Guard against the `A && B &` subshell-wait trap by default; callers
# that already produce shell-safe wrappers (spawn_via_env) pass False.
if rewrite_compound_background:
from tools.terminal_tool import _rewrite_compound_background
from tools.terminal_tool_sudo import _rewrite_compound_background
exec_command = _rewrite_compound_background(exec_command)
effective_timeout = timeout or self.timeout
effective_cwd = cwd or self.cwd
@@ -518,10 +512,6 @@ class BaseEnvironment(ABC):
logger.debug("terminal wait-bound kill_process_tree failed", exc_info=True)
# --- Shared helpers ---
def stop(self):
"""Alias for cleanup (compat with older callers)."""
self.cleanup()
def __del__(self):
try:
self.cleanup()
@@ -530,5 +520,5 @@ class BaseEnvironment(ABC):
def _prepare_command(self, command: str) -> tuple[str, str | None]:
"""Transform sudo commands if SUDO_PASSWORD is available."""
from tools.terminal_tool import _transform_sudo_command
from tools.terminal_tool_sudo import _transform_sudo_command
return _transform_sudo_command(command)
tools/environments/daytona.py
+2 -1
View File
@@ -12,7 +12,8 @@ import shlex
import threading
from pathlib import Path
from tools.environments.base import BaseEnvironment, _ThreadedProcessHandle
from tools.environments.base import BaseEnvironment
from tools.environments.base_output import _ThreadedProcessHandle
from tools.environments.file_sync import (
FileSyncManager, iter_sync_files, quoted_mkdir_command, quoted_rm_command, unique_parent_dirs)
from tools.environments.remote_common import ensure_lazy_dep
tools/environments/docker.py
+6 -5
View File
@@ -20,15 +20,16 @@ import uuid
from pathlib import Path
from typing import Optional
from tools.environments.base import BaseEnvironment, EnvironmentConnectionError, _SHELL_ENV_NAME_RE, _popen_bash
from tools.environments.docker_egress import ( # noqa: F401 — re-exported for tests/patch targets
from tools.environments.base import BaseEnvironment, EnvironmentConnectionError, _SHELL_ENV_NAME_RE
from tools.environments.base_output import _popen_bash
from tools.environments.docker_egress import (
_EGRESS_LABEL_KEY, _critical_egress_env_names, _egress_enforce_on_docker, _egress_proxy_args_for_docker,
_egress_reuse_fingerprint, _extra_args_egress_collisions, check_docker_env_collisions,
check_extra_args_collisions, check_forward_env_collisions, merge_egress_env,
_egress_reuse_fingerprint, check_docker_env_collisions, check_extra_args_collisions,
check_forward_env_collisions, merge_egress_env,
)
from tools.environments.path_utils import sanitize_task_id_for_path
from tools.environments.remote_common import bash_argv, run_capture
from tools.environments.local import _HERMES_PROVIDER_ENV_BLOCKLIST, _is_hermes_internal_secret
from tools.environments.local_env_policy import _HERMES_PROVIDER_ENV_BLOCKLIST, _is_hermes_internal_secret
logger = logging.getLogger(__name__)
tools/environments/local.py
+11 -17
View File
@@ -17,25 +17,19 @@ from collections.abc import Mapping
from pathlib import Path
from hermes_constants import get_process_hermes_home
from tools.environments.base import BaseEnvironment, _pipe_stdin
from tools.environments.base import BaseEnvironment
from tools.environments.base_output import _pipe_stdin
from hermes_cli._subprocess_compat import windows_hide_flags
# Re-exported so ``from tools.environments.local import X`` and
# ``patch("tools.environments.local.X")`` keep working after the split.
from tools.environments.local_env_policy import ( # noqa: F401
_ACTIVE_VENV_MARKER_VARS, _ALWAYS_STRIP_KEYS, _AWS_SDK_CREDENTIAL_ENV_VARS,
_HERMES_PROVIDER_ENV_BLOCKLIST, _HERMES_PROVIDER_ENV_FORCE_PREFIX,
_TERMINAL_FIRST_PARTY_ENV_PREFIXES, _build_provider_env_blocklist,
_buzz_terminal_context_active, _is_hermes_internal_secret,
_is_terminal_first_party_env, _matches_terminal_first_party_prefix,
_plugin_terminal_env_strip_keys)
from tools.environments.local_gitbash_probe import ( # noqa: F401
_BASH_EXTERNAL_PROGRAM_PROBE, _bash_probe_details_cache, _bash_starts,
_bash_starts_cache, _git_bash_aslr_help, _git_root_from_bash,
from tools.environments.local_env_policy import (
_ALWAYS_STRIP_KEYS, _HERMES_PROVIDER_ENV_BLOCKLIST, _HERMES_PROVIDER_ENV_FORCE_PREFIX,
_is_hermes_internal_secret, _is_terminal_first_party_env,
_matches_terminal_first_party_prefix, _plugin_terminal_env_strip_keys)
from tools.environments.local_gitbash_probe import (
_bash_probe_details_cache, _bash_starts, _git_bash_aslr_help,
_looks_like_msys_spawn_failure, _mandatory_aslr_enabled)
from tools.environments.local_pythonpath import ( # noqa: F401
_build_hermes_repo_root_aliases, _get_hermes_site_packages, _same_path,
_strip_hermes_owned_pythonpath, _strip_hermes_owned_pythonpath_and_runtime_markers,
_validated_runtime_venv)
from tools.environments.local_pythonpath import (
_build_hermes_repo_root_aliases, _strip_hermes_owned_pythonpath_and_runtime_markers)
_IS_WINDOWS = platform.system() == "Windows"
tools/environments/modal.py
+2 -1
View File
@@ -13,7 +13,8 @@ from pathlib import Path
from typing import Any, Optional
from hermes_constants import get_hermes_home
from tools.environments.base import BaseEnvironment, _ThreadedProcessHandle, _load_json_store, _save_json_store
from tools.environments.base import BaseEnvironment, _load_json_store, _save_json_store
from tools.environments.base_output import _ThreadedProcessHandle
from tools.environments.file_sync import (
FileSyncManager, iter_sync_files, quoted_mkdir_command, quoted_rm_command, unique_parent_dirs)
from tools.environments.remote_common import bash_argv, ensure_lazy_dep
tools/environments/singularity.py
+2 -1
View File
@@ -14,7 +14,8 @@ from pathlib import Path
from typing import Optional
from hermes_constants import get_hermes_home
from tools.environments.base import BaseEnvironment, _load_json_store, _popen_bash, _save_json_store
from tools.environments.base import BaseEnvironment, _load_json_store, _save_json_store
from tools.environments.base_output import _popen_bash
from tools.environments.path_utils import sanitize_task_id_for_path
from tools.environments.remote_common import bash_argv, run_capture
tools/environments/ssh.py
+2 -1
View File
@@ -10,7 +10,8 @@ import subprocess
import tempfile
from pathlib import Path
from tools.environments.base import BaseEnvironment, EnvironmentConnectionError, _popen_bash
from tools.environments.base import BaseEnvironment, EnvironmentConnectionError
from tools.environments.base_output import _popen_bash
from tools.environments.file_sync import (
FileSyncManager, iter_sync_files, quoted_mkdir_command, quoted_rm_command, unique_parent_dirs)
from tools.environments.remote_common import bash_argv, run_capture
tools/environments/vercel_sandbox.py
+2 -1
View File
@@ -22,7 +22,8 @@ from typing import TYPE_CHECKING, Any
import httpx
from hermes_constants import get_hermes_home
from tools.environments.base import BaseEnvironment, _ThreadedProcessHandle, _load_json_store, _save_json_store
from tools.environments.base import BaseEnvironment, _load_json_store, _save_json_store
from tools.environments.base_output import _ThreadedProcessHandle
from tools.environments.file_sync import FileSyncManager, iter_sync_files, quoted_rm_command
from tools.environments.remote_common import ensure_lazy_dep
tools/image_generation_tool.py
+1 -1
View File
@@ -289,7 +289,7 @@ def _looks_like_absolute_file_path(value: str) -> bool:
def _active_terminal_env(task_id: str | None):
try:
from tools.terminal_tool import get_active_env
from tools.terminal_tool_lifecycle import get_active_env
return get_active_env(task_id or "default")
except Exception as exc: # noqa: BLE001 - artifact hinting must not break generation
logger.debug("Could not inspect active terminal environment: %s", exc)
tools/image_source.py
+1 -1
View File
@@ -187,7 +187,7 @@ def _get_active_env(task_id: Optional[str]):
if not task_id:
return None
try:
from tools.terminal_tool import get_active_env
from tools.terminal_tool_lifecycle import get_active_env
return get_active_env(task_id)
except Exception:
return None
tools/process_registry.py
+1 -1
View File
@@ -823,7 +823,7 @@ class ProcessRegistry:
# pipe open forever when B is a long-running server. The rewriter turns it into
# ``A && { B & }``. Lazy import: terminal_tool imports this module.
# Guard against the `A && B &` subshell-wait trap (issue #68915).
from tools.terminal_tool import _rewrite_compound_background as _rewrite_bg
from tools.terminal_tool_sudo import _rewrite_compound_background as _rewrite_bg
safe_command = _rewrite_bg(command)
session = self._new_session(command, task_id, owner_task_id, session_key, _resolve_safe_cwd(cwd or os.getcwd()))
tools/terminal_tool.py
+12 -36
View File
@@ -27,8 +27,6 @@ import atexit
from dataclasses import dataclass
from typing import Optional, Dict, Any, List
from utils import env_var_enabled # noqa: F401 (terminal_tool_result imports it lazily via origin)
logger = logging.getLogger(__name__)
@@ -39,36 +37,20 @@ def _redact_terminal_error_text(value: Any) -> str:
return redact_sensitive_text("" if value is None else str(value), force=True)
# Interrupt event set by the agent on user interrupt; executors poll it to
# kill long-running subprocesses instead of blocking until timeout.
from tools.interrupt import _interrupt_event # noqa: F401 — re-exported (tests patch it here)
from tools.registry import tool_error
from tools.terminal_tool_lifecycle import ( # noqa: F401 (re-exported; tests patch tools.terminal_tool.<name>)
_check_disk_usage_warning, _cleanup_env, _cleanup_inactive_envs, _create_configured_env,
_evict_environment_for_task, cleanup_all_environments, cleanup_vm, ensure_task_env,
get_active_env, is_persistent_env,
from tools.terminal_tool_lifecycle import (
_check_disk_usage_warning, _cleanup_inactive_envs, _create_configured_env,
_evict_environment_for_task, cleanup_all_environments, ensure_task_env,
)
from tools.terminal_tool_config import ( # noqa: F401 (re-exported; tests patch tools.terminal_tool.<name>)
_CONTAINER_BACKENDS, _HOST_CWD_PREFIXES, _get_plugin_env_provider, _is_container_backend,
_is_unusable_container_cwd, _parse_env_var, _plugin_env_flag, _quiet, _safe_getcwd, _tenv,
_tenv_bool,
from tools.terminal_tool_config import (
_HOST_CWD_PREFIXES, _is_container_backend, _is_unusable_container_cwd, _parse_env_var,
_plugin_env_flag, _quiet, _safe_getcwd, _tenv, _tenv_bool,
)
from tools.terminal_tool_backends import ( # noqa: F401 (re-exported; tests patch tools.terminal_tool.<name>)
_REQUIREMENT_CHECKERS, _SUPPORTED_VERCEL_RUNTIMES, _VERCEL_SANDBOX_DEFAULT_CWD,
_check_plugin_requirements, _check_vercel_sandbox_requirements, _create_environment,
)
from tools.terminal_tool_sudo import ( # noqa: F401 (re-exported; tests patch tools.terminal_tool.<name>)
_count_real_sudo_invocations, _get_cached_sudo_password, _handle_sudo_failure,
_in_delegated_child_context, _invalidate_cached_sudo_on_auth_failure,
_prompt_for_sudo_password, _reset_cached_sudo_passwords, _rewrite_compound_background,
_set_cached_sudo_password, _sudo_nopasswd_works, _sudo_wrong_password_failure,
_transform_sudo_command,
from tools.terminal_tool_backends import (
_REQUIREMENT_CHECKERS, _VERCEL_SANDBOX_DEFAULT_CWD, _check_plugin_requirements,
)
# display_hermes_home imported lazily at call site (stale-module safety during hermes update)
from tools.tool_backend_helpers import ( # noqa: F401 (managed_nous_tools_enabled: test patch target)
coerce_modal_mode,
managed_nous_tools_enabled,
)
from tools.tool_backend_helpers import coerce_modal_mode, managed_nous_tools_enabled
def _safe_parse_import_env(name: str, default: Any, converter, type_label: str):
@@ -169,10 +151,6 @@ def _check_all_guards(command: str, env_type: str,
from tools.environments.base import EnvironmentConnectionError
# Resolved lazily by terminal_tool_backends through this module so tests can
# monkeypatch ``tools.terminal_tool._DockerEnvironment`` / the gateway probes.
from tools.environments.docker import DockerEnvironment as _DockerEnvironment # noqa: F401
from tools.managed_tool_gateway import is_managed_tool_gateway_ready # noqa: F401
# Tool description for LLM
@@ -748,14 +726,12 @@ def _command_requires_pipe_stdin(command: str) -> bool:
return normalized.startswith("gh auth login") and "--with-token" in normalized
from tools.terminal_tool_guards import ( # noqa: F401 — re-exported (tests, plugins)
_foreground_background_guidance, _safe_command_preview, _strip_quotes, _validate_workdir,
from tools.terminal_tool_guards import (
_foreground_background_guidance, _safe_command_preview, _validate_workdir,
gateway_lifecycle_block, self_repo_block,
)
from tools.terminal_tool_background import spawn_background_process
from tools.terminal_tool_result import ( # noqa: F401 — re-exported (tests)
_interpret_exit_code, _interpret_signal_exit, finalize_foreground_result,
)
from tools.terminal_tool_result import finalize_foreground_result
def _resolve_notification_flag_conflict(*, notify_on_complete: bool, watch_patterns, background: bool) -> tuple:
tools/terminal_tool_backends.py
+8 -9
View File
@@ -1,6 +1,6 @@
"""Execution-environment backends for the terminal tool: per-backend builders, config-to-kwargs
shapers, and requirement checkers, routed by dispatch/spec tables. Split out of ``tools/terminal_tool.py``;
every public/patched name is re-imported there, so ``tools.terminal_tool.<name>`` keeps resolving."""
the terminal tool imports the builders/checkers it uses from here."""
import functools
import importlib.util
@@ -10,13 +10,16 @@ import shutil
import subprocess
from typing import Any, Dict, Optional
from tools.environments.docker import DockerEnvironment as _DockerEnvironment
from tools.environments.local import LocalEnvironment as _LocalEnvironment
from tools.environments.managed_modal import ManagedModalEnvironment as _ManagedModalEnvironment
from tools.environments.modal import ModalEnvironment as _ModalEnvironment
from tools.environments.singularity import SingularityEnvironment as _SingularityEnvironment
from tools.environments.ssh import SSHEnvironment as _SSHEnvironment
from tools.tool_backend_helpers import (has_direct_modal_credentials, nous_tool_gateway_unavailable_message,
resolve_modal_backend_state)
from tools.managed_tool_gateway import is_managed_tool_gateway_ready
from tools.terminal_tool_config import _get_plugin_env_provider
from tools.tool_backend_helpers import (has_direct_modal_credentials, managed_nous_tools_enabled,
nous_tool_gateway_unavailable_message, resolve_modal_backend_state)
# Log-record parity with the origin module.
logger = logging.getLogger("tools.terminal_tool")
@@ -69,7 +72,6 @@ def _is_supported_vercel_runtime(runtime: str) -> bool:
def _get_modal_backend_state(modal_mode: object | None) -> Dict[str, Any]:
"""Resolve direct vs managed Modal backend selection."""
from tools.terminal_tool import is_managed_tool_gateway_ready
return resolve_modal_backend_state(modal_mode, has_direct=has_direct_modal_credentials(),
managed_ready=is_managed_tool_gateway_ready("modal"))
@@ -77,7 +79,6 @@ def _get_modal_backend_state(modal_mode: object | None) -> Dict[str, Any]:
def _modal_unavailable_reason(modal_state: Dict[str, Any]) -> tuple[str, str]:
"""(log message, ValueError message) for a modal_state with no selected backend.
Single decision shared by the requirements checker and the env builder."""
from tools.terminal_tool import managed_nous_tools_enabled
gateway = nous_tool_gateway_unavailable_message("managed Modal execution")
if modal_state["managed_mode_blocked"] or modal_state["mode"] == "managed":
tail = (("Nous Tool Gateway access is not currently available and no direct Modal credentials/config "
@@ -103,8 +104,8 @@ def _build_local_env(*, cwd, timeout, **_):
def _build_docker_env(*, image, cwd, timeout, cc, task_id, host_cwd, **_):
from tools.terminal_tool import (_DockerEnvironment, _docker_session_isolation_enabled,
_has_isolation_overrides, _maybe_reap_docker_orphans)
from tools.terminal_tool import (_docker_session_isolation_enabled, _has_isolation_overrides,
_maybe_reap_docker_orphans)
# One-shot reaper for labeled containers orphaned by prior Hermes processes that died before
# atexit (SIGKILL / OOM / closed terminal); ``terminal.docker_orphan_reaper: false`` disables it.
_maybe_reap_docker_orphans(cc)
@@ -179,7 +180,6 @@ def _build_ssh_env(*, cwd, timeout, ssh_config, **_):
def _build_plugin_env(*, env_type, image, cwd, timeout, cc, task_id, **_):
from tools.terminal_tool import _get_plugin_env_provider
provider = _get_plugin_env_provider(env_type)
if provider is not None:
env_obj = provider.create_environment(cwd=cwd, timeout=timeout, task_id=task_id, image=image,
@@ -309,7 +309,6 @@ def _check_requirements(env_type: str, config: Dict[str, Any]) -> bool:
def _check_plugin_requirements(config: Dict[str, Any]) -> bool:
from tools.terminal_tool import _get_plugin_env_provider
env_type = config["env_type"]
provider = _get_plugin_env_provider(env_type)
if provider is not None:
tools/terminal_tool_lifecycle.py
+4 -5
View File
@@ -69,7 +69,8 @@ def _create_configured_env(
):
"""``_create_environment`` with the ssh/container kwargs shaped from *config*
(shared by the terminal tool and the lazy :func:`ensure_task_env` bring-up)."""
from tools.terminal_tool import _create_environment, _is_container_backend
from tools.terminal_tool_backends import _create_environment
from tools.terminal_tool_config import _is_container_backend
return _create_environment(
env_type=env_type, image=image, cwd=cwd, timeout=timeout,
ssh_config=_ssh_config_from_config(config) if env_type == "ssh" else None,
@@ -197,8 +198,7 @@ def ensure_task_env(task_id: Optional[str] = None):
from tools.terminal_tool import (
_active_environments, _creation_locks, _creation_locks_lock, _env_lock,
_get_env_config, _last_activity, _resolve_container_task_id,
_resolve_task_host_cwd, _select_image, _start_cleanup_thread, get_active_env,
resolve_task_overrides,
_resolve_task_host_cwd, _select_image, _start_cleanup_thread, resolve_task_overrides,
)
config = _get_env_config()
env_type = config["env_type"]
@@ -256,7 +256,6 @@ def is_persistent_env(task_id: str) -> bool:
docker containers count as persistent HERE: their lifetime is the session
(removed by ``AIAgent.close()`` → ``cleanup_vm`` and the idle reaper).
"""
from tools.terminal_tool import get_active_env
env = get_active_env(task_id)
if env is None:
return False
@@ -265,7 +264,7 @@ def is_persistent_env(task_id: str) -> bool:
def cleanup_all_environments():
"""Clean up ALL active environments. Use with caution."""
from tools.terminal_tool import _active_environments, cleanup_vm
from tools.terminal_tool import _active_environments
cleaned = 0
for task_id in list(_active_environments.keys()):
try:
tools/terminal_tool_result.py
+10 -5
View File
@@ -104,12 +104,17 @@ def _interpret_exit_code(command: str, exit_code: int) -> str | None:
def _sudo_annotations(command: str, output: str, env_type: str) -> tuple[str, bool, bool]:
"""Sudo failure handling -> (output, auth_failed, cache_cleared)."""
import tools.terminal_tool as tt
output = tt._handle_sudo_failure(output, env_type)
auth_failed = tt._sudo_wrong_password_failure(output)
cache_cleared = tt._invalidate_cached_sudo_on_auth_failure(command, output)
from tools.terminal_tool_sudo import (
_handle_sudo_failure, _in_delegated_child_context, _invalidate_cached_sudo_on_auth_failure,
_sudo_wrong_password_failure,
)
from utils import env_var_enabled
output = _handle_sudo_failure(output, env_type)
auth_failed = _sudo_wrong_password_failure(output)
cache_cleared = _invalidate_cached_sudo_on_auth_failure(command, output)
can_reprompt = cache_cleared and (
tt._get_sudo_password_callback() is not None or tt.env_var_enabled("HERMES_INTERACTIVE")
) and not tt._in_delegated_child_context()
tt._get_sudo_password_callback() is not None or env_var_enabled("HERMES_INTERACTIVE")
) and not _in_delegated_child_context()
if can_reprompt:
output += ("\n\n⚠️ Sudo authentication failed — cached password "
"cleared. You will be prompted again on the next sudo command.")
tools/terminal_tool_sudo.py
+1 -2
View File
@@ -116,7 +116,6 @@ def _sudo_wrong_password_failure(output: str) -> bool:
def _invalidate_cached_sudo_on_auth_failure(command: str | None, output: str) -> bool:
"""Drop a session-cached sudo password after sudo rejects it. Env-configured
``SUDO_PASSWORD`` is left alone — an explicit operator choice, not a cache entry."""
from tools.terminal_tool import _count_real_sudo_invocations, _sudo_wrong_password_failure
if (
"SUDO_PASSWORD" in os.environ
or not _sudo_wrong_password_failure(output)
@@ -402,7 +401,7 @@ def _transform_sudo_command(command: str | None) -> tuple[str | None, str | None
returned unchanged and ``sudo_stdin`` is None, so it fails gracefully with "sudo: a password
is required". Password sources, in order: configured SUDO_PASSWORD, the session cache, then
an interactive prompt (45s timeout, cached on success) when a UI is reachable."""
from tools.terminal_tool import _get_sudo_password_callback, _prompt_for_sudo_password, _sudo_nopasswd_works
from tools.terminal_tool import _get_sudo_password_callback
if command is None:
return None, None
transformed, sudo_count = _rewrite_real_sudo_invocations(command)
tui_gateway/session_workdir.py
+1 -1
View File
@@ -523,7 +523,7 @@ def _set_session_cwd(session: dict, cwd: str) -> str:
# The synchronous DB write claims ordering authority; git probes may publish only for that exact generation.
_persist_session_cwd_and_schedule_git_meta(session, resolved)
with contextlib.suppress(Exception):
from tools.terminal_tool import cleanup_vm
from tools.terminal_tool_lifecycle import cleanup_vm
cleanup_vm(session["session_key"])
return resolved