diff --git a/acp_adapter/server.py b/acp_adapter/server.py index 3586768b67..083b1c45bc 100644 --- a/acp_adapter/server.py +++ b/acp_adapter/server.py @@ -37,7 +37,7 @@ from acp_adapter.session import SessionManager, SessionState, _expand_acp_enable from acp_adapter.tools import build_tool_complete, build_tool_start, coerce_tool_args from agent.context_compressor import (COMPRESSED_SUMMARY_METADATA_KEY, ContextCompressor) from agent.interrupt_compat import request_hard_interrupt -from tools.approval import (reset_hermes_interactive_context, set_hermes_interactive_context) +from tools.approval_context import reset_hermes_interactive_context, set_hermes_interactive_context logger = logging.getLogger(__name__) diff --git a/agent/tool_executor.py b/agent/tool_executor.py index 31e32c1cbb..fa04770c6b 100644 --- a/agent/tool_executor.py +++ b/agent/tool_executor.py @@ -127,7 +127,7 @@ def _authorization_gate_lock_timeout() -> float: once per gate (per batch), so a mid-process ``approvals.timeout`` change applies from the next batch. """ try: - from tools.approval import human_wait_ceiling + from tools.approval_human_wait import human_wait_ceiling # human_wait_ceiling is platform-safety-capped (agent/deadline.py MAX_SAFE_TIMEOUT_S): a huge # approvals.timeout can no longer overflow Lock.acquire's time_t on macOS (#83220). Deliberately NOT @@ -473,7 +473,7 @@ class _ConcurrentToolAuthorizationGate: # Snapshot on the SUBMITTING thread: excluded_seconds() is polled from the # batch wait loop, whose context may differ from the workers'. try: - from tools.approval import get_current_session_key + from tools.approval_context import get_current_session_key self._session_key = get_current_session_key() except Exception: @@ -486,7 +486,7 @@ class _ConcurrentToolAuthorizationGate: def _human_wait_seconds(self) -> float: try: - from tools.approval import human_wait_seconds + from tools.approval_human_wait import human_wait_seconds return human_wait_seconds(self._session_key) except Exception: diff --git a/gateway/hosted_room_execution_policy.py b/gateway/hosted_room_execution_policy.py index b425d850ff..29fbf66496 100644 --- a/gateway/hosted_room_execution_policy.py +++ b/gateway/hosted_room_execution_policy.py @@ -77,7 +77,8 @@ def execution_policy_mapping(*, target_profile: str, config: Mapping[str, Any] | raise RoomExecutionPolicyError("gateway config is invalid") from hermes_cli.config import resolve_turn_limit from hermes_cli.tools_config import _get_platform_tools - from tools.approval import _YOLO_MODE_FROZEN, _normalize_approval_mode + from tools.approval import _YOLO_MODE_FROZEN + from tools.approval_context import _normalize_approval_mode toolsets = sorted({*_get_platform_tools(dict(config), "api_server"), "bot_room"}) agent = config.get("agent") if isinstance(config.get("agent"), Mapping) else {} approvals = config.get("approvals") if isinstance(config.get("approvals"), Mapping) else {} diff --git a/gateway/platforms/api_server_runs.py b/gateway/platforms/api_server_runs.py index 2231b4f22a..a72a000c19 100644 --- a/gateway/platforms/api_server_runs.py +++ b/gateway/platforms/api_server_runs.py @@ -471,9 +471,8 @@ def _run_agent_sync(self, run: _RunLaunch, agent, approval_notify, *, _api_serve # a config with a few stdio servers — even for sessions that never run a worker-routed command. Sessions # held by a live transport are never reaped, so with the desktop app open for days those fleets # accumulate until the OS refuses new process spawns. - from tools.approval import ( - register_gateway_notify, reset_current_session_key, set_current_session_key, - unregister_gateway_notify) + from tools.approval import register_gateway_notify, unregister_gateway_notify + from tools.approval_context import reset_current_session_key, set_current_session_key session_id = run.session_id effective_task_id = session_id or run.run_id # (token, reset) pairs unwound in the finally block; bound only once each step succeeds. diff --git a/gateway/run_turn_runner.py b/gateway/run_turn_runner.py index c84f85650b..6449454b19 100644 --- a/gateway/run_turn_runner.py +++ b/gateway/run_turn_runner.py @@ -1413,7 +1413,8 @@ class TurnRunner: """Run the turn with the per-session gateway approval callback registered: dangerous-command approval blocks the agent thread (mirrors CLI input()); the callback bridges sync→async.""" from gateway.run import _wrap_current_message_with_observed_context - from tools.approval import register_gateway_notify, reset_current_session_key, set_current_session_key, unregister_gateway_notify + from tools.approval import register_gateway_notify, unregister_gateway_notify + from tools.approval_context import reset_current_session_key, set_current_session_key ctx = self._ctx session_key = ctx.session_key or "" token = set_current_session_key(session_key) diff --git a/gateway/slash_commands_goals.py b/gateway/slash_commands_goals.py index 9936af9605..a36cfff565 100644 --- a/gateway/slash_commands_goals.py +++ b/gateway/slash_commands_goals.py @@ -314,7 +314,7 @@ class GatewayGoalCommandsMixin: if error: return error snapshot = list(getattr(agent, "_session_messages", None) or []) - from tools.approval import reset_current_session_key, set_current_session_key + from tools.approval_context import reset_current_session_key, set_current_session_key def _dispatch(): token = set_current_session_key(quick_key) diff --git a/hermes_cli/approval_mode.py b/hermes_cli/approval_mode.py index 6f019bbd01..76399faf7a 100644 --- a/hermes_cli/approval_mode.py +++ b/hermes_cli/approval_mode.py @@ -26,7 +26,7 @@ class ApprovalModeResult: def _effective_mode() -> str: """Return the exact mode enforced by the terminal approval guard.""" - from tools.approval import _get_approval_mode + from tools.approval_context import _get_approval_mode return _get_approval_mode() diff --git a/hermes_cli/approvals_suggest.py b/hermes_cli/approvals_suggest.py index 29acffa183..430664636f 100644 --- a/hermes_cli/approvals_suggest.py +++ b/hermes_cli/approvals_suggest.py @@ -148,7 +148,7 @@ def scan_approval_history(db_path: Optional[Path] = None, days: int = 90) -> lis """``(command, dangerous_class_description)`` records for dangerous-classified terminal commands that actually executed (i.e. carried an implied user approval). """ - from tools.approval import detect_dangerous_command, detect_hardline_command + from tools.approval_detection import detect_dangerous_command, detect_hardline_command path = Path(db_path) if db_path else default_db_path() if not path.exists(): return [] @@ -180,7 +180,7 @@ def scan_approval_history(db_path: Optional[Path] = None, days: int = 90) -> lis def normalize_command(command: str) -> str: """Fold user/hermes home prefixes and collapse whitespace.""" - from tools.approval import _rewrite_resolved_hermes_home, _rewrite_resolved_user_home + from tools.approval_detection import _rewrite_resolved_hermes_home, _rewrite_resolved_user_home return " ".join(_rewrite_resolved_user_home(_rewrite_resolved_hermes_home(command)).split()) @@ -200,7 +200,7 @@ def derive_glob(normalized: str) -> Optional[str]: Returns None for compound commands (shell operators — the runtime allowlist matcher refuses those anyway) and for commands anchored on an unsafe root binary. """ - from tools.approval import _has_allowlist_shell_operator + from tools.approval_floors import _has_allowlist_shell_operator tokens = normalized.split() if _has_allowlist_shell_operator(normalized) or not tokens or _unsafe_root_binary(tokens[0]): return None diff --git a/hermes_cli/approvals_test.py b/hermes_cli/approvals_test.py index 495732ebb6..0aaafb6a74 100644 --- a/hermes_cli/approvals_test.py +++ b/hermes_cli/approvals_test.py @@ -33,6 +33,7 @@ def evaluate_command(command: str, env_type: str = "local") -> dict: de-obfuscated forms the detectors actually evaluated). """ import tools.approval as approval + from tools import approval_context, approval_detection, approval_floors # Sync config-persisted "always" patterns so the allowlist check below sees what the runtime # would see (load is read-only). try: @@ -40,7 +41,7 @@ def evaluate_command(command: str, env_type: str = "local") -> dict: except Exception: pass - variants = list(approval._command_detection_variants(command)) + variants = list(approval_detection._command_detection_variants(command)) def result(verdict: str, rule=None, detail: str = "") -> dict: return { @@ -58,7 +59,7 @@ def evaluate_command(command: str, env_type: str = "local") -> dict: ) # 2. Hardline blocklist — never bypassable, even under yolo. - is_hardline, hardline_desc = approval.detect_hardline_command(command) + is_hardline, hardline_desc = approval_detection.detect_hardline_command(command) if is_hardline: return result( "hardline-deny", rule=hardline_desc, @@ -67,12 +68,12 @@ def evaluate_command(command: str, env_type: str = "local") -> dict: ) # 3. Sudo stdin guard — unconditional, like the hardline floor. - is_sudo_guess, sudo_desc = approval._check_sudo_stdin_guard(command) + is_sudo_guess, sudo_desc = approval_detection._check_sudo_stdin_guard(command) if is_sudo_guess: return result("hardline-deny", rule=sudo_desc, detail="sudo stdin guard (unconditional block)") # 4. User-defined approvals.deny rules — fire before yolo/off. - deny_pattern = approval._match_user_deny_rule(command) + deny_pattern = approval_floors._match_user_deny_rule(command) if deny_pattern is not None: return result( "user-deny", rule=deny_pattern, @@ -83,7 +84,7 @@ def evaluate_command(command: str, env_type: str = "local") -> dict: # 5. Yolo / approvals.mode=off bypass. if (approval._YOLO_MODE_FROZEN or approval.is_current_session_yolo_enabled() - or approval._get_approval_mode() == "off"): + or approval_context._get_approval_mode() == "off"): return result( "allow", detail="approval bypass active (--yolo or approvals.mode: off); " @@ -91,11 +92,11 @@ def evaluate_command(command: str, env_type: str = "local") -> dict: ) # 6. Permanent command_allowlist. - if approval._command_matches_permanent_allowlist(command): + if approval_floors._command_matches_permanent_allowlist(command): return result("allow", detail="matches command_allowlist in config.yaml (permanently approved)") # 7. Dangerous-pattern detection → would prompt. - is_dangerous, pattern_key, description = approval.detect_dangerous_command(command) + is_dangerous, pattern_key, description = approval_detection.detect_dangerous_command(command) if is_dangerous: return result( "ask-approval", rule=description, diff --git a/hermes_cli/cli_chat_turn_mixin.py b/hermes_cli/cli_chat_turn_mixin.py index 8d706fd456..790005f40d 100644 --- a/hermes_cli/cli_chat_turn_mixin.py +++ b/hermes_cli/cli_chat_turn_mixin.py @@ -278,7 +278,7 @@ class CLIChatTurnMixin: # Bind the approval session key so ``is_current_session_yolo_enabled()`` resolves # against the same key ``/yolo`` toggles under (``enable_session_yolo(self.session_id)``). try: - from tools.approval import reset_current_session_key, set_current_session_key + from tools.approval_context import reset_current_session_key, set_current_session_key _approval_session_token = set_current_session_key(self.session_id or "default") except Exception: reset_current_session_key = None # type: ignore[assignment] diff --git a/hermes_cli/cli_session_mixin.py b/hermes_cli/cli_session_mixin.py index 45c0d738b8..e3400b5cb9 100644 --- a/hermes_cli/cli_session_mixin.py +++ b/hermes_cli/cli_session_mixin.py @@ -278,7 +278,8 @@ class CLISessionMixin: approval_label = None try: - from tools.approval import _get_approval_mode, is_approval_bypass_active_for_session + from tools.approval import is_approval_bypass_active_for_session + from tools.approval_context import _get_approval_mode approval_label = _get_approval_mode() if is_approval_bypass_active_for_session(getattr(self, "session_key", "") or ""): approval_label += " (YOLO bypass active)" @@ -340,7 +341,7 @@ class CLISessionMixin: if not sessions: return False - from hermes_cli.main import _relative_time + from hermes_cli.timefmt import relative_time as _relative_time _cli_visible_print() if reason == "history": diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 684f41a912..1cf03f9984 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -855,7 +855,7 @@ class PluginContext: if not all(c.isalnum() or c == "_" for c in key): raise ValueError(f"Plugin '{me}' auxiliary task key {key!r} " f"must contain only alphanumeric characters and underscores") - from hermes_cli.main import _AUX_TASKS as _BUILTIN_AUX_TASKS + from hermes_cli.main_provider_setup import _AUX_TASKS as _BUILTIN_AUX_TASKS if key in {k for k, _name, _desc in _BUILTIN_AUX_TASKS}: raise ValueError(f"Plugin '{me}' cannot register auxiliary task {key!r} — that key is reserved " f"for a built-in task. Pick a plugin-namespaced key (e.g. '{me}_{key}').") @@ -1021,7 +1021,7 @@ _SCOPED_PROVIDER_REGISTRARS: Tuple[Tuple[str, str, str, str, str, str, Dict[str, "agent.browser_provider:BrowserProvider", "browser provider", "Register an :class:`agent.browser_provider.BrowserProvider`; " "``provider.name`` is matched by ``browser.cloud_provider`` (consulted by " - "``tools.browser_tool._get_cloud_provider``).", {}), + "``tools.browser_tool_cloud._get_cloud_provider``).", {}), ("register_terminal_environment_provider", "terminal_environment_provider", "agent.terminal_env_registry", "agent.terminal_env_provider:TerminalEnvironmentProvider", "terminal environment provider", @@ -1834,9 +1834,8 @@ def _resolve_block_from_details( if details.action != "approve": return None try: - from tools.approval import ( - request_tool_approval, reset_current_observability_context, set_current_observability_context, - ) + from tools.approval import request_tool_approval + from tools.approval_context import reset_current_observability_context, set_current_observability_context approval_tokens = None with suppress(Exception): approval_tokens = set_current_observability_context( diff --git a/hermes_cli/web_server_profiles.py b/hermes_cli/web_server_profiles.py index 6b1cf02aab..fcc64f90a4 100644 --- a/hermes_cli/web_server_profiles.py +++ b/hermes_cli/web_server_profiles.py @@ -52,7 +52,6 @@ def _hermes_home_scope(path) -> Any: def _is_other_profile(profile: Optional[str]) -> bool: """True when ``profile`` names a profile other than this process's own.""" - from hermes_cli.web_server import _resolve_profile_dir if _is_current_profile(profile): return False try: @@ -67,7 +66,7 @@ def _approval_mode_of(config: Dict[str, Any]) -> str: broadcast comparison use in-memory documents: re-reading through the config cache after a save can serve the pre-save document when the replacement file collides on the (mtime_ns, size) cache key, suppressing the broadcast exactly when the mode changed.""" - from tools.approval import _normalize_approval_mode + from tools.approval_context import _normalize_approval_mode approvals = config.get("approvals") default_mode = (DEFAULT_CONFIG.get("approvals") or {}).get("mode", "manual") mode = approvals.get("mode", default_mode) if isinstance(approvals, dict) else default_mode @@ -165,7 +164,7 @@ def _write_profile_mcp_servers(profile_dir: Path, servers: List["MCPServerCreate Mirrors the per-server shape ``POST /api/mcp/servers`` builds, batched so the whole profile-create write is one config save. Returns the number of servers written. """ - from hermes_cli.web_server import load_config, save_config + from hermes_cli.config import load_config, save_config from hermes_cli.mcp_config import _save_bearer_auth_token written = 0 with _hermes_home_scope(profile_dir): @@ -239,7 +238,6 @@ def _config_profile_scope(profile: Optional[str]): contextvar, never the process-global skills-module attributes ``_profile_scope`` swaps (holding those across an ``await`` lets a concurrent request restore THIS request's dir on its ``finally``). None/""/"current" = no override.""" - from hermes_cli.web_server import _resolve_profile_dir if _is_current_profile(profile): yield None return diff --git a/model_tools.py b/model_tools.py index 52a18efa7c..50944003ed 100644 --- a/model_tools.py +++ b/model_tools.py @@ -736,7 +736,7 @@ def _pre_dispatch_guards(function_name: str, function_args: Dict[str, Any], skip def _approval_observability(ids: _CallIds): """Bind the approval observability context (turn/tool_call/session ids) for the block.""" try: - from tools.approval import reset_current_observability_context, set_current_observability_context + from tools.approval_context import reset_current_observability_context, set_current_observability_context tokens = set_current_observability_context(turn_id=ids.turn_id or "", tool_call_id=ids.tool_call_id or "", session_id=ids.session_id or "") except Exception: diff --git a/tests/acp/test_approval_isolation.py b/tests/acp/test_approval_isolation.py index df61356071..b45e96789a 100644 --- a/tests/acp/test_approval_isolation.py +++ b/tests/acp/test_approval_isolation.py @@ -16,6 +16,7 @@ Both fixed together by: import threading import pytest +from tools import approval_context @pytest.fixture(autouse=True) @@ -30,6 +31,7 @@ def _isolate_approval_state(monkeypatch): for reasons unrelated to the code under test. """ import tools.approval as _approval + from tools import approval_context monkeypatch.setattr(_approval, "_permanent_approved", set()) monkeypatch.setattr(_approval, "_session_approved", {}) @@ -38,7 +40,7 @@ def _isolate_approval_state(monkeypatch): # command before the callback is consulted (test-order dependent, since # load_config() caching decides which config file is in effect). Pin the # mode so the GHSA regression path is what actually runs. - monkeypatch.setattr(_approval, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual") class TestThreadLocalApprovalCallback: diff --git a/tests/agent/test_auxiliary_explicit_cancellation.py b/tests/agent/test_auxiliary_explicit_cancellation.py index 991738af8a..4fabf8f98f 100644 --- a/tests/agent/test_auxiliary_explicit_cancellation.py +++ b/tests/agent/test_auxiliary_explicit_cancellation.py @@ -552,11 +552,8 @@ def test_isolated_provider_worker_inherits_protection_and_progress_hook() -> Non def test_isolated_provider_worker_inherits_caller_contextvars() -> None: - from tools.approval import ( - get_current_session_key, - reset_current_session_key, - set_current_session_key, - ) + from tools.approval import get_current_session_key + from tools.approval_context import reset_current_session_key, set_current_session_key arbitrary = contextvars.ContextVar("isolated-provider-test", default="missing") arbitrary_token = arbitrary.set("caller-value") diff --git a/tests/cli/test_cli_status_command.py b/tests/cli/test_cli_status_command.py index ac1e4d42d8..1feaee104d 100644 --- a/tests/cli/test_cli_status_command.py +++ b/tests/cli/test_cli_status_command.py @@ -108,7 +108,7 @@ def test_show_session_status_includes_reasoning_approvals_context(): } with patch("hermes_constants.display_hermes_home", return_value="~/.hermes"), \ - patch("tools.approval._get_approval_mode", return_value="manual"), \ + patch("tools.approval_context._get_approval_mode", return_value="manual"), \ patch("tools.approval.is_approval_bypass_active_for_session", return_value=False): cli_obj._show_session_status() diff --git a/tests/cli/test_cli_yolo_resume_persistence.py b/tests/cli/test_cli_yolo_resume_persistence.py index ce771d4d8d..82b62d7e83 100644 --- a/tests/cli/test_cli_yolo_resume_persistence.py +++ b/tests/cli/test_cli_yolo_resume_persistence.py @@ -24,6 +24,7 @@ from unittest.mock import MagicMock, patch import pytest import tools.approval as approval_module +from tools import approval_context from cli import HermesCLI from hermes_state import SessionDB @@ -216,14 +217,14 @@ class TestEndToEndPersistAndRestore: HermesCLI._restore_session_yolo(cli_two, meta) assert approval_module.is_session_yolo_enabled(SESSION_ID) is True - token = approval_module.set_current_session_key(SESSION_ID) + token = approval_context.set_current_session_key(SESSION_ID) try: result = approval_module.check_all_command_guards( "rm -rf /tmp/scratch-xyzzy", "local", ) assert result["approved"] is True finally: - approval_module.reset_current_session_key(token) + approval_context.reset_current_session_key(token) def test_toggle_off_round_trip(self, db): """OFF must persist too — a resumed session must not resurrect a diff --git a/tests/cli/test_cli_yolo_toggle.py b/tests/cli/test_cli_yolo_toggle.py index 637ac7083f..4d2ca5146b 100644 --- a/tests/cli/test_cli_yolo_toggle.py +++ b/tests/cli/test_cli_yolo_toggle.py @@ -30,6 +30,7 @@ from unittest.mock import patch import pytest import tools.approval as approval_module +from tools import approval_context from cli import HermesCLI @@ -167,7 +168,7 @@ class TestToggleYoloEndToEnd: def test_toggle_yolo_bypasses_dangerous_command_check(self): stand_in = _make_stand_in() - token = approval_module.set_current_session_key(SESSION_KEY) + token = approval_context.set_current_session_key(SESSION_KEY) try: with patch("cli._cprint"): HermesCLI._toggle_yolo(stand_in) # YOLO ON @@ -179,7 +180,7 @@ class TestToggleYoloEndToEnd: f"YOLO toggle should auto-approve dangerous commands, got: {result}" ) finally: - approval_module.reset_current_session_key(token) + approval_context.reset_current_session_key(token) diff --git a/tests/cron/test_scheduler_cron_session_isolation.py b/tests/cron/test_scheduler_cron_session_isolation.py index 5ba43caa14..c386be4a16 100644 --- a/tests/cron/test_scheduler_cron_session_isolation.py +++ b/tests/cron/test_scheduler_cron_session_isolation.py @@ -19,6 +19,7 @@ from gateway.session_context import ( set_session_vars, ) from tools import approval as approval_module +from tools import approval_context class _DummySessionDB: @@ -92,8 +93,8 @@ def test_run_job_cron_execute_code_deny_does_not_pollute_later_gateway_execute_c """Cron deny stays scoped; a later gateway approval still reaches its user.""" monkeypatch.setenv("HERMES_MODEL", "test-model") monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") - monkeypatch.setattr(approval_module, "_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") monkeypatch.setattr("hermes_state_registry.acquire", _DummySessionDB) monkeypatch.setattr("run_agent.AIAgent", _FakeCronAgent) monkeypatch.setattr( @@ -140,7 +141,7 @@ def test_run_job_cron_execute_code_deny_does_not_pollute_later_gateway_execute_c monkeypatch.delenv("HERMES_CRON_SESSION") session_key = "cron-isolation-session" - key_token = approval_module.set_current_session_key(session_key) + key_token = approval_context.set_current_session_key(session_key) session_tokens = set_session_vars( platform="discord", chat_id="123", @@ -157,7 +158,7 @@ def test_run_job_cron_execute_code_deny_does_not_pollute_later_gateway_execute_c assert result.get("user_approved") is True finally: clear_session_vars(session_tokens) - approval_module.reset_current_session_key(key_token) + approval_context.reset_current_session_key(key_token) with approval_module._lock: approval_module._gateway_queues.pop(session_key, None) approval_module._gateway_notify_cbs.pop(session_key, None) diff --git a/tests/gateway/test_api_server_runs.py b/tests/gateway/test_api_server_runs.py index eb3c7a55d9..43fdbc4a26 100644 --- a/tests/gateway/test_api_server_runs.py +++ b/tests/gateway/test_api_server_runs.py @@ -28,6 +28,7 @@ from gateway.platforms.api_server import ( security_headers_middleware, ) from tools import approval as approval_mod +from tools import approval_gateway_wait # --------------------------------------------------------------------------- @@ -421,12 +422,12 @@ class TestRunEvents: assert auth_adapter._run_approval_sessions[attacker_run] == attacker_run assert auth_adapter._run_approval_sessions[victim_run] != auth_adapter._run_approval_sessions[attacker_run] - victim_entry = approval_mod._ApprovalEntry({ + victim_entry = approval_gateway_wait._ApprovalEntry({ "command": "bash -c victim-danger", "description": "victim approval", "pattern_keys": ["shell-c"], }) - attacker_entry = approval_mod._ApprovalEntry({ + attacker_entry = approval_gateway_wait._ApprovalEntry({ "command": "bash -c attacker-danger", "description": "attacker approval", "pattern_keys": ["shell-c"], @@ -652,7 +653,7 @@ class TestRunLifecycleSweep: assert isinstance(task, asyncio.Task) assert not task.done() - pending = approval_mod._ApprovalEntry({ + pending = approval_gateway_wait._ApprovalEntry({ "command": "bash -c long-running", "description": "approval after stream TTL", "pattern_keys": ["shell-c"], @@ -1466,7 +1467,7 @@ class TestHostedRoomRuns: self, auth_adapter ): run_id = "run-room-approval" - current = approval_mod._ApprovalEntry({ + current = approval_gateway_wait._ApprovalEntry({ "request_id": "approval-B", "command": "rm -rf build-B", }) diff --git a/tests/gateway/test_approve_deny_commands.py b/tests/gateway/test_approve_deny_commands.py index ed668b526d..fb9eb1ad0d 100644 --- a/tests/gateway/test_approve_deny_commands.py +++ b/tests/gateway/test_approve_deny_commands.py @@ -109,11 +109,8 @@ class TestBlockingGatewayApproval: def test_register_and_resolve_unblocks_entry(self): """resolve_gateway_approval signals the entry's event.""" - from tools.approval import ( - register_gateway_notify, unregister_gateway_notify, - resolve_gateway_approval, has_blocking_approval, - _ApprovalEntry, _gateway_queues, - ) + from tools.approval import register_gateway_notify, unregister_gateway_notify, resolve_gateway_approval, has_blocking_approval, _gateway_queues + from tools.approval_gateway_wait import _ApprovalEntry session_key = "test-session" register_gateway_notify(session_key, lambda d: None) @@ -140,10 +137,8 @@ class TestBlockingGatewayApproval: def test_resolve_single_pops_oldest_fifo(self): """resolve_gateway_approval without resolve_all resolves oldest first.""" - from tools.approval import ( - resolve_gateway_approval, - _ApprovalEntry, _gateway_queues, - ) + from tools.approval import resolve_gateway_approval, _gateway_queues + from tools.approval_gateway_wait import _ApprovalEntry session_key = "test-fifo" e1 = _ApprovalEntry({"command": "first"}) e2 = _ApprovalEntry({"command": "second"}) @@ -171,7 +166,8 @@ class TestApproveCommand: @pytest.mark.asyncio async def test_approve_all_resolves_multiple(self): """/approve all resolves all pending approvals.""" - from tools.approval import _ApprovalEntry, _gateway_queues + from tools.approval import _gateway_queues + from tools.approval_gateway_wait import _ApprovalEntry runner = _make_runner() source = _make_source() @@ -189,7 +185,8 @@ class TestApproveCommand: @pytest.mark.asyncio async def test_approve_all_session(self): """/approve all session resolves all with session scope.""" - from tools.approval import _ApprovalEntry, _gateway_queues + from tools.approval import _gateway_queues + from tools.approval_gateway_wait import _ApprovalEntry runner = _make_runner() source = _make_source() @@ -219,7 +216,8 @@ class TestDenyCommand: @pytest.mark.asyncio async def test_deny_with_reason_attaches_reason(self): """/deny attaches the reason to the resolved entry.""" - from tools.approval import _ApprovalEntry, _gateway_queues + from tools.approval import _gateway_queues + from tools.approval_gateway_wait import _ApprovalEntry runner = _make_runner() source = _make_source() @@ -238,7 +236,8 @@ class TestDenyCommand: @pytest.mark.asyncio async def test_deny_all_with_reason(self): """/deny all denies everything and relays one reason.""" - from tools.approval import _ApprovalEntry, _gateway_queues + from tools.approval import _gateway_queues + from tools.approval_gateway_wait import _ApprovalEntry runner = _make_runner() source = _make_source() @@ -269,7 +268,8 @@ class TestBareTextNoLongerApproves: @pytest.mark.asyncio async def test_yes_does_not_execute_pending_command(self): """Saying 'yes' must not trigger approval. Only /approve works.""" - from tools.approval import _ApprovalEntry, _gateway_queues + from tools.approval import _gateway_queues + from tools.approval_gateway_wait import _ApprovalEntry runner = _make_runner() source = _make_source() @@ -292,7 +292,7 @@ class TestBlockingApprovalE2E: @pytest.fixture(autouse=True) def _manual_approval_mode(self, monkeypatch): - monkeypatch.setattr("tools.approval._get_approval_mode", lambda: "manual") + monkeypatch.setattr("tools.approval_context._get_approval_mode", lambda: "manual") def setup_method(self): _clear_approval_state() @@ -305,7 +305,7 @@ class TestBlockingApprovalE2E: # approvals.mode=smart which may auto-approve/deny via aux LLM before # notify_cb runs (flaky on CI when the LLM is slow or unavailable). self._approval_mode_patch = patch( - "tools.approval._get_approval_mode", return_value="manual" + "tools.approval_context._get_approval_mode", return_value="manual" ) self._approval_mode_patch.start() @@ -324,14 +324,8 @@ class TestBlockingApprovalE2E: def test_blocking_approval_uses_canonical_timeout(self, approval_config, monkeypatch): """Gateway waits use approvals.timeout, without a second timeout knob.""" from tools import approval as approval_module - from tools.approval import ( - check_all_command_guards, - register_gateway_notify, - reset_current_session_key, - resolve_gateway_approval, - set_current_session_key, - unregister_gateway_notify, - ) + from tools.approval import check_all_command_guards, register_gateway_notify, resolve_gateway_approval, unregister_gateway_notify + from tools.approval_context import reset_current_session_key, set_current_session_key monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False) session_key = "e2e-timeout" @@ -346,7 +340,7 @@ class TestBlockingApprovalE2E: os.environ["HERMES_SESSION_KEY"] = session_key try: with patch( - "tools.approval._get_approval_config", + "tools.approval_context._get_approval_config", return_value=approval_config, ): result_holder[0] = check_all_command_guards( @@ -386,7 +380,7 @@ class TestBlockingApprovalE2E: def make_agent(idx, cmd): def run(): - from tools.approval import reset_current_session_key, set_current_session_key + from tools.approval_context import reset_current_session_key, set_current_session_key token = set_current_session_key(session_key) os.environ["HERMES_GATEWAY_SESSION"] = "1" @@ -484,7 +478,7 @@ class TestCrossSessionApprovalIsolation: @pytest.fixture(autouse=True) def _manual_approval_mode(self, monkeypatch): - monkeypatch.setattr("tools.approval._get_approval_mode", lambda: "manual") + monkeypatch.setattr("tools.approval_context._get_approval_mode", lambda: "manual") def setup_method(self): _clear_approval_state() @@ -495,11 +489,8 @@ class TestCrossSessionApprovalIsolation: def test_contextvar_wins_over_clobbered_environ(self): """get_current_session_key honors the contextvar, not stale env.""" - from tools.approval import ( - get_current_session_key, - reset_current_session_key, - set_current_session_key, - ) + from tools.approval import get_current_session_key + from tools.approval_context import reset_current_session_key, set_current_session_key # Simulate a concurrent session B having written process-global env # last (the "last writer wins" clobber that caused #24100). @@ -554,14 +545,8 @@ class TestCrossSessionApprovalIsolation: def test_approval_prompt_routes_to_originating_session(self): """A dangerous command in session A's worker thread notifies session A's callback, even though os.environ points at session B.""" - from tools.approval import ( - check_all_command_guards, - register_gateway_notify, - reset_current_session_key, - resolve_gateway_approval, - set_current_session_key, - unregister_gateway_notify, - ) + from tools.approval import check_all_command_guards, register_gateway_notify, resolve_gateway_approval, unregister_gateway_notify + from tools.approval_context import reset_current_session_key, set_current_session_key notified_a = [] notified_b = [] register_gateway_notify("session-A", lambda d: notified_a.append(d)) @@ -618,15 +603,8 @@ class TestCrossSessionApprovalIsolation: must land in its OWN gateway queue, and resolving one must not resolve the other. """ - from tools.approval import ( - _gateway_queues, - check_all_command_guards, - register_gateway_notify, - reset_current_session_key, - resolve_gateway_approval, - set_current_session_key, - unregister_gateway_notify, - ) + from tools.approval import _gateway_queues, check_all_command_guards, register_gateway_notify, resolve_gateway_approval, unregister_gateway_notify + from tools.approval_context import reset_current_session_key, set_current_session_key # No HERMES_SESSION_KEY in os.environ at all — pure contextvar routing. os.environ.pop("HERMES_SESSION_KEY", None) diff --git a/tests/gateway/test_hosted_room_execution_policy.py b/tests/gateway/test_hosted_room_execution_policy.py index 5a60ee822a..45f446208d 100644 --- a/tests/gateway/test_hosted_room_execution_policy.py +++ b/tests/gateway/test_hosted_room_execution_policy.py @@ -22,7 +22,7 @@ from gateway.hosted_room_peer import ( issue_room_grant, verify_room_grant, ) -from tools import approval +from tools import approval_context from tui_gateway.hosted_room_peer_http import PeerRunsHTTPError from tui_gateway.hosted_room_service import _RouteStatusPeerClient @@ -102,10 +102,10 @@ def test_unlimited_policy_survives_the_catalog_json_round_trip_exactly(): def test_room_policy_overrides_broader_live_approval_config(monkeypatch): policy = RoomExecutionPolicy.from_mapping(_policy(approval_mode="manual")) - monkeypatch.setattr(approval, "_get_approval_config", lambda: {"mode": "off"}) + monkeypatch.setattr(approval_context, "_get_approval_config", lambda: {"mode": "off"}) token = bind_room_execution_policy(policy) try: - assert approval._get_approval_mode() == "manual" + assert approval_context._get_approval_mode() == "manual" finally: reset_room_execution_policy(token) diff --git a/tests/gateway/test_plaintext_approval_routing.py b/tests/gateway/test_plaintext_approval_routing.py index 400c609178..bbe5e58af5 100644 --- a/tests/gateway/test_plaintext_approval_routing.py +++ b/tests/gateway/test_plaintext_approval_routing.py @@ -84,7 +84,8 @@ def _make_runner(): def _register_blocking_approval(runner): """Register a real blocking approval entry for the runner's session.""" - from tools.approval import _ApprovalEntry, _gateway_queues + from tools.approval import _gateway_queues + from tools.approval_gateway_wait import _ApprovalEntry source = _make_source() session_key = runner._session_key_for_source(source) entry = _ApprovalEntry({"command": "rm -rf /tmp/test"}) diff --git a/tests/gateway/test_session_boundary_security_state.py b/tests/gateway/test_session_boundary_security_state.py index 63067117b1..0db26f46d6 100644 --- a/tests/gateway/test_session_boundary_security_state.py +++ b/tests/gateway/test_session_boundary_security_state.py @@ -11,13 +11,8 @@ from gateway.platforms.base import MessageEvent from gateway.session import SessionEntry, SessionSource, build_session_key from tools import approval as approval_mod from tools import slash_confirm as slash_confirm_mod -from tools.approval import ( - _ApprovalEntry, - approve_session, - enable_session_yolo, - is_approved, - is_session_yolo_enabled, -) +from tools.approval import approve_session, enable_session_yolo, is_approved, is_session_yolo_enabled +from tools.approval_gateway_wait import _ApprovalEntry @pytest.fixture(autouse=True) diff --git a/tests/hermes_cli/test_approval_transport.py b/tests/hermes_cli/test_approval_transport.py index e555e4a8ae..a0fde80a12 100644 --- a/tests/hermes_cli/test_approval_transport.py +++ b/tests/hermes_cli/test_approval_transport.py @@ -11,6 +11,7 @@ import pytest import yaml from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest +from tools import approval_context, approval_prompt def _manifest(name: str = "fixture-approval") -> PluginManifest: @@ -249,7 +250,7 @@ def test_host_caps_hung_transport_workers(): def _configure_manual_guard(monkeypatch, approval_module, manager, *, fallback=None): - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual") monkeypatch.setattr(approval_module, "_is_interactive_cli", lambda: True) monkeypatch.setattr(approval_module, "_is_gateway_approval_context", lambda: False) monkeypatch.setattr(approval_module, "detect_hardline_command", lambda command: (False, "")) @@ -263,13 +264,8 @@ def _configure_manual_guard(monkeypatch, approval_module, manager, *, fallback=N lambda *args, **kwargs: "session-a", ) monkeypatch.setattr(approval_module, "is_approved", lambda *args: False) - monkeypatch.setattr(approval_module, "get_plugin_manager", lambda: manager, raising=False) - monkeypatch.setattr( - approval_module, - "_get_approval_transport_config", - lambda: ("phone", fallback), - raising=False, - ) + monkeypatch.setattr(approval_prompt, "get_plugin_manager", lambda: manager) + monkeypatch.setattr(approval_context, "_get_approval_transport_config", lambda: ("phone", fallback)) monkeypatch.setattr("tools.tirith_security.check_command_security", lambda command: {"action": "allow"}) @@ -325,16 +321,16 @@ def test_execute_code_gateway_uses_selected_transport(monkeypatch): _context(manager).register_approval_transport( "phone", lambda request: seen.append(request) or request.respond("once") ) - monkeypatch.setattr(approval, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual") monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: True) monkeypatch.setattr(approval, "_is_cron_approval_context", lambda: False) monkeypatch.setattr( approval, "get_current_session_key", lambda *args, **kwargs: "session-a" ) monkeypatch.setattr(approval, "is_approved", lambda *args: False) - monkeypatch.setattr(approval, "get_plugin_manager", lambda: manager) + monkeypatch.setattr(approval_prompt, "get_plugin_manager", lambda: manager) monkeypatch.setattr( - approval, "_get_approval_transport_config", lambda: ("phone", None) + approval_context, "_get_approval_transport_config", lambda: ("phone", None) ) monkeypatch.setattr(approval, "_gateway_notify_cbs", {}) @@ -369,13 +365,13 @@ def test_transport_resolution_error_does_not_log_plugin_exception(monkeypatch, c from tools import approval monkeypatch.setattr( - approval, "_get_approval_transport_config", lambda: ("phone", None) + approval_context, "_get_approval_transport_config", lambda: ("phone", None) ) def broken_manager(): raise RuntimeError("plugin-owned-secret-value") - monkeypatch.setattr(approval, "get_plugin_manager", broken_manager) + monkeypatch.setattr(approval_prompt, "get_plugin_manager", broken_manager) result = approval._present_with_selected_transport( command="rm -rf /tmp/example", @@ -500,7 +496,7 @@ def register(ctx): monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", False) manager = PluginManager() monkeypatch.setattr(plugins_module, "_plugin_manager", manager) - token = approval.set_hermes_interactive_context(True) + token = approval_context.set_hermes_interactive_context(True) approval.clear_session("local") approval._permanent_approved.clear() try: @@ -511,17 +507,17 @@ def register(ctx): reloaded = approval.check_all_command_guards( "rm -rf /tmp/hermes-approval-transport-fixture-reloaded", "local" ) - gateway_token = approval.set_hermes_interactive_context(False) + gateway_token = approval_context.set_hermes_interactive_context(False) monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") try: gateway_routed = approval.check_all_command_guards( "rm -rf /tmp/hermes-approval-transport-fixture-gateway", "local" ) finally: - approval.reset_hermes_interactive_context(gateway_token) + approval_context.reset_hermes_interactive_context(gateway_token) hardline = approval.check_all_command_guards("rm -rf /", "local") finally: - approval.reset_hermes_interactive_context(token) + approval_context.reset_hermes_interactive_context(token) records = [ json.loads(line) diff --git a/tests/hermes_cli/test_approvals_test.py b/tests/hermes_cli/test_approvals_test.py index cda10a05f6..d8b78358ae 100644 --- a/tests/hermes_cli/test_approvals_test.py +++ b/tests/hermes_cli/test_approvals_test.py @@ -16,6 +16,8 @@ import json import pytest import tools.approval as A +from tools import approval_context +from tools import approval_detection, approval_floors from hermes_cli import approvals_test as at @@ -30,7 +32,7 @@ def _args(command, env_type="local", as_json=False): @pytest.fixture def isolated_approvals(monkeypatch): """Isolate the evaluators from the dev machine's real config/state.""" - monkeypatch.setattr(A, "_get_approval_config", lambda: {"mode": "manual"}) + monkeypatch.setattr(approval_context, "_get_approval_config", lambda: {"mode": "manual"}) monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False) monkeypatch.setattr(A, "is_current_session_yolo_enabled", lambda: False) monkeypatch.setattr(A, "load_permanent_allowlist", lambda: set()) @@ -71,7 +73,7 @@ class TestVerdicts: def test_user_deny_rule_from_config_honored(self, isolated_approvals, capsys, monkeypatch): monkeypatch.setattr( - A, "_get_approval_config", + approval_context, "_get_approval_config", lambda: {"mode": "manual", "deny": ["git push *"]}) rc = at.approvals_test_command(_args(["git", "push", "origin", "main"])) out = capsys.readouterr().out @@ -91,7 +93,7 @@ class TestVerdicts: def test_mode_off_bypasses_dangerous_but_not_hardline(self, isolated_approvals, capsys, monkeypatch): - monkeypatch.setattr(A, "_get_approval_config", lambda: {"mode": "off"}) + monkeypatch.setattr(approval_context, "_get_approval_config", lambda: {"mode": "off"}) rc = at.approvals_test_command(_args(["rm", "-rf", "~/project/build"])) out = capsys.readouterr().out assert rc == 0 @@ -134,14 +136,14 @@ class TestNormalizationParity: return real(c) return wrapper - monkeypatch.setattr(A, "detect_hardline_command", - _spy("hardline", A.detect_hardline_command)) - monkeypatch.setattr(A, "detect_dangerous_command", - _spy("dangerous", A.detect_dangerous_command)) - monkeypatch.setattr(A, "_match_user_deny_rule", - _spy("deny", A._match_user_deny_rule)) - monkeypatch.setattr(A, "_command_detection_variants", - _spy("variants", A._command_detection_variants)) + monkeypatch.setattr(approval_detection, "detect_hardline_command", + _spy("hardline", approval_detection.detect_hardline_command)) + monkeypatch.setattr(approval_detection, "detect_dangerous_command", + _spy("dangerous", approval_detection.detect_dangerous_command)) + monkeypatch.setattr(approval_floors, "_match_user_deny_rule", + _spy("deny", approval_floors._match_user_deny_rule)) + monkeypatch.setattr(approval_detection, "_command_detection_variants", + _spy("variants", approval_detection._command_detection_variants)) cmd = "rm -rf ~/project/build" at.approvals_test_command(_args(cmd.split())) capsys.readouterr() diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index d44abafb4f..d0bb39e53e 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -1354,6 +1354,7 @@ class TestResolvePreToolBlock: def test_approve_gate_receives_tool_observability_context(self, monkeypatch): from hermes_cli.plugins import resolve_pre_tool_block from tools import approval + from tools import approval_context seen = {} monkeypatch.setattr( @@ -1364,8 +1365,8 @@ class TestResolvePreToolBlock: ) def _approve(*args, **kwargs): - seen["turn_id"] = approval._approval_turn_id.get() - seen["tool_call_id"] = approval._approval_tool_call_id.get() + seen["turn_id"] = approval_context._approval_turn_id.get() + seen["tool_call_id"] = approval_context._approval_tool_call_id.get() return {"approved": True, "message": None} monkeypatch.setattr("tools.approval.request_tool_approval", _approve) diff --git a/tests/run_agent/test_authorization_gate.py b/tests/run_agent/test_authorization_gate.py index 7b29c0a26e..354731914e 100644 --- a/tests/run_agent/test_authorization_gate.py +++ b/tests/run_agent/test_authorization_gate.py @@ -26,15 +26,17 @@ import pytest from agent.tool_executor import _ConcurrentToolAuthorizationGate from tools import approval as approval_mod +from tools import approval_context +from tools import approval_human_wait @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() SESSION = "test-session-79719" @@ -48,14 +50,14 @@ def _make_gate(**kwargs) -> _ConcurrentToolAuthorizationGate: class TestHumanWaitTracker: def test_no_wait_reports_zero(self): - assert approval_mod.human_wait_seconds(SESSION) == 0.0 + assert approval_human_wait.human_wait_seconds(SESSION) == 0.0 def test_open_window_counts(self): opened = threading.Event() release = threading.Event() def _wait(): - with approval_mod.human_wait_window(SESSION): + with approval_human_wait.human_wait_window(SESSION): opened.set() release.wait(timeout=5) @@ -63,13 +65,13 @@ class TestHumanWaitTracker: t.start() assert opened.wait(timeout=5) time.sleep(0.05) - assert approval_mod.human_wait_seconds(SESSION) > 0.0 + assert approval_human_wait.human_wait_seconds(SESSION) > 0.0 release.set() t.join(timeout=5) # Window closed: total is frozen (completed_seconds), not still growing. - first = approval_mod.human_wait_seconds(SESSION) + first = approval_human_wait.human_wait_seconds(SESSION) time.sleep(0.05) - assert approval_mod.human_wait_seconds(SESSION) == pytest.approx(first) + assert approval_human_wait.human_wait_seconds(SESSION) == pytest.approx(first) def test_overlapping_windows_coalesce(self): """Two concurrent windows on one session must not double-count wall clock.""" @@ -77,7 +79,7 @@ class TestHumanWaitTracker: started = threading.Barrier(3) def _wait(): - with approval_mod.human_wait_window(SESSION): + with approval_human_wait.human_wait_window(SESSION): started.wait(timeout=5) release.wait(timeout=5) @@ -92,47 +94,47 @@ class TestHumanWaitTracker: t.join(timeout=5) elapsed = time.monotonic() - start # Coalesced: recorded ≤ wall clock (a double count would be ~2×). - assert approval_mod.human_wait_seconds(SESSION) <= elapsed + 0.05 + assert approval_human_wait.human_wait_seconds(SESSION) <= elapsed + 0.05 def test_sessions_are_isolated(self): - with approval_mod.human_wait_window("other-session"): + with approval_human_wait.human_wait_window("other-session"): time.sleep(0.05) - assert approval_mod.human_wait_seconds(SESSION) == 0.0 - assert approval_mod.human_wait_seconds("other-session") > 0.0 + assert approval_human_wait.human_wait_seconds(SESSION) == 0.0 + assert approval_human_wait.human_wait_seconds("other-session") > 0.0 def test_open_window_clamped_to_approval_timeout(self, monkeypatch): """A window that overstays approvals.timeout is itself wedged and must stop extending the exclusion (belt-and-braces for #79719).""" - monkeypatch.setattr(approval_mod, "_get_approval_timeout", lambda: 300) - with approval_mod.human_wait_window(SESSION): - state = approval_mod._human_wait_states[SESSION] + monkeypatch.setattr(approval_context, "_get_approval_timeout", lambda: 300) + with approval_human_wait.human_wait_window(SESSION): + state = approval_human_wait._human_wait_states[SESSION] # Simulate a window that has been open for a full day. state.window_started = time.monotonic() - 86_400.0 - assert approval_mod.human_wait_seconds(SESSION) <= 300.0 + 60.0 + assert approval_human_wait.human_wait_seconds(SESSION) <= 300.0 + 60.0 def test_eviction_keeps_pending_sessions(self): - with approval_mod.human_wait_window(SESSION): - for i in range(approval_mod._HUMAN_WAIT_MAX_SESSIONS + 8): - with approval_mod.human_wait_window(f"burst-{i}"): + with approval_human_wait.human_wait_window(SESSION): + for i in range(approval_human_wait._HUMAN_WAIT_MAX_SESSIONS + 8): + with approval_human_wait.human_wait_window(f"burst-{i}"): pass # The active session survived the eviction pressure and the table # stayed at (or under) its cap. - assert SESSION in approval_mod._human_wait_states - assert approval_mod._human_wait_states[SESSION].pending == 1 + assert SESSION in approval_human_wait._human_wait_states + assert approval_human_wait._human_wait_states[SESSION].pending == 1 assert ( - len(approval_mod._human_wait_states) - <= approval_mod._HUMAN_WAIT_MAX_SESSIONS + len(approval_human_wait._human_wait_states) + <= approval_human_wait._HUMAN_WAIT_MAX_SESSIONS ) def test_late_close_of_wedged_window_is_clamped(self, monkeypatch): """A wedged window that eventually CLOSES must not retroactively inject its full overstay into completed_seconds (close-side clamp).""" - monkeypatch.setattr(approval_mod, "_get_approval_timeout", lambda: 300) - with approval_mod.human_wait_window(SESSION): - state = approval_mod._human_wait_states[SESSION] + monkeypatch.setattr(approval_context, "_get_approval_timeout", lambda: 300) + with approval_human_wait.human_wait_window(SESSION): + state = approval_human_wait._human_wait_states[SESSION] # Simulate the window having been open for a full day before close. state.window_started = time.monotonic() - 86_400.0 - assert approval_mod.human_wait_seconds(SESSION) <= 300.0 + 60.0 + assert approval_human_wait.human_wait_seconds(SESSION) <= 300.0 + 60.0 class TestAuthorizationGate: @@ -247,20 +249,20 @@ class TestAuthorizationGate: def test_human_wait_is_excluded(self): """A genuine approval wait during the batch extends the deadline.""" gate = _make_gate() - with approval_mod.human_wait_window(SESSION): + with approval_human_wait.human_wait_window(SESSION): time.sleep(0.1) assert gate.excluded_seconds() >= 0.09 def test_baseline_ignores_waits_before_batch(self): """Approval waits from BEFORE this batch must not extend its deadline.""" - with approval_mod.human_wait_window(SESSION): + with approval_human_wait.human_wait_window(SESSION): time.sleep(0.1) gate = _make_gate() assert gate.excluded_seconds() == 0.0 def test_other_sessions_wait_not_excluded(self): gate = _make_gate() - with approval_mod.human_wait_window("unrelated-session"): + with approval_human_wait.human_wait_window("unrelated-session"): time.sleep(0.05) assert gate.excluded_seconds() == 0.0 @@ -268,7 +270,7 @@ class TestAuthorizationGate: class TestApprovalPathsRecordHumanWait: def test_await_gateway_decision_records_wait(self, monkeypatch): """The gateway approval poll loop must mark itself as human wait.""" - monkeypatch.setattr(approval_mod, "_get_approval_timeout", lambda: 300) + monkeypatch.setattr(approval_context, "_get_approval_timeout", lambda: 300) approval_data = { "command": "rm -rf /tmp/x", "description": "test", @@ -288,21 +290,21 @@ class TestApprovalPathsRecordHumanWait: assert notified.wait(timeout=5) time.sleep(0.1) try: - assert approval_mod.human_wait_seconds(SESSION) > 0.0 + assert approval_human_wait.human_wait_seconds(SESSION) > 0.0 finally: # Resolve the pending entry via the real production path. approval_mod.resolve_gateway_approval(SESSION, "deny", resolve_all=True) t.join(timeout=5) assert not t.is_alive() # Window closed once the wait resolved. - assert approval_mod._human_wait_states[SESSION].pending == 0 + assert approval_human_wait._human_wait_states[SESSION].pending == 0 def test_prompt_dangerous_approval_records_wait(self, monkeypatch): """The CLI prompt path must mark itself as human wait.""" observed = {} def _callback(_command, _description, **_kwargs): - observed["during"] = approval_mod.human_wait_seconds() + observed["during"] = approval_human_wait.human_wait_seconds() return "deny" choice = approval_mod.prompt_dangerous_approval( @@ -310,7 +312,7 @@ class TestApprovalPathsRecordHumanWait: ) assert choice == "deny" # The window was open while the callback (the human prompt) ran. - state = approval_mod._human_wait_states.get( + state = approval_human_wait._human_wait_states.get( approval_mod.get_current_session_key() ) assert state is not None diff --git a/tests/run_agent/test_tool_executor_contextvar_propagation.py b/tests/run_agent/test_tool_executor_contextvar_propagation.py index 5f32648a4a..613295a61c 100644 --- a/tests/run_agent/test_tool_executor_contextvar_propagation.py +++ b/tests/run_agent/test_tool_executor_contextvar_propagation.py @@ -84,10 +84,8 @@ def test_run_tool_worker_sees_parent_approval_session_key(): If the PR's ``copy_context().run`` wrapper is reverted, this test fails with ``Expected 'session-A' but worker saw 'default'``. """ - from tools.approval import ( - _approval_session_key, - get_current_session_key, - ) + from tools.approval import get_current_session_key + from tools.approval_context import _approval_session_key observed: dict = {} barrier = threading.Event() @@ -129,10 +127,8 @@ def test_two_concurrent_tool_batches_keep_session_keys_isolated(): snapshot across callers (which would collapse isolation the same way the unfixed ``submit`` does). """ - from tools.approval import ( - _approval_session_key, - get_current_session_key, - ) + from tools.approval import get_current_session_key + from tools.approval_context import _approval_session_key results: dict = {} diff --git a/tests/tools/test_allowlist_quoted_metachars.py b/tests/tools/test_allowlist_quoted_metachars.py index c4332bf2f1..386977aff3 100644 --- a/tests/tools/test_allowlist_quoted_metachars.py +++ b/tests/tools/test_allowlist_quoted_metachars.py @@ -11,10 +11,8 @@ a ``-c``/``-e``-style option would hand to another interpreter. import pytest -from tools.approval import ( - _command_matches_permanent_allowlist, - _has_allowlist_shell_operator, -) +from tools.approval import _command_matches_permanent_allowlist +from tools.approval_floors import _has_allowlist_shell_operator class TestHasAllowlistShellOperator: diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index 0d0a666bcb..406b230787 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -10,18 +10,13 @@ from unittest.mock import patch as mock_patch import pytest import tools.approval as approval_module +from tools import approval_context +from tools import approval_smart from hermes_constants import get_hermes_home -from tools.approval import ( - _get_approval_mode, - _normalize_approval_mode, - _smart_approve, - approve_session, - detect_dangerous_command, - detect_hardline_command, - is_approved, - load_permanent, - prompt_dangerous_approval, -) +from tools.approval import approve_session, detect_dangerous_command, detect_hardline_command, is_approved, load_permanent, prompt_dangerous_approval +from tools.approval_context import _get_approval_mode +from tools.approval_context import _normalize_approval_mode +from tools.approval_smart import _smart_approve class TestApprovalModeParsing: @@ -63,12 +58,11 @@ class TestSmartApproval: monkeypatch.setenv("HERMES_EXEC_ASK", "1") monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) monkeypatch.setattr( - approval_module, - "_get_approval_config", + approval_context, "_get_approval_config", lambda: {"mode": "smart"}, ) monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(approval_module, "_smart_approve", lambda *_: "approve") + monkeypatch.setattr(approval_smart, "_smart_approve", lambda *_: "approve") monkeypatch.setattr( "tools.tirith_security.check_command_security", lambda _command: {"action": "allow", "findings": [], "summary": ""}, @@ -237,12 +231,12 @@ class TestApproveAndCheckSession: class TestSessionKeyContext: def test_context_session_key_overrides_process_env(self): - token = approval_module.set_current_session_key("alice") + token = approval_context.set_current_session_key("alice") try: with mock_patch.dict("os.environ", {"HERMES_SESSION_KEY": "bob"}, clear=False): assert approval_module.get_current_session_key() == "alice" finally: - approval_module.reset_current_session_key(token) + approval_context.reset_current_session_key(token) class TestRmFalsePositiveFix: @@ -730,10 +724,8 @@ class TestWebhookApprovalExclusion: def test_all_unattended_platforms_return_false(self, monkeypatch): """Every unattended programmatic platform is excluded, not just webhook.""" - from tools.approval import ( - _UNATTENDED_APPROVAL_PLATFORMS, - _is_gateway_approval_context, - ) + from tools.approval import _is_gateway_approval_context + from tools.approval_context import _UNATTENDED_APPROVAL_PLATFORMS monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) @@ -773,9 +765,11 @@ class TestWebhookApprovalExclusion: def _isolate(self, monkeypatch): """Neutralize host leakage: yolo frozen at import time + real config.""" import tools.approval as approval_mod + from tools import approval_context + from tools import approval_context monkeypatch.setattr(approval_mod, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(approval_mod, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart") def test_webhook_dangerous_command_denies_by_default(self, monkeypatch): """Webhook sessions that trigger dangerous commands DENY instantly. @@ -811,7 +805,7 @@ class TestWebhookApprovalExclusion: monkeypatch.setenv("HERMES_SESSION_PLATFORM", "webhook") monkeypatch.setenv("HERMES_SESSION_KEY", "test-webhook-session") monkeypatch.setattr( - approval_mod, "_get_unattended_approval_mode", lambda: "approve" + approval_context, "_get_unattended_approval_mode", lambda: "approve" ) result = check_all_command_guards("sudo systemctl restart nginx", "local") @@ -1333,6 +1327,8 @@ class TestApprovalTimeoutIsNotConsent: def setup_method(self): """Reset module state and force a tight approval timeout for fast tests.""" from tools import approval as mod + from tools import approval_context + from tools import approval_context mod._gateway_queues.clear() mod._gateway_notify_cbs.clear() mod._session_approved.clear() @@ -1367,7 +1363,7 @@ class TestApprovalTimeoutIsNotConsent: def _force_short_timeout(self, monkeypatch, seconds=0.05): from tools import approval as mod monkeypatch.setattr( - mod, "_get_approval_config", + approval_context, "_get_approval_config", lambda: {"mode": "manual", "timeout": seconds}, ) @@ -1384,13 +1380,13 @@ class TestApprovalTimeoutIsNotConsent: mod.register_gateway_notify(self.SESSION_KEY, lambda data: notified.append(data)) hook_calls = [] - original_fire = mod._fire_approval_hook + original_fire = approval_context._fire_approval_hook def _capture(event_name, **kwargs): hook_calls.append((event_name, kwargs)) return original_fire(event_name, **kwargs) - monkeypatch.setattr(mod, "_fire_approval_hook", _capture) + monkeypatch.setattr(approval_context, "_fire_approval_hook", _capture) result = mod.check_all_command_guards("rm -rf .git", "local") @@ -1462,13 +1458,13 @@ class TestApprovalTimeoutIsNotConsent: mod.register_gateway_notify(self.SESSION_KEY, lambda data: None) hook_calls = [] - original_fire = mod._fire_approval_hook + original_fire = approval_context._fire_approval_hook def _capture(event_name, **kwargs): hook_calls.append((event_name, kwargs)) return original_fire(event_name, **kwargs) - monkeypatch.setattr(mod, "_fire_approval_hook", _capture) + monkeypatch.setattr(approval_context, "_fire_approval_hook", _capture) mod.check_all_command_guards("rm -rf .git", "local") @@ -1489,7 +1485,7 @@ class TestApprovalTimeoutIsNotConsent: def _capture(event_name, **kwargs): hook_calls.append((event_name, kwargs)) - monkeypatch.setattr(mod, "_fire_approval_hook", _capture) + monkeypatch.setattr(approval_context, "_fire_approval_hook", _capture) def _fail_notify(_data): raise RuntimeError("private gateway failure") @@ -1635,7 +1631,7 @@ class TestConcurrentApprovalCoalescing: def test_identical_concurrent_approvals_send_one_prompt(self, monkeypatch): from tools import approval as mod - monkeypatch.setattr(mod, "_get_approval_timeout", lambda: 30) + monkeypatch.setattr(approval_context, "_get_approval_timeout", lambda: 30) notified = [] results, threads = self._spawn_waits(mod, notified, n=3) @@ -1655,7 +1651,7 @@ class TestConcurrentApprovalCoalescing: def test_deny_propagates_to_followers(self, monkeypatch): from tools import approval as mod - monkeypatch.setattr(mod, "_get_approval_timeout", lambda: 30) + monkeypatch.setattr(approval_context, "_get_approval_timeout", lambda: 30) notified = [] results, threads = self._spawn_waits(mod, notified, n=2) @@ -1671,7 +1667,7 @@ class TestConcurrentApprovalCoalescing: def test_once_makes_follower_reprompt(self, monkeypatch): from tools import approval as mod - monkeypatch.setattr(mod, "_get_approval_timeout", lambda: 30) + monkeypatch.setattr(approval_context, "_get_approval_timeout", lambda: 30) notified = [] results, threads = self._spawn_waits(mod, notified, n=2) @@ -1692,7 +1688,7 @@ class TestConcurrentApprovalCoalescing: def test_different_commands_are_not_coalesced(self, monkeypatch): from tools import approval as mod - monkeypatch.setattr(mod, "_get_approval_timeout", lambda: 30) + monkeypatch.setattr(approval_context, "_get_approval_timeout", lambda: 30) import threading notified = [] @@ -1873,7 +1869,7 @@ class TestApprovalPromptRedaction: with _patch("hermes_cli.config.load_config_readonly", return_value=cfg): with _patch("tools.approval._is_gateway_approval_context", return_value=True): - with _patch("tools.approval._get_approval_mode", + with _patch("tools.approval_context._get_approval_mode", return_value="manual"): # No gateway notify callback registered -> pending fallback. result = check_execute_code_guard(code, "local") diff --git a/tests/tools/test_approval_config_readonly.py b/tests/tools/test_approval_config_readonly.py index 933b46e225..4e91773a46 100644 --- a/tests/tools/test_approval_config_readonly.py +++ b/tests/tools/test_approval_config_readonly.py @@ -15,13 +15,9 @@ These tests drive the REAL functions against a temp HERMES_HOME config import pytest import hermes_cli.config as hc -from tools.approval import ( - _get_approval_config, - _get_approval_mode, - _get_cron_approval_mode, - check_all_command_guards, - load_permanent_allowlist, -) +from tools.approval import check_all_command_guards, load_permanent_allowlist +from tools.approval_context import _get_approval_config, _get_approval_mode +from tools.approval_context import _get_cron_approval_mode from tools.tirith_security import _load_security_config diff --git a/tests/tools/test_approval_deny_rules.py b/tests/tools/test_approval_deny_rules.py index 4fe7dedb3e..e809c7912c 100644 --- a/tests/tools/test_approval_deny_rules.py +++ b/tests/tools/test_approval_deny_rules.py @@ -10,6 +10,7 @@ import os import pytest from tools import approval as mod +from tools import approval_context @pytest.fixture @@ -21,7 +22,7 @@ def deny_config(monkeypatch): def set_deny(patterns, **extra): state["config"] = {"mode": "manual", "deny": list(patterns), **extra} - monkeypatch.setattr(mod, "_get_approval_config", lambda: state["config"]) + monkeypatch.setattr(approval_context, "_get_approval_config", lambda: state["config"]) return set_deny @@ -41,14 +42,14 @@ class TestMatchUserDenyRule: assert mod._match_user_deny_rule("git push --force origin main") is None def test_missing_key_is_noop(self, monkeypatch): - monkeypatch.setattr(mod, "_get_approval_config", lambda: {"mode": "manual"}) + monkeypatch.setattr(approval_context, "_get_approval_config", lambda: {"mode": "manual"}) assert mod._match_user_deny_rule("rm -rf build/") is None def test_config_load_failure_fails_open(self, monkeypatch): def boom(): raise RuntimeError("config unavailable") - monkeypatch.setattr(mod, "_get_approval_config", boom) + monkeypatch.setattr(approval_context, "_get_approval_config", boom) assert mod._match_user_deny_rule("git push --force") is None def test_quote_obfuscation_still_matches(self, deny_config): diff --git a/tests/tools/test_approval_hook_session_id.py b/tests/tools/test_approval_hook_session_id.py index b079c4d5be..f24b4bf1f8 100644 --- a/tests/tools/test_approval_hook_session_id.py +++ b/tests/tools/test_approval_hook_session_id.py @@ -17,6 +17,7 @@ from __future__ import annotations from unittest.mock import patch from tools import approval as approval_mod +from tools import approval_context def _capture_hook(captured): @@ -28,7 +29,7 @@ def _capture_hook(captured): class TestApprovalHookSessionId: def test_session_id_forwarded_when_bound(self): captured = [] - tokens = approval_mod.set_current_observability_context( + tokens = approval_context.set_current_observability_context( turn_id="turn-1", tool_call_id="call-1", session_id="20260810_test_session", @@ -38,14 +39,14 @@ class TestApprovalHookSessionId: "hermes_cli.lifecycle.invoke_hook", side_effect=_capture_hook(captured), ): - approval_mod._fire_approval_hook( + approval_context._fire_approval_hook( "pre_approval_request", command="rm -rf /etc/hosts", description="dangerous", surface="gateway", ) finally: - approval_mod.reset_current_observability_context(tokens) + approval_context.reset_current_observability_context(tokens) assert captured, "hook must dispatch" _, kwargs = captured[0] @@ -55,7 +56,7 @@ class TestApprovalHookSessionId: def test_explicit_session_id_not_clobbered(self): captured = [] - tokens = approval_mod.set_current_observability_context( + tokens = approval_context.set_current_observability_context( session_id="context-session", ) try: @@ -63,13 +64,13 @@ class TestApprovalHookSessionId: "hermes_cli.lifecycle.invoke_hook", side_effect=_capture_hook(captured), ): - approval_mod._fire_approval_hook( + approval_context._fire_approval_hook( "post_approval_response", session_id="explicit-session", choice="approved", ) finally: - approval_mod.reset_current_observability_context(tokens) + approval_context.reset_current_observability_context(tokens) _, kwargs = captured[0] assert kwargs.get("session_id") == "explicit-session" @@ -80,7 +81,7 @@ class TestApprovalHookSessionId: "hermes_cli.lifecycle.invoke_hook", side_effect=_capture_hook(captured), ): - approval_mod._fire_approval_hook( + approval_context._fire_approval_hook( "pre_approval_request", command="x", description="y", diff --git a/tests/tools/test_approval_interrupt.py b/tests/tools/test_approval_interrupt.py index 2ef91f752c..3db0fcc2d1 100644 --- a/tests/tools/test_approval_interrupt.py +++ b/tests/tools/test_approval_interrupt.py @@ -65,15 +65,16 @@ class TestApprovalInterrupt: os.environ[k] = v _clear_approval_state() - def test_interrupt_unblocks_pending_approval_quickly(self): + def test_interrupt_unblocks_pending_approval_quickly(self, monkeypatch): """An interrupt on the waiting thread must resolve the wait as deny well before the (here, intentionally long) approval timeout.""" from tools import approval as mod + from tools import approval_context from tools.interrupt import set_interrupt # Force a long timeout so a *passing* test can only happen via the # interrupt path, never by the deadline elapsing. - mod._get_approval_config = lambda: {"timeout": 300} + monkeypatch.setattr(approval_context, "_get_approval_config", lambda: {"timeout": 300}) approval_data = { "command": "rm -rf /tmp/whatever", @@ -120,15 +121,16 @@ class TestApprovalInterrupt: # Queue entry was cleaned up. assert not mod.has_blocking_approval(self.SESSION_KEY) - def test_unrelated_thread_interrupt_does_not_unblock(self): + def test_unrelated_thread_interrupt_does_not_unblock(self, monkeypatch): """An interrupt flagged on a *different* thread must NOT release this session's approval wait — interrupts are thread-scoped.""" from tools import approval as mod + from tools import approval_context from tools.interrupt import set_interrupt # Short timeout so the test finishes fast via the deadline, proving the # foreign interrupt did not short-circuit the wait. - mod._get_approval_config = lambda: {"timeout": 1} + monkeypatch.setattr(approval_context, "_get_approval_config", lambda: {"timeout": 1}) approval_data = { "command": "rm -rf /tmp/whatever", diff --git a/tests/tools/test_approval_mode_parity.py b/tests/tools/test_approval_mode_parity.py index 41545761ad..d35014117f 100644 --- a/tests/tools/test_approval_mode_parity.py +++ b/tests/tools/test_approval_mode_parity.py @@ -3,8 +3,8 @@ The approval mode (``approvals.mode``) and timeout (``approvals.timeout``) must resolve identically on every surface that consults them: - - the canonical core: ``tools.approval._get_approval_mode`` / - ``tools.approval._get_approval_timeout`` + - the canonical core: ``tools.approval_context._get_approval_mode`` / + ``tools.approval_context._get_approval_timeout`` - the TUI gateway: ``tui_gateway.server._load_approval_mode`` (delegates to the core as of the decision-core migration) - the codex app-server surface: ``agent/codex_runtime.py`` feeds @@ -113,8 +113,9 @@ def test_mode_and_timeout_parity_across_surfaces( _write_config(hermes_home, yaml_text) - core_mode = approval_mod._get_approval_mode() - core_timeout = approval_mod._get_approval_timeout() + ctx = importlib.import_module("tools.approval_context") + core_mode = ctx._get_approval_mode() + core_timeout = ctx._get_approval_timeout() tui_mode = tui_server._load_approval_mode() # Canonical resolver matches expectations. @@ -142,15 +143,15 @@ def test_tui_loader_delegates_to_core(hermes_home, tui_server): Pin the delegation seam directly: patching the core resolver changes what the TUI reports, proving there is no independent config read left. """ - approval_mod = _approval_module() + approval_context = importlib.import_module("tools.approval_context") - with patch.object(approval_mod, "_get_approval_mode", return_value="smart"): + with patch.object(approval_context, "_get_approval_mode", return_value="smart"): assert tui_server._load_approval_mode() == "smart" - with patch.object(approval_mod, "_get_approval_mode", return_value="off"): + with patch.object(approval_context, "_get_approval_mode", return_value="off"): assert tui_server._load_approval_mode() == "off" # Defensive clamp: an out-of-vocabulary value from the core is coerced # to manual rather than leaking an unknown mode to the TUI client. with patch.object( - approval_mod, "_get_approval_mode", return_value="weird" + approval_context, "_get_approval_mode", return_value="weird" ): assert tui_server._load_approval_mode() == "manual" diff --git a/tests/tools/test_approval_outcome_parity.py b/tests/tools/test_approval_outcome_parity.py index 57416673b6..3e6176885b 100644 --- a/tests/tools/test_approval_outcome_parity.py +++ b/tests/tools/test_approval_outcome_parity.py @@ -12,7 +12,7 @@ from __future__ import annotations import pytest import tools.approval as approval_mod -from tools import approval_human_wait +from tools import approval_context, approval_human_wait import tools.terminal_tool as terminal_tool import tools.terminal_tool_sudo as terminal_tool_sudo @@ -33,7 +33,7 @@ class TestSudoWaitExcludedFromDeadlines: def test_sudo_callback_wait_accrues_human_wait(self, monkeypatch): session = "sudo-test-session" monkeypatch.setattr( - approval_mod, "get_current_session_key", lambda default="": session + approval_context, "get_current_session_key", lambda default="": session ) def _slow_cb(): @@ -59,7 +59,7 @@ class TestSudoWaitExcludedFromDeadlines: """The non-callback path (thread + join) must be wrapped too.""" session = "sudo-join-session" monkeypatch.setattr( - approval_mod, "get_current_session_key", lambda default="": session + approval_context, "get_current_session_key", lambda default="": session ) monkeypatch.setattr(terminal_tool, "_get_sudo_password_callback", lambda: None) monkeypatch.setattr(terminal_tool, "_is_windows", False, raising=False) diff --git a/tests/tools/test_approval_plugin_hooks.py b/tests/tools/test_approval_plugin_hooks.py index 5493d274da..f03dff5ba3 100644 --- a/tests/tools/test_approval_plugin_hooks.py +++ b/tests/tools/test_approval_plugin_hooks.py @@ -11,12 +11,10 @@ from unittest.mock import patch import pytest import tools.approval as approval_module -from tools.approval import ( - check_all_command_guards, - check_execute_code_guard, - set_current_session_key, - clear_session, -) +from tools import approval_context +from tools import approval_smart +from tools.approval import check_all_command_guards, check_execute_code_guard, clear_session +from tools.approval_context import set_current_session_key @pytest.fixture @@ -24,6 +22,7 @@ def isolated_session(monkeypatch, tmp_path): """Give each test a fresh session_key, clean approval-state, and isolated HERMES_HOME so the real user's command_allowlist doesn't leak in.""" import tools.approval as _am + from tools import approval_context session_key = "test:session:approval_hooks" token = set_current_session_key(session_key) @@ -41,7 +40,7 @@ def isolated_session(monkeypatch, tmp_path): _am._permanent_approved.update(_saved_permanent) _am._session_approved.update(_saved_session) try: - _am._approval_session_key.reset(token) + approval_context._approval_session_key.reset(token) except Exception: pass clear_session(session_key) @@ -58,7 +57,7 @@ class TestCliPathFiresHooks: monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) # approvals.mode=manual so we actually reach the prompt site - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual") captured = [] @@ -98,7 +97,7 @@ class TestCliPathFiresHooks: monkeypatch.setenv("HERMES_INTERACTIVE", "1") monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual") captured = [] @@ -127,7 +126,7 @@ class TestCliPathFiresHooks: monkeypatch.setenv("HERMES_INTERACTIVE", "1") monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual") def boom(hook_name, **kwargs): raise RuntimeError("plugin crashed") @@ -158,8 +157,8 @@ class TestSmartModeFiresHooks: monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") - monkeypatch.setattr(approval_module, "_smart_approve", lambda *_: verdict) + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_smart, "_smart_approve", lambda *_: verdict) monkeypatch.setattr( "tools.tirith_security.check_command_security", lambda _: {"action": "allow", "findings": [], "summary": ""}, @@ -221,7 +220,7 @@ class TestSmartModeFiresHooks: events.append("smart_approve") return "approve" - monkeypatch.setattr(approval_module, "_smart_approve", decide) + monkeypatch.setattr(approval_smart, "_smart_approve", decide) with patch( "hermes_cli.plugins.invoke_hook", side_effect=lambda name, **kwargs: events.append(name), @@ -321,8 +320,8 @@ class TestSmartModeFiresHooks: monkeypatch.setenv("HERMES_EXEC_ASK", "1") monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "smart") - monkeypatch.setattr(approval_module, "_smart_approve", lambda *_: next(verdicts)) + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_smart, "_smart_approve", lambda *_: next(verdicts)) monkeypatch.setattr( "tools.tirith_security.check_command_security", lambda _: {"action": "allow", "findings": [], "summary": ""}, diff --git a/tests/tools/test_approval_timeout_overflow.py b/tests/tools/test_approval_timeout_overflow.py index 18d7105880..953a7f4a3b 100644 --- a/tests/tools/test_approval_timeout_overflow.py +++ b/tests/tools/test_approval_timeout_overflow.py @@ -17,26 +17,26 @@ from agent.deadline import MAX_SAFE_TIMEOUT_S def _with_configured_timeout(value): return patch( - "tools.approval._get_approval_config", + "tools.approval_context._get_approval_config", return_value={"timeout": value}, ) class TestApprovalTimeoutOverflowClamp: def test_normal_value_passes_through(self): - from tools.approval import _get_approval_timeout + from tools.approval_context import _get_approval_timeout with _with_configured_timeout(300): assert _get_approval_timeout() == 300 def test_oversized_value_clamped(self): - from tools.approval import _get_approval_timeout + from tools.approval_context import _get_approval_timeout with _with_configured_timeout(10**18): assert _get_approval_timeout() == int(MAX_SAFE_TIMEOUT_S) def test_invalid_value_falls_back_to_default(self): - from tools.approval import _get_approval_timeout + from tools.approval_context import _get_approval_timeout with _with_configured_timeout("soon"): assert _get_approval_timeout() == 300 @@ -44,7 +44,7 @@ class TestApprovalTimeoutOverflowClamp: def test_oversized_float_value_clamped(self): # YAML `1e18` arrives as a float, not an int — different int() path # than the string/int forms; the clamp must cover it too. - from tools.approval import _get_approval_timeout + from tools.approval_context import _get_approval_timeout with _with_configured_timeout(1e18): assert _get_approval_timeout() == int(MAX_SAFE_TIMEOUT_S) @@ -53,10 +53,11 @@ class TestApprovalTimeoutOverflowClamp: # Capping silently changes behavior for every consumer; operators # must see it happen. import tools.approval as approval_mod + from tools import approval_context with _with_configured_timeout(10**18): with caplog.at_level("WARNING", logger=approval_mod.__name__): - approval_mod._get_approval_timeout() + approval_context._get_approval_timeout() assert "exceeds the platform-safe maximum" in caplog.text def test_deadline_import_failure_fails_closed(self, monkeypatch): @@ -65,7 +66,7 @@ class TestApprovalTimeoutOverflowClamp: # exact time_t overflow this fix exists to prevent. import builtins - from tools.approval import _get_approval_timeout + from tools.approval_context import _get_approval_timeout real_import = builtins.__import__ @@ -86,7 +87,7 @@ class TestApprovalTimeoutOverflowClamp: def test_clamped_value_safe_for_lock_acquire(self): # The exact primitive that crashed in #83220: Lock.acquire on macOS # converts the relative timeout to an absolute time_t timestamp. - from tools.approval import _get_approval_timeout + from tools.approval_context import _get_approval_timeout with _with_configured_timeout(10**18): timeout = _get_approval_timeout() @@ -97,7 +98,7 @@ class TestApprovalTimeoutOverflowClamp: def test_clamped_value_safe_for_thread_join(self): # Sibling crash site: the CLI prompt fallback joins the input thread # with the configured timeout (tools/approval.py get_input path). - from tools.approval import _get_approval_timeout + from tools.approval_context import _get_approval_timeout with _with_configured_timeout(10**18): timeout = _get_approval_timeout() @@ -107,7 +108,7 @@ class TestApprovalTimeoutOverflowClamp: assert not t.is_alive() def test_human_wait_ceiling_inherits_clamp(self): - from tools.approval import HUMAN_WAIT_MARGIN_S, human_wait_ceiling + from tools.approval_human_wait import HUMAN_WAIT_MARGIN_S, human_wait_ceiling with _with_configured_timeout(10**18): ceiling = human_wait_ceiling() diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index 5fc37773f1..54e9943179 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -659,7 +659,7 @@ def test_delegate_task_background_uses_live_tui_agent_session_id(monkeypatch): from unittest.mock import MagicMock import tools.delegate_tool as dt from gateway.session_context import clear_session_vars, set_session_vars - from tools.approval import reset_current_session_key, set_current_session_key + from tools.approval_context import reset_current_session_key, set_current_session_key parent = MagicMock() parent._delegate_depth = 0 diff --git a/tests/tools/test_blocked_command_guidance.py b/tests/tools/test_blocked_command_guidance.py index 51378c6976..4b5a87a95c 100644 --- a/tests/tools/test_blocked_command_guidance.py +++ b/tests/tools/test_blocked_command_guidance.py @@ -2,8 +2,10 @@ import pytest -from tools.approval import _hardline_block_result, _PARSER_LIMIT_DESCRIPTION, _MALFORMED_EXEC_DESCRIPTION +from tools.approval import _hardline_block_result +from tools.approval_detection import _PARSER_LIMIT_DESCRIPTION, _MALFORMED_EXEC_DESCRIPTION from tools.terminal_tool import _foreground_background_guidance +from tools import approval_floors class TestParserLimitRecovery: @@ -27,7 +29,8 @@ class TestParserLimitRecovery: def test_save_failure_falls_back_to_manual_recipe(self, monkeypatch): import tools.approval as ap - monkeypatch.setattr(ap, "_save_blocked_payload", lambda c: None) + from tools import approval_floors + monkeypatch.setattr(approval_floors, "_save_blocked_payload", lambda c: None) r = _hardline_block_result(_PARSER_LIMIT_DESCRIPTION, "python3 -c 'x'") assert "write_file" in r["message"] assert "bash /path/script.sh" in r["message"] diff --git a/tests/tools/test_cli_approval_exec_ask_leak.py b/tests/tools/test_cli_approval_exec_ask_leak.py index a44933883a..f2f6d29508 100644 --- a/tests/tools/test_cli_approval_exec_ask_leak.py +++ b/tests/tools/test_cli_approval_exec_ask_leak.py @@ -20,6 +20,7 @@ from unittest.mock import patch import pytest import tools.approval as approval_module +from tools import approval_context from tools.approval import check_all_command_guards, check_execute_code_guard from tools.terminal_tool import set_approval_callback @@ -40,8 +41,7 @@ def _clean_approval_env(monkeypatch): monkeypatch.setenv("HERMES_INTERACTIVE", "1") monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False) monkeypatch.setattr( - approval_module, - "_get_approval_mode", + approval_context, "_get_approval_mode", lambda: "manual", ) monkeypatch.setattr( diff --git a/tests/tools/test_code_kernel.py b/tests/tools/test_code_kernel.py index 3741ffae65..bba5163844 100644 --- a/tests/tools/test_code_kernel.py +++ b/tests/tools/test_code_kernel.py @@ -216,7 +216,7 @@ class TestKernelOwnershipAndLifecycle(unittest.TestCase): """ def _run_as(self, session_key, code, task_id, **kwargs): - from tools.approval import reset_current_session_key, set_current_session_key + from tools.approval_context import reset_current_session_key, set_current_session_key token = set_current_session_key(session_key) try: diff --git a/tests/tools/test_command_guards.py b/tests/tools/test_command_guards.py index da385afcc3..953996fc2f 100644 --- a/tests/tools/test_command_guards.py +++ b/tests/tools/test_command_guards.py @@ -6,14 +6,9 @@ from unittest.mock import patch, MagicMock import pytest import tools.approval as approval_module -from tools.approval import ( - approve_session, - check_all_command_guards, - check_dangerous_command, - is_approved, - set_current_session_key, - reset_current_session_key, -) +from tools import approval_context +from tools.approval import approve_session, check_all_command_guards, check_dangerous_command, is_approved +from tools.approval_context import set_current_session_key, reset_current_session_key # Ensure the module is importable so we can patch it import tools.tirith_security @@ -43,7 +38,7 @@ def _mode_manual(monkeypatch): inside every prompting test — slow and flaky. These tests exercise the manual prompt flow, so force manual mode. """ - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual") @pytest.fixture(autouse=True) diff --git a/tests/tools/test_cron_approval_mode.py b/tests/tools/test_cron_approval_mode.py index 8ad17f5bda..8cdbcab47a 100644 --- a/tests/tools/test_cron_approval_mode.py +++ b/tests/tools/test_cron_approval_mode.py @@ -3,13 +3,11 @@ import pytest import tools.approval as approval_module +from tools import approval_context +from tools import approval_context from gateway.session_context import clear_session_vars, reset_session_vars, set_session_vars -from tools.approval import ( - _get_cron_approval_mode, - check_all_command_guards, - check_dangerous_command, - detect_dangerous_command, -) +from tools.approval import check_all_command_guards, check_dangerous_command, detect_dangerous_command +from tools.approval_context import _get_cron_approval_mode @pytest.fixture(autouse=True) @@ -115,8 +113,8 @@ class TestCronContextVarDetection: monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") - monkeypatch.setattr(approval_module, "_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: @@ -137,8 +135,8 @@ class TestCronContextVarDetection: monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") - monkeypatch.setattr(approval_module, "_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: @@ -163,7 +161,7 @@ class TestCronDenyMode: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"): result = check_dangerous_command("rm -rf /tmp/stuff", "local") assert not result["approved"] assert "BLOCKED" in result["message"] @@ -177,7 +175,7 @@ class TestCronDenyMode: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"): result = check_dangerous_command("ls -la", "local") assert result["approved"] @@ -196,7 +194,7 @@ class TestCronDenyMode: ] from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"): for cmd in dangerous_commands: is_dangerous, _, _ = detect_dangerous_command(cmd) if is_dangerous: @@ -212,7 +210,7 @@ class TestCronDenyMode: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"): result = check_dangerous_command("rm -rf /tmp/stuff", "local") assert not result["approved"] # Should contain the description of what was flagged @@ -229,7 +227,7 @@ class TestCronApproveMode: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="approve"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="approve"): result = check_dangerous_command("rm -rf /tmp/stuff", "local") assert result["approved"] @@ -249,7 +247,7 @@ class TestCronDenyModeAllGuards: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"): result = check_all_command_guards("rm -rf /tmp/stuff", "local") assert not result["approved"] assert "BLOCKED" in result["message"] @@ -262,7 +260,7 @@ class TestCronDenyModeAllGuards: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"): result = check_all_command_guards("echo hello", "local") assert result["approved"] @@ -274,7 +272,7 @@ class TestCronDenyModeAllGuards: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="approve"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="approve"): result = check_all_command_guards("rm -rf /tmp/stuff", "local") assert result["approved"] @@ -299,7 +297,7 @@ class TestCronDenyModeAllGuards: "summary": "homograph url", } with ( - mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"), + mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"), mock_patch("tools.approval.detect_dangerous_command", return_value=(False, None, None)), mock_patch("tools.tirith_security.check_command_security", @@ -329,7 +327,7 @@ class TestCronDenyModeAllGuards: return _real_import(name, *a, **k) with ( - mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"), + mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"), mock_patch("tools.approval.detect_dangerous_command", return_value=(False, None, None)), mock_patch("hermes_cli.config.load_config_readonly", @@ -360,7 +358,7 @@ class TestCronDenyModeAllGuards: return _real_import(name, *a, **k) with ( - mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"), + mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"), mock_patch("tools.approval.detect_dangerous_command", return_value=(False, None, None)), mock_patch("hermes_cli.config.load_config_readonly", @@ -387,7 +385,7 @@ class TestCronModeInteractions: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"): result = check_dangerous_command("rm -rf /", "docker") assert result["approved"] @@ -408,7 +406,7 @@ class TestCronModeInteractions: import tools.approval with ( mock_patch.object(tools.approval, "_YOLO_MODE_FROZEN", True), - mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"), + mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"), ): # Use a dangerous-but-not-hardline command — `rm -rf /` is now # hardline-blocked regardless of yolo (see test_hardline_blocklist.py). @@ -449,7 +447,7 @@ class TestCronWithGatewayOrigin: tokens = set_session_vars(platform="telegram", chat_id="123") try: from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="deny"): result = check_dangerous_command("rm -rf /tmp/stuff", "local") # Cron-mode path: BLOCKED message, NOT pending/approval_required. assert not result["approved"] @@ -471,7 +469,7 @@ class TestCronWithGatewayOrigin: tokens = set_session_vars(platform="discord", chat_id="456") try: from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_cron_approval_mode", return_value="approve"): + with mock_patch("tools.approval_context._get_cron_approval_mode", return_value="approve"): result = check_dangerous_command("rm -rf /tmp/stuff", "local") assert result["approved"] # Should NOT be a gateway-approval response. diff --git a/tests/tools/test_cronjob_run_background.py b/tests/tools/test_cronjob_run_background.py index c120273c03..58190889c5 100644 --- a/tests/tools/test_cronjob_run_background.py +++ b/tests/tools/test_cronjob_run_background.py @@ -42,7 +42,7 @@ def _bound_session_key(key="agent:main:telegram:dm:123"): """Context manager binding the approval session key contextvar.""" import contextlib - from tools.approval import _approval_session_key + from tools.approval_context import _approval_session_key @contextlib.contextmanager def _cm(): diff --git a/tests/tools/test_cronjob_run_delivery_notice.py b/tests/tools/test_cronjob_run_delivery_notice.py index e82b2ad090..34eecbbf4f 100644 --- a/tests/tools/test_cronjob_run_delivery_notice.py +++ b/tests/tools/test_cronjob_run_delivery_notice.py @@ -82,7 +82,7 @@ def _job(job_id, deliver): @contextlib.contextmanager def _bound_session_key(key): """Bind the approval session key contextvar (background dispatch gate).""" - from tools.approval import _approval_session_key + from tools.approval_context import _approval_session_key token = _approval_session_key.set(key) try: diff --git a/tests/tools/test_denial_circuit_breaker.py b/tests/tools/test_denial_circuit_breaker.py index 47ed8179de..63feb7b53e 100644 --- a/tests/tools/test_denial_circuit_breaker.py +++ b/tests/tools/test_denial_circuit_breaker.py @@ -16,6 +16,8 @@ from __future__ import annotations import pytest from tools import approval as A +from tools import approval_context +from tools import approval_smart BREAKER_MARKER = "CIRCUIT BREAKER:" @@ -32,9 +34,9 @@ def breaker_session(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: "smart") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart") monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(A, "_smart_approve", lambda _c, _d: "deny") + monkeypatch.setattr(approval_smart, "_smart_approve", lambda _c, _d: "deny") monkeypatch.setattr(A, "_get_denial_breaker_threshold", lambda: 3) monkeypatch.setattr( A, "detect_dangerous_command", @@ -47,7 +49,7 @@ def breaker_session(monkeypatch): ) session_key = "breaker-test-session" - token = A.set_current_session_key(session_key) + token = approval_context.set_current_session_key(session_key) A._reset_denials(session_key) with A._lock: A._permanent_approved.discard("breaker-test-danger") @@ -59,7 +61,7 @@ def breaker_session(monkeypatch): try: yield session_key finally: - A.reset_current_session_key(token) + approval_context.reset_current_session_key(token) A._reset_denials(session_key) with A._lock: A._gateway_queues.pop(session_key, None) @@ -117,12 +119,12 @@ def test_approval_resets_tally(breaker_session, monkeypatch): _denied_terminal("dangerous two") # Guardian approves the next command → tally resets. - monkeypatch.setattr(A, "_smart_approve", lambda _c, _d: "approve") + monkeypatch.setattr(approval_smart, "_smart_approve", lambda _c, _d: "approve") ok = _denied_terminal("benign command") assert ok["approved"] is True and ok.get("smart_approved") is True # Back to denials: the count restarts, so the next deny is #1, not #3. - monkeypatch.setattr(A, "_smart_approve", lambda _c, _d: "deny") + monkeypatch.setattr(approval_smart, "_smart_approve", lambda _c, _d: "deny") after = _denied_terminal("dangerous again") assert after["approved"] is False assert BREAKER_MARKER not in after["message"] @@ -168,9 +170,9 @@ def test_headless_smart_deny_increments_and_trips(monkeypatch): monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) monkeypatch.setenv("HERMES_EXEC_ASK", "0") - monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "smart") monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(A, "_smart_approve", lambda _c, _d: "deny") + monkeypatch.setattr(approval_smart, "_smart_approve", lambda _c, _d: "deny") monkeypatch.setattr(A, "_get_denial_breaker_threshold", lambda: 3) monkeypatch.setattr(A, "_is_interactive_cli", lambda: True) monkeypatch.setattr( @@ -187,7 +189,7 @@ def test_headless_smart_deny_increments_and_trips(monkeypatch): lambda *args, **kwargs: "deny") session_key = "headless-breaker-session" - token = A.set_current_session_key(session_key) + token = approval_context.set_current_session_key(session_key) A._reset_denials(session_key) with A._lock: A._permanent_approved.discard("headless-breaker-danger") @@ -201,7 +203,7 @@ def test_headless_smart_deny_increments_and_trips(monkeypatch): assert BREAKER_MARKER not in second["message"] assert BREAKER_MARKER in third["message"] finally: - A.reset_current_session_key(token) + approval_context.reset_current_session_key(token) A._reset_denials(session_key) diff --git a/tests/tools/test_execution_flag_detection.py b/tests/tools/test_execution_flag_detection.py index 65c1aeeb49..c397959e3c 100644 --- a/tests/tools/test_execution_flag_detection.py +++ b/tests/tools/test_execution_flag_detection.py @@ -284,7 +284,7 @@ def test_benign_segment_scaling_benchmark(): def test_max_accepted_separator_free_input_is_fast(): - from tools.approval import _MAX_SEPARATOR_FREE_COMMAND_CHARS + from tools.approval_detection import _MAX_SEPARATOR_FREE_COMMAND_CHARS command = "x" * _MAX_SEPARATOR_FREE_COMMAND_CHARS started = time.perf_counter() diff --git a/tests/tools/test_hardline_blocklist.py b/tests/tools/test_hardline_blocklist.py index 5057c2272e..07d1edb45c 100644 --- a/tests/tools/test_hardline_blocklist.py +++ b/tests/tools/test_hardline_blocklist.py @@ -9,17 +9,10 @@ Inspired by Mercury Agent's permission-hardened blocklist. import pytest -from tools.approval import ( - HARDLINE_PATTERNS, - check_all_command_guards, - check_dangerous_command, - detect_dangerous_command, - detect_hardline_command, - disable_session_yolo, - enable_session_yolo, - reset_current_session_key, - set_current_session_key, -) +from tools.approval import check_all_command_guards, check_dangerous_command, detect_dangerous_command, detect_hardline_command, disable_session_yolo, enable_session_yolo +from tools.approval_context import reset_current_session_key, set_current_session_key +from tools.approval_detection import HARDLINE_PATTERNS +from tools import approval_context # ------------------------------------------------------------------------- @@ -688,7 +681,9 @@ def test_approvals_mode_off_cannot_bypass_hardline(clean_session, monkeypatch, t """config approvals.mode=off (yolo-equivalent) must not bypass hardline.""" # _get_approval_mode() reads from hermes config; simplest path: monkeypatch the helper. import tools.approval as approval_mod - monkeypatch.setattr(approval_mod, "_get_approval_mode", lambda: "off") + from tools import approval_context + from tools import approval_context + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "off") result = check_all_command_guards("rm -rf /", "local") assert result["approved"] is False @@ -699,7 +694,7 @@ def test_cron_approve_mode_cannot_bypass_hardline(clean_session, monkeypatch): """Cron sessions with cron_mode=approve must not bypass hardline.""" monkeypatch.setenv("HERMES_CRON_SESSION", "1") import tools.approval as approval_mod - monkeypatch.setattr(approval_mod, "_get_cron_approval_mode", lambda: "approve") + monkeypatch.setattr(approval_context, "_get_cron_approval_mode", lambda: "approve") result = check_all_command_guards("rm -rf /", "local") assert result["approved"] is False diff --git a/tests/tools/test_mcp_elicitation.py b/tests/tools/test_mcp_elicitation.py index 9af33f9589..dd5fe9d36b 100644 --- a/tests/tools/test_mcp_elicitation.py +++ b/tests/tools/test_mcp_elicitation.py @@ -73,7 +73,7 @@ class TestElicitationHandlerFormMode: {"properties": {"approved": {"type": "boolean"}}}, ) - with patch("tools.approval.request_elicitation_consent", return_value="accept"): + with patch("tools.approval_prompt.request_elicitation_consent", return_value="accept"): result = asyncio.run(handler(context=None, params=params)) assert isinstance(result, ElicitResult) @@ -111,7 +111,7 @@ class TestElicitationHandlerFormMode: ) return "decline" - with patch("tools.approval.request_elicitation_consent", _capture): + with patch("tools.approval_prompt.request_elicitation_consent", _capture): asyncio.run(handler(context=None, params=params)) assert "card_number" in (captured.get("description") or ""), captured @@ -124,7 +124,7 @@ class TestElicitationHandlerFormMode: handler = ElicitationHandler("pay", {"timeout": 5}) params = _form_params() - with patch("tools.approval.request_elicitation_consent", return_value="cancel"): + with patch("tools.approval_prompt.request_elicitation_consent", return_value="cancel"): result = asyncio.run(handler(context=None, params=params)) assert result.action == "cancel" @@ -139,7 +139,7 @@ class TestElicitationHandlerFailureModes: # If the handler tried to prompt, this would raise AssertionError # because the side_effect treats the call as a test failure. with patch( - "tools.approval.request_elicitation_consent", + "tools.approval_prompt.request_elicitation_consent", side_effect=AssertionError("URL mode must not prompt"), ): result = asyncio.run(handler(context=None, params=params)) @@ -152,7 +152,7 @@ class TestElicitationHandlerFailureModes: params = _form_params() with patch( - "tools.approval.request_elicitation_consent", + "tools.approval_prompt.request_elicitation_consent", side_effect=RuntimeError("approval system blew up"), ): result = asyncio.run(handler(context=None, params=params)) @@ -178,7 +178,7 @@ class TestElicitationHandlerFailureModes: _t.sleep(2) return "accept" - with patch("tools.approval.request_elicitation_consent", side_effect=stall): + with patch("tools.approval_prompt.request_elicitation_consent", side_effect=stall): result = asyncio.run(handler(context=None, params=params)) assert result.action == "cancel" @@ -242,7 +242,7 @@ class TestElicitationHandlerContextBridge: handler = ElicitationHandler("pay", {"timeout": 5}, owner=owner) params = _form_params() - with patch("tools.approval.request_elicitation_consent", side_effect=fake_consent): + with patch("tools.approval_prompt.request_elicitation_consent", side_effect=fake_consent): result = asyncio.run(handler(context=None, params=params)) assert result.action == "accept" @@ -259,7 +259,7 @@ class TestElicitationHandlerContextBridge: handler = ElicitationHandler("pay", {"timeout": 5}, owner=None) params = _form_params() - with patch("tools.approval.request_elicitation_consent", return_value="accept") as m: + with patch("tools.approval_prompt.request_elicitation_consent", return_value="accept") as m: result = asyncio.run(handler(context=None, params=params)) assert result.action == "accept" @@ -275,7 +275,7 @@ class TestElicitationHandlerContextBridge: handler = ElicitationHandler("pay", {"timeout": 5}, owner=owner) params = _form_params() - with patch("tools.approval.request_elicitation_consent", return_value="decline"): + with patch("tools.approval_prompt.request_elicitation_consent", return_value="decline"): result = asyncio.run(handler(context=None, params=params)) assert result.action == "decline" @@ -321,7 +321,7 @@ class TestRequestedSchemaFieldName: ) return "decline" - with patch("tools.approval.request_elicitation_consent", _capture): + with patch("tools.approval_prompt.request_elicitation_consent", _capture): asyncio.run(handler(context=None, params=params)) # An empty schema renders the generic "Approval requested by ..." diff --git a/tests/tools/test_modal_sandbox_fixes.py b/tests/tools/test_modal_sandbox_fixes.py index 9d9f7d9825..9baed6aa63 100644 --- a/tests/tools/test_modal_sandbox_fixes.py +++ b/tests/tools/test_modal_sandbox_fixes.py @@ -14,6 +14,7 @@ import os import sys from pathlib import Path import pytest +from tools import approval_context # Ensure repo root is importable _repo_root = Path(__file__).resolve().parent.parent.parent @@ -371,6 +372,7 @@ class TestDockerHostBindApproval: def test_should_skip_container_guards(self): """Docker skips only when isolated; other sandboxes always skip.""" import tools.approval as A + from tools import approval_context assert A._should_skip_container_guards("docker", has_host_access=False) is True assert A._should_skip_container_guards("docker", has_host_access=True) is False assert A._should_skip_container_guards("modal", has_host_access=True) is True @@ -413,7 +415,7 @@ class TestDockerHostBindApproval: monkeypatch.setattr(A, "_permanent_approved", set()) monkeypatch.setattr(A, "_session_approved", {}) monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual") def test_host_bound_docker_requires_approval(self, monkeypatch): """Host-bound Docker dangerous command escalates instead of bypassing.""" diff --git a/tests/tools/test_request_tool_approval.py b/tests/tools/test_request_tool_approval.py index 5f2c8ef401..94b1292b75 100644 --- a/tests/tools/test_request_tool_approval.py +++ b/tests/tools/test_request_tool_approval.py @@ -9,6 +9,7 @@ the gateway submit_pending path, cron_mode, and fail-closed timeouts. import pytest import tools.approval as approval +from tools import approval_context from tools.approval import request_tool_approval @@ -61,14 +62,14 @@ class TestRequestToolApproval: "invoke_hook", lambda hook_name, **kwargs: events.append((hook_name, kwargs)) or [], ) - tokens = approval.set_current_observability_context( + tokens = approval_context.set_current_observability_context( turn_id="turn-1", tool_call_id="call-1", ) try: res = request_tool_approval("terminal", "curl PUT to external API") finally: - approval.reset_current_observability_context(tokens) + approval_context.reset_current_observability_context(tokens) assert res["approved"] is False assert "denied" in res["message"].lower() assert res["pattern_key"].startswith("plugin_rule:") @@ -100,7 +101,7 @@ class TestRequestToolApproval: monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False) monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) monkeypatch.setattr(approval, "_is_cron_approval_context", lambda: True) - monkeypatch.setattr(approval, "_get_cron_approval_mode", lambda: "deny") + monkeypatch.setattr(approval_context, "_get_cron_approval_mode", lambda: "deny") res = request_tool_approval("terminal", "smtp send") assert res["approved"] is False assert "cron" in res["message"].lower() @@ -109,7 +110,7 @@ class TestRequestToolApproval: monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False) monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) monkeypatch.setattr(approval, "_is_cron_approval_context", lambda: True) - monkeypatch.setattr(approval, "_get_cron_approval_mode", lambda: "approve") + monkeypatch.setattr(approval_context, "_get_cron_approval_mode", lambda: "approve") res = request_tool_approval("terminal", "smtp send") assert res["approved"] is True diff --git a/tests/tools/test_single_query_approval_mode.py b/tests/tools/test_single_query_approval_mode.py index 5a9ae63db5..878a363792 100644 --- a/tests/tools/test_single_query_approval_mode.py +++ b/tests/tools/test_single_query_approval_mode.py @@ -15,13 +15,10 @@ that decision deterministic and explicit. import pytest import tools.approval as approval_module +from tools import approval_context from gateway.session_context import clear_session_vars, reset_session_vars, set_session_vars -from tools.approval import ( - _get_single_query_approval_mode, - check_all_command_guards, - check_dangerous_command, - detect_dangerous_command, -) +from tools.approval import check_all_command_guards, check_dangerous_command, detect_dangerous_command +from tools.approval_context import _get_single_query_approval_mode @pytest.fixture(autouse=True) @@ -143,7 +140,7 @@ class TestSingleQueryDenyMode: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="deny"): result = check_dangerous_command("rm -rf /tmp/stuff", "local") assert not result["approved"] assert "BLOCKED" in result["message"] @@ -157,7 +154,7 @@ class TestSingleQueryDenyMode: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="deny"): result = check_dangerous_command("ls -la", "local") assert result["approved"] @@ -169,7 +166,7 @@ class TestSingleQueryDenyMode: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="deny"): result = check_dangerous_command("rm -rf /tmp/stuff", "local") assert not result["approved"] assert "dangerous" in result["message"].lower() or "delete" in result["message"].lower() @@ -186,7 +183,7 @@ class TestSingleQueryApproveMode: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="approve"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="approve"): result = check_dangerous_command("rm -rf /tmp/stuff", "local") assert result["approved"] @@ -206,7 +203,7 @@ class TestSingleQueryDenyModeAllGuards: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="deny"): result = check_all_command_guards("rm -rf /tmp/stuff", "local") assert not result["approved"] assert "BLOCKED" in result["message"] @@ -220,7 +217,7 @@ class TestSingleQueryDenyModeAllGuards: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="deny"): result = check_all_command_guards("echo hello", "local") assert result["approved"] @@ -232,7 +229,7 @@ class TestSingleQueryDenyModeAllGuards: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="approve"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="approve"): result = check_all_command_guards("rm -rf /tmp/stuff", "local") assert result["approved"] @@ -254,7 +251,7 @@ class TestSingleQueryDenyModeAllGuards: "summary": "homograph url", } with ( - mock_patch("tools.approval._get_single_query_approval_mode", return_value="deny"), + mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="deny"), mock_patch("tools.approval.detect_dangerous_command", return_value=(False, None, None)), mock_patch("tools.tirith_security.check_command_security", @@ -281,7 +278,7 @@ class TestSingleQueryExecuteCode: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="deny"): result = approval_module.check_execute_code_guard("import os", "local") assert not result["approved"] assert result["outcome"] == "blocked" @@ -295,7 +292,7 @@ class TestSingleQueryExecuteCode: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="approve"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="approve"): result = approval_module.check_execute_code_guard("import os", "local") assert result["approved"] @@ -306,7 +303,7 @@ class TestSingleQueryExecuteCode: monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual") result = approval_module.check_execute_code_guard("import os", "local") assert result["approved"] @@ -327,7 +324,7 @@ class TestSingleQueryModeInteractions: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="deny"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="deny"): result = check_dangerous_command("rm -rf /", "docker") assert result["approved"] @@ -346,7 +343,7 @@ class TestSingleQueryModeInteractions: from unittest.mock import patch as mock_patch with ( mock_patch.object(approval_module, "_YOLO_MODE_FROZEN", True), - mock_patch("tools.approval._get_single_query_approval_mode", return_value="deny"), + mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="deny"), ): result = check_dangerous_command("rm -rf /tmp/stuff", "local") assert result["approved"] @@ -359,6 +356,6 @@ class TestSingleQueryModeInteractions: monkeypatch.delenv("HERMES_YOLO_MODE", raising=False) from unittest.mock import patch as mock_patch - with mock_patch("tools.approval._get_single_query_approval_mode", return_value="approve"): + with mock_patch("tools.approval_context._get_single_query_approval_mode", return_value="approve"): result = check_all_command_guards("rm -rf /", "local") assert not result["approved"] \ No newline at end of file diff --git a/tests/tools/test_smart_approval_injection.py b/tests/tools/test_smart_approval_injection.py index 7dc7b8d4ba..955f093416 100644 --- a/tests/tools/test_smart_approval_injection.py +++ b/tests/tools/test_smart_approval_injection.py @@ -14,11 +14,7 @@ Defenses under test: import unittest from unittest.mock import MagicMock, patch -from tools.approval import ( - _strip_line_comment, - _strip_shell_comments, - _smart_approve, -) +from tools.approval_smart import _strip_line_comment, _strip_shell_comments, _smart_approve # ── _strip_line_comment ────────────────────────────────────────────────── diff --git a/tests/tools/test_smart_approval_policy.py b/tests/tools/test_smart_approval_policy.py index 59683afc7f..cf3fef7115 100644 --- a/tests/tools/test_smart_approval_policy.py +++ b/tests/tools/test_smart_approval_policy.py @@ -17,7 +17,7 @@ Inspired by ChatGPT Work's customizable auto-review guardian policy. import unittest from unittest.mock import MagicMock, patch -from tools.approval import _get_smart_policy, _smart_approve +from tools.approval_smart import _get_smart_policy, _smart_approve POLICY_TEXT = "Always ESCALATE commands that modify anything under /etc." @@ -39,13 +39,13 @@ def _messages_from(mock_call_llm): class TestGetSmartPolicy(unittest.TestCase): """Unit tests for the config reader.""" - @patch("tools.approval._get_approval_config") + @patch("tools.approval_context._get_approval_config") def test_missing_key_returns_empty(self, mock_cfg): mock_cfg.return_value = {"mode": "smart"} assert _get_smart_policy() == "" - @patch("tools.approval._get_approval_config") + @patch("tools.approval_context._get_approval_config") def test_policy_text_is_stripped(self, mock_cfg): mock_cfg.return_value = {"smart_policy": f" {POLICY_TEXT}\n"} assert _get_smart_policy() == POLICY_TEXT @@ -57,11 +57,11 @@ class TestSmartApprovePolicyInjection(unittest.TestCase): Follows the mocking pattern of test_smart_approval_injection.py: ``call_llm`` is patched at its source module (``agent.auxiliary_client``) because _smart_approve imports it lazily inside the function. The - config read is isolated by patching ``tools.approval._get_approval_config`` + config read is isolated by patching ``tools.approval_context._get_approval_config`` so tests never touch a real config.yaml. """ - @patch("tools.approval._get_approval_config") + @patch("tools.approval_context._get_approval_config") @patch("agent.auxiliary_client.call_llm") def test_empty_policy_leaves_prompts_unchanged(self, mock_call_llm, mock_cfg): """With no policy configured, prompts must be byte-identical to the @@ -81,7 +81,7 @@ class TestSmartApprovePolicyInjection(unittest.TestCase): assert "Additional policy rules from the operator" not in sys_content - @patch("tools.approval._get_approval_config") + @patch("tools.approval_context._get_approval_config") @patch("agent.auxiliary_client.call_llm") def test_policy_never_in_user_message(self, mock_call_llm, mock_cfg): """The policy is trusted; the user message carries untrusted command @@ -100,7 +100,7 @@ class TestSmartApprovePolicyInjection(unittest.TestCase): assert "rm -rf /etc/nginx" in user_content assert "" in user_content - @patch("tools.approval._get_approval_config") + @patch("tools.approval_context._get_approval_config") @patch("agent.auxiliary_client.call_llm") def test_config_read_failure_does_not_break_approval(self, mock_call_llm, mock_cfg): """If the config reader itself blows up, _smart_approve fails safe.""" @@ -111,7 +111,7 @@ class TestSmartApprovePolicyInjection(unittest.TestCase): @patch("agent.auxiliary_client._get_task_timeout") - @patch("tools.approval._get_approval_config") + @patch("tools.approval_context._get_approval_config") @patch("agent.auxiliary_client.call_llm") def test_smart_approve_passes_explicit_timeout( self, mock_call_llm, mock_cfg, mock_task_timeout @@ -130,7 +130,7 @@ class TestSmartApprovePolicyInjection(unittest.TestCase): assert kwargs.get("timeout") == 42.0 - @patch("tools.approval._get_approval_config") + @patch("tools.approval_context._get_approval_config") @patch("agent.auxiliary_client.call_llm") def test_smart_approve_failure_logs_warning_and_escalates( self, mock_call_llm, mock_cfg diff --git a/tests/tools/test_yolo_mode.py b/tests/tools/test_yolo_mode.py index 74d9de67a2..5266d9c45d 100644 --- a/tests/tools/test_yolo_mode.py +++ b/tests/tools/test_yolo_mode.py @@ -4,19 +4,11 @@ import os import pytest import tools.approval as approval_module +from tools import approval_context import tools.tirith_security -from tools.approval import ( - check_all_command_guards, - check_dangerous_command, - detect_dangerous_command, - disable_session_yolo, - enable_session_yolo, - is_approval_bypass_active_for_session, - is_session_yolo_enabled, - reset_current_session_key, - set_current_session_key, -) +from tools.approval import check_all_command_guards, check_dangerous_command, detect_dangerous_command, disable_session_yolo, enable_session_yolo, is_approval_bypass_active_for_session, is_session_yolo_enabled +from tools.approval_context import reset_current_session_key, set_current_session_key @pytest.fixture(autouse=True) @@ -176,7 +168,7 @@ class TestYoloMode: def test_bypass_query_uses_the_requested_session(self, monkeypatch): """Backend mode selection must not leak YOLO across sessions.""" monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False) - monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual") + monkeypatch.setattr(approval_context, "_get_approval_mode", lambda: "manual") enable_session_yolo("session-a") diff --git a/tools/approval.py b/tools/approval.py index c6bb88bc5e..ff53ea9c1d 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -7,9 +7,8 @@ behind them. Leaves: ``approval_detection`` (hardline/dangerous patterns), ``app (contextvars, config readers), ``approval_floors`` (pre-gate blocks, allowlist match), ``approval_prompt`` (CLI prompt, plugin transports, MCP elicitation), ``approval_gateway_wait`` (blocking gateway round-trip), ``approval_smart`` (guardian LLM), ``approval_human_wait``. -Every private name is re-exported here so ``from tools.approval import X`` and -``patch("tools.approval.X")`` keep working; leaves call back through ``tools.approval`` at -call time for the same reason. +Leaves read facade-owned state (``_lock``, queues, denial breaker) back through ``tools.approval`` at +call time; sibling-defined names are imported from their defining module. """ from dataclasses import dataclass @@ -21,39 +20,23 @@ import threading from typing import Optional from utils import env_var_enabled, is_truthy_value -from tools.approval_context import ( # noqa: F401 -- re-exported for callers/tests - _approval_session_key, _approval_turn_id, _approval_tool_call_id, set_hermes_interactive_context, - reset_hermes_interactive_context, _is_interactive_cli, _fire_approval_hook, set_current_session_key, - reset_current_session_key, set_current_observability_context, reset_current_observability_context, - get_current_session_key, _get_session_platform, _is_cron_approval_context, _UNATTENDED_APPROVAL_PLATFORMS, - _is_unattended_platform_approval_context, _is_single_query_approval_context, _is_gateway_approval_context, - _resolve_cli_approval_callback, _should_fall_through_to_cli_approval, _normalize_approval_mode, - _get_approval_config, _get_approval_mode, _get_approval_timeout, _get_cron_approval_mode, - _get_single_query_approval_mode, _get_unattended_approval_mode, _tirith_fail_open, _get_approval_transport_config, +from tools import approval_context +from tools.approval_context import ( + _get_session_platform, _is_cron_approval_context, + _is_gateway_approval_context, _is_interactive_cli, _is_single_query_approval_context, + _is_unattended_platform_approval_context, _resolve_cli_approval_callback, _should_fall_through_to_cli_approval, + _tirith_fail_open, get_current_session_key, ) -from tools.approval_prompt import ( # noqa: F401 -- re-exported for callers/tests - prompt_dangerous_approval, get_plugin_manager, _present_with_selected_transport, _transport_choice, - request_elicitation_consent, +from tools.approval_detection import ( + _approval_key_aliases, _check_sudo_stdin_guard, detect_dangerous_command, detect_hardline_command, ) -from tools.approval_floors import ( # noqa: F401 -- re-exported for callers/tests - _match_user_deny_rule, _user_deny_block_result, _save_blocked_payload, _hardline_block_result, - _sudo_stdin_block_result, _has_allowlist_shell_operator, _command_matches_permanent_allowlist, +from tools.approval_floors import ( + _command_matches_permanent_allowlist, _hardline_block_result, _match_user_deny_rule, _sudo_stdin_block_result, + _user_deny_block_result, ) -from tools.approval_detection import ( # noqa: F401 -- re-exported for callers/tests - HARDLINE_PATTERNS, _check_sudo_stdin_guard, detect_hardline_command, _approval_key_aliases, - _rewrite_resolved_user_home, _rewrite_resolved_hermes_home, _MAX_SEPARATOR_FREE_COMMAND_CHARS, - _PARSER_LIMIT_DESCRIPTION, _MALFORMED_EXEC_DESCRIPTION, _bash_exec_payload, _read_shell_word, - _deobfuscate_shell_word_for_detection, _iter_shell_command_starts, _command_detection_variants, - detect_dangerous_command, -) -from tools.approval_human_wait import ( # noqa: F401 -- re-exported for callers/tests - _human_wait_lock, _human_wait_states, _HUMAN_WAIT_MAX_SESSIONS, HUMAN_WAIT_MARGIN_S, human_wait_ceiling, - human_wait_window, human_wait_seconds, -) -from tools.approval_smart import ( # noqa: F401 -- re-exported for callers/tests - _strip_shell_comments, _strip_line_comment, _get_smart_policy, _smart_approve, _smart_verdict, -) -from tools.approval_gateway_wait import _ApprovalEntry, _await_gateway_decision # noqa: F401 -- re-exported +from tools.approval_gateway_wait import _await_gateway_decision +from tools.approval_prompt import _present_with_selected_transport, _transport_choice, prompt_dangerous_approval +from tools.approval_smart import _smart_verdict logger = logging.getLogger(__name__) @@ -83,7 +66,7 @@ _DENIAL_TALLY_MAX_SESSIONS = 256 def _get_denial_breaker_threshold() -> int: """``approvals.denial_breaker_threshold``: default 3; 0 or negative disables.""" try: - return int(_get_approval_config().get("denial_breaker_threshold", 3)) + return int(approval_context._get_approval_config().get("denial_breaker_threshold", 3)) except (ValueError, TypeError): return 3 @@ -373,7 +356,7 @@ def is_approval_bypass_active_for_session(session_key: str) -> bool: """Canonical three-source bypass check: process ``--yolo`` (frozen at import), the session-scoped gateway ``/yolo`` toggle, ``approvals.mode: off``. Pure bypass sub-expression only — hardline blocklist / permanent allowlist are the caller's job.""" - return (_YOLO_MODE_FROZEN or is_session_yolo_enabled(session_key) or _get_approval_mode() == "off") + return (_YOLO_MODE_FROZEN or is_session_yolo_enabled(session_key) or approval_context._get_approval_mode() == "off") def is_approval_bypass_active() -> bool: @@ -458,8 +441,8 @@ class _Unattended: trust: str # execute_code: "approve only if {trust}" def mode(self) -> str: - # Looked up through the module at call time so tests patching the getters keep working. - return globals()[f"_get_{self.name}_approval_mode"]() + # Looked up on the defining module at call time so tests patching the getters keep working. + return getattr(approval_context, f"_get_{self.name}_approval_mode")() def block_message(self, subject: str, *, noun: str, advice: str) -> str: return (f"BLOCKED: {subject} but {self.clause}. {advice} To allow {noun} {self.scope}, set " @@ -772,10 +755,10 @@ def _human_decision(spec: _GateSpec, *, command: str, description: str, prompt_description = redact_sensitive_text(description) hook_kwargs = dict(command=prompt_command, description=prompt_description, pattern_key=pattern_key, pattern_keys=list(pattern_keys), session_key=session_key, surface="cli") - _fire_approval_hook("pre_approval_request", **hook_kwargs) + approval_context._fire_approval_hook("pre_approval_request", **hook_kwargs) choice = prompt_dangerous_approval(prompt_command, prompt_description, allow_permanent=allow_permanent, smart_denied=smart_denied, approval_callback=approval_callback) - _fire_approval_hook("post_approval_response", **hook_kwargs, choice=choice) + approval_context._fire_approval_hook("post_approval_response", **hook_kwargs, choice=choice) if choice == "timeout": return deny(spec.cli_timeout, "timeout") if choice == "deny": @@ -999,7 +982,7 @@ def check_all_command_guards(command: str, env_type: str, if blocked is not None: return blocked - approval_mode = _get_approval_mode() + approval_mode = approval_context._get_approval_mode() if _yolo_active() or approval_mode == "off": return _approved() if _command_matches_permanent_allowlist(command): @@ -1077,7 +1060,7 @@ def check_execute_code_guard(code: str, env_type: str, has_host_access: bool = F return _approved() if _should_skip_container_guards(env_type, has_host_access=has_host_access): return _approved() - approval_mode = _get_approval_mode() + approval_mode = approval_context._get_approval_mode() if _yolo_active() or approval_mode == "off": return _approved() diff --git a/tools/approval_context.py b/tools/approval_context.py index 150522f747..7bcaca32c8 100644 --- a/tools/approval_context.py +++ b/tools/approval_context.py @@ -2,7 +2,7 @@ Session identity and observability contextvars, the interactive/gateway/cron/ unattended predicates, and the ``approvals.*`` config readers used by every -gate in :mod:`tools.approval` (which re-exports all of them). +gate in :mod:`tools.approval`. """ import contextvars @@ -169,8 +169,7 @@ def _is_gateway_approval_context() -> bool: human who can resolve it (#37284, 87509). Their dangerous-command handling is governed by ``approvals.unattended_mode`` config (default deny), mirroring cron. """ - from tools import approval as _a - if _a._is_cron_approval_context() or _is_unattended_platform_approval_context(): + if _is_cron_approval_context() or _is_unattended_platform_approval_context(): return False return env_var_enabled("HERMES_GATEWAY_SESSION") or bool(_get_session_platform()) @@ -228,14 +227,13 @@ def _get_approval_config() -> dict: def _get_approval_mode() -> str: """Return 'manual', 'smart', or 'off' (a hosted-room policy overrides config).""" - from tools import approval as _a try: from gateway.hosted_room_execution_policy import current_room_execution_policy if (room_policy := current_room_execution_policy()) is not None: return room_policy.approval_mode except Exception: pass - return _a._normalize_approval_mode(_a._get_approval_config().get("mode", "manual")) + return _normalize_approval_mode(_get_approval_config().get("mode", "manual")) def _get_approval_timeout() -> int: @@ -245,9 +243,8 @@ def _get_approval_timeout() -> int: overflows ``time_t`` inside ``Thread.join`` / ``Lock.acquire`` on macOS and crashed every parallel tool batch; clamping at the single config-read site keeps every consumer platform-safe at once.""" - from tools import approval as _a try: - raw = int(_a._get_approval_config().get("timeout", 300)) + raw = int(_get_approval_config().get("timeout", 300)) except (ValueError, TypeError): return 300 try: diff --git a/tools/approval_detection.py b/tools/approval_detection.py index 2649da6ee8..1ef703670a 100644 --- a/tools/approval_detection.py +++ b/tools/approval_detection.py @@ -1,8 +1,7 @@ """Dangerous-command detection: normalization, tokenizing, and pattern tables. Pure command classification for :mod:`tools.approval` — no approval state, config reads, or -prompting live here. ``tools.approval`` re-exports every public and private name so -``from tools.approval import X`` and ``patch("tools.approval.X")`` keep working. +prompting live here. """ import functools @@ -178,9 +177,7 @@ def detect_hardline_command(command: str) -> tuple: _, malformed_grep = _grep_safe_detection_variant(normalized) if malformed_grep: return (True, _MALFORMED_EXEC_DESCRIPTION) - # Call-time lookup through the facade: tests/plugins patch tools.approval._command_detection_variants. - from tools.approval import _command_detection_variants as _variants - for command_variant in _variants(command): + for command_variant in _command_detection_variants(command): variant_lower = command_variant.lower() masked_lower: str | None = None for pattern_re, description, quote_masked in HARDLINE_PATTERNS_COMPILED: @@ -1155,9 +1152,7 @@ def detect_dangerous_command(command: str) -> tuple: return (True, _PARSER_LIMIT_DESCRIPTION, _PARSER_LIMIT_DESCRIPTION) if _is_verification_artifact_cleanup(command): return (False, None, None) - # Call-time lookup through the facade: tests/plugins patch tools.approval._command_detection_variants. - from tools.approval import _command_detection_variants as _variants - for command_variant in _variants(command): + for command_variant in _command_detection_variants(command): command_lower = command_variant.lower() for pattern_re, description in DANGEROUS_PATTERNS_COMPILED: if pattern_re.search(command_lower): diff --git a/tools/approval_floors.py b/tools/approval_floors.py index 5a648fc910..17f8d4e408 100644 --- a/tools/approval_floors.py +++ b/tools/approval_floors.py @@ -4,7 +4,7 @@ Unconditional blocks (hardline, ``sudo -S`` password piping, the user's own ``approvals.deny`` globs) and the permanent command allowlist match. All of them run BEFORE yolo / ``approvals.mode: off`` / cron approve-mode; the allowlist runs after. Session state stays in ``tools.approval`` and is read -through it at call time so tests that rebind it keep working. +through it at call time. """ import contextlib @@ -13,7 +13,9 @@ import logging import re import time import uuid -from tools.approval_detection import _MALFORMED_EXEC_DESCRIPTION, _PARSER_LIMIT_DESCRIPTION +from tools import approval_context as _ctx +from tools.approval_detection import ( + _MALFORMED_EXEC_DESCRIPTION, _PARSER_LIMIT_DESCRIPTION, _command_detection_variants) logger = logging.getLogger("tools.approval") @@ -25,15 +27,14 @@ def _match_user_deny_rule(command: str) -> str | None: yolo"). Case-insensitive, run over the same normalized/deobfuscated variants the dangerous-pattern detector uses so quoting tricks (``r\\m``, ``git st""atus``) can't sidestep a rule.""" - from tools import approval as _a try: - deny_patterns = _a._get_approval_config().get("deny") or [] + deny_patterns = _ctx._get_approval_config().get("deny") or [] except Exception: return None globs = [p.strip() for p in deny_patterns if isinstance(p, str) and p.strip()] if not globs: return None - for command_variant in _a._command_detection_variants(command): + for command_variant in _command_detection_variants(command): candidate = command_variant.lower().strip() for pattern in globs: if fnmatch.fnmatchcase(candidate, pattern.lower()): @@ -96,7 +97,6 @@ _RECOVERY_PREFIX = ( def _hardline_block_result(description: str, command: str = "") -> dict: """Build the standard block result for a hardline match.""" - from tools import approval as _a message = ( f"BLOCKED (hardline): {description}. " "This command is on the unconditional blocklist and cannot " @@ -107,7 +107,7 @@ def _hardline_block_result(description: str, command: str = "") -> dict: # The parser-limit block is almost always a giant inline payload, not a forbidden operation, and is typically # followed by blind rephrase retries — point at the saved script (or the write_file recipe). if description in (_PARSER_LIMIT_DESCRIPTION, _MALFORMED_EXEC_DESCRIPTION): - saved = _a._save_blocked_payload(command) if command else None + saved = _save_blocked_payload(command) if command else None if saved: message += _RECOVERY_PREFIX + ( f"Your command was saved to {saved} — review it, then run: terminal(command=\"bash {saved}\"). " @@ -192,7 +192,7 @@ def _command_matches_permanent_allowlist(command: str) -> bool: shell-style wildcards like ``podman *``.""" from tools import approval as _a command = (command or "").strip() - if not command or _a._has_allowlist_shell_operator(command): + if not command or _has_allowlist_shell_operator(command): return False with _a._lock: patterns = tuple(_a._permanent_approved) diff --git a/tools/approval_gateway_wait.py b/tools/approval_gateway_wait.py index f75044661b..1a127cb1bf 100644 --- a/tools/approval_gateway_wait.py +++ b/tools/approval_gateway_wait.py @@ -7,7 +7,7 @@ threads (parallel subagents, execute_code RPC handlers) can block concurrently — each gets its own ``threading.Event``; ``/approve`` resolves the oldest, ``/approve all`` every pending entry. Queue state (``_gateway_queues``, ``_lock``) is owned by ``tools.approval`` and reached through that module at -call time so tests patching it keep working. +call time. """ import logging @@ -16,7 +16,8 @@ import time import uuid from tools.interrupt import is_interrupted -from tools.approval_human_wait import activity_heartbeat +from tools import approval_context as _ctx +from tools.approval_human_wait import activity_heartbeat, human_wait_window logger = logging.getLogger("tools.approval") @@ -48,9 +49,7 @@ def _poll_event(event: threading.Event, session_key: str, *, interrupt_log: str) The per-thread interrupt flag carries no stable machine-checkable cause, so a fail-closed deny preserves the historical semantics; changing this needs a dedicated interrupt-cause channel, not string matching.""" - from tools.approval import _get_approval_timeout, human_wait_window - - deadline = time.monotonic() + max(_get_approval_timeout(), 0) + deadline = time.monotonic() + max(_ctx._get_approval_timeout(), 0) heartbeat = activity_heartbeat("waiting for user approval") with human_wait_window(session_key): while True: @@ -71,8 +70,7 @@ def _poll_event(event: threading.Event, session_key: str, *, interrupt_log: str) def _finish(payload: dict, resolved: bool, choice: str | None, reason, **extra) -> dict: """Fire the post hook and build the decision dict. Unresolved (timeout) and a None choice both mean the user never answered.""" - from tools.approval import _fire_approval_hook - _fire_approval_hook("post_approval_response", **payload, + _ctx._fire_approval_hook("post_approval_response", **payload, choice="timeout" if not resolved else (choice or "timeout"), **extra) return {"resolved": resolved, "choice": choice, "reason": reason, **extra} @@ -86,8 +84,7 @@ def _await_coalesced_leader(session_key: str, leader, payload: dict): returns ``None``: single-use consent covers only the leader's execution, so the caller must issue a fresh prompt. Hooks fire with ``coalesced=True`` so observers see the follower's lifecycle without a duplicate prompt.""" - from tools.approval import _fire_approval_hook - _fire_approval_hook("pre_approval_request", **payload, coalesced=True) + _ctx._fire_approval_hook("pre_approval_request", **payload, coalesced=True) state = _poll_event(leader.event, session_key, interrupt_log="Coalesced approval wait interrupted by user signal — " "returning deny for session %s") @@ -151,14 +148,14 @@ def _await_gateway_decision(session_key: str, notify_cb, approval_data: dict, *, _approval._gateway_queues.pop(session_key, None) # Plugins hear about the request before the gateway does (real-time observers). - _approval._fire_approval_hook("pre_approval_request", **payload) + _ctx._fire_approval_hook("pre_approval_request", **payload) # Bridges sync agent thread → async gateway. try: notify_cb(dict(entry.data)) except Exception as exc: logger.warning("Gateway approval notify failed: %s", exc) _drop_entry() - _approval._fire_approval_hook("post_approval_response", **payload, choice="notify_failed") + _ctx._fire_approval_hook("post_approval_response", **payload, choice="notify_failed") return {"resolved": False, "choice": None, "notify_failed": True} state = _poll_event(entry.event, session_key, diff --git a/tools/approval_human_wait.py b/tools/approval_human_wait.py index 4020d3aae0..37ad9942e4 100644 --- a/tools/approval_human_wait.py +++ b/tools/approval_human_wait.py @@ -55,8 +55,8 @@ def human_wait_ceiling() -> float: holding ``_human_wait_lock`` — it reads the config cache. ``_get_approval_timeout`` caps at ``agent.deadline.MAX_SAFE_TIMEOUT_S`` so the value is always safe for ``Lock.acquire(timeout=...)`` / ``Thread.join(timeout=...)``.""" - from tools.approval import _get_approval_timeout - return float(_get_approval_timeout()) + HUMAN_WAIT_MARGIN_S + from tools import approval_context + return float(approval_context._get_approval_timeout()) + HUMAN_WAIT_MARGIN_S def _clamped_window_seconds(started: float, now: float, ceiling: float) -> float: @@ -87,8 +87,8 @@ def _human_wait_state(session_key: str) -> _HumanWaitState: def _resolve_key(session_key: str | None) -> str: if session_key is not None: return session_key - from tools.approval import get_current_session_key - return get_current_session_key() + from tools import approval_context + return approval_context.get_current_session_key() def activity_heartbeat(label: str): diff --git a/tools/approval_prompt.py b/tools/approval_prompt.py index d0fcf5719f..c216d9b0ef 100644 --- a/tools/approval_prompt.py +++ b/tools/approval_prompt.py @@ -10,6 +10,7 @@ import logging import os import sys import threading +from tools import approval_context as _ctx, approval_gateway_wait as _gw from tools.approval_human_wait import activity_heartbeat, human_wait_window from tools.interrupt import is_interrupted @@ -37,9 +38,8 @@ def prompt_dangerous_approval(command: str, description: str, timeout_seconds: i See #81887. """ - from tools import approval as _a if timeout_seconds is None: - timeout_seconds = _a._get_approval_timeout() + timeout_seconds = _ctx._get_approval_timeout() # Everything below is a human prompt (callback panel or input() fallback, both bounded by the approval deadline): # record it as human-wait time so the concurrent batch deadline excludes it. # See #79719. @@ -172,13 +172,12 @@ def _present_with_selected_transport(*, command: str, description: str, pattern_ persistence, timeout, and final authorization stay host-owned. A failed transport reaches a built-in surface only under the explicit ``transport_fallback: builtin`` opt-in.""" - from tools import approval as _a - name, fallback = _a._get_approval_transport_config() + name, fallback = _ctx._get_approval_transport_config() if name == "builtin": return {"selected": False} try: - registered = _a.get_plugin_manager().get_approval_transport(name) + registered = get_plugin_manager().get_approval_transport(name) except Exception: # Plugin/discovery exception text may contain plugin-owned secrets. logger.warning("Could not resolve selected approval transport %r", name) @@ -191,7 +190,7 @@ def _present_with_selected_transport(*, command: str, description: str, pattern_ from agent.redact import redact_sensitive_text from hermes_cli.approval_transport import ApprovalRequest, invoke_approval_transport - timeout_seconds = _a._get_approval_timeout() + timeout_seconds = _ctx._get_approval_timeout() request = ApprovalRequest.create( command=redact_sensitive_text(command, force=True), description=redact_sensitive_text(description, force=True), pattern_key=pattern_key, @@ -208,7 +207,7 @@ def _present_with_selected_transport(*, command: str, description: str, pattern_ pattern_keys=list(pattern_keys), session_key=session_key, surface=f"transport:{name}", request_id=request.request_id, request_digest=request.digest, ) - _a._fire_approval_hook("pre_approval_request", **hook_kwargs) + _ctx._fire_approval_hook("pre_approval_request", **hook_kwargs) with human_wait_window(session_key): result = invoke_approval_transport( registered.present, request, timeout_seconds=timeout_seconds, @@ -216,7 +215,7 @@ def _present_with_selected_transport(*, command: str, description: str, pattern_ is_interrupted=is_interrupted, ) hook_choice = result.choice if result.failure is None else f"transport_{result.failure}" - _a._fire_approval_hook("post_approval_response", **hook_kwargs, choice=hook_choice) + _ctx._fire_approval_hook("post_approval_response", **hook_kwargs, choice=hook_choice) return _attempt(name, result.choice, result.failure, fallback) @@ -235,7 +234,7 @@ def _transport_choice(attempt: dict, *, pattern_key: str, description: str): attempt.get("name"), failure) return None, None from tools import approval as _a - breaker_addendum = _a._denial_breaker_addendum(_a.get_current_session_key()) + breaker_addendum = _a._denial_breaker_addendum(_ctx.get_current_session_key()) return None, _a._denied( f"BLOCKED: Selected approval transport failed ({failure}); the user " "has NOT consented to this action. Do NOT retry this command or " @@ -262,19 +261,19 @@ def request_elicitation_consent(message: str, description: str, *, Returns ``"accept" | "decline" | "cancel"``.""" from tools import approval as _a try: - session_key = _a.get_current_session_key() + session_key = _ctx.get_current_session_key() except Exception as exc: # pragma: no cover -- defensive logger.warning("Elicitation consent: session lookup failed: %s", exc) return "decline" - if _a._is_gateway_approval_context(): + if _ctx._is_gateway_approval_context(): notify_cb = _a._gateway_notify_cb(session_key) if notify_cb is None: logger.warning("Elicitation requested in gateway session %s but no " "notify_cb is registered — failing closed", session_key) return "decline" try: - decision = _a._await_gateway_decision( + decision = _gw._await_gateway_decision( session_key, notify_cb, {"command": message, "description": description, "pattern_key": "mcp_elicitation", "pattern_keys": ["mcp_elicitation"]}, surface=surface) @@ -289,8 +288,8 @@ def request_elicitation_consent(message: str, description: str, *, # allow_permanent=False: elicitation is a per-call confirmation — no pattern to remember. try: - choice = _a.prompt_dangerous_approval(message, description, timeout_seconds=timeout_seconds, - allow_permanent=False) + choice = prompt_dangerous_approval(message, description, timeout_seconds=timeout_seconds, + allow_permanent=False) except Exception as exc: logger.error("Elicitation CLI prompt failed: %s", exc, exc_info=True) return "decline" diff --git a/tools/approval_smart.py b/tools/approval_smart.py index 6c11c377b7..f705daa3ca 100644 --- a/tools/approval_smart.py +++ b/tools/approval_smart.py @@ -10,6 +10,7 @@ Inspired by OpenAI Codex's Smart Approvals guardian subagent. import logging import time +from tools import approval_context as _ctx logger = logging.getLogger("tools.approval") @@ -66,8 +67,7 @@ def _strip_shell_comments(command: str) -> str: def _get_smart_policy() -> str: """Operator rules (``approvals.smart_policy``) appended to the guardian's system prompt.""" - from tools.approval import _get_approval_config - policy = _get_approval_config().get("smart_policy", "") + policy = _ctx._get_approval_config().get("smart_policy", "") return policy.strip() if isinstance(policy, str) else "" @@ -126,7 +126,6 @@ def _smart_verdict(command: str, description: str, pattern_key: str, """Run the guardian LLM with observer hooks; 'approve' | 'deny' | 'escalate'. Redaction is observer-payload preparation, not approval policy: if it fails, skip observability rather than leak raw data or block the LLM decision.""" - from tools import approval as _a try: from agent.redact import redact_sensitive_text payload = { @@ -139,8 +138,8 @@ def _smart_verdict(command: str, description: str, pattern_key: str, logger.debug("Smart approval hook redaction failed: %s", exc) payload = None else: - _a._fire_approval_hook("pre_approval_request", **payload) - verdict = _a._smart_approve(command, description) + _ctx._fire_approval_hook("pre_approval_request", **payload) + verdict = _smart_approve(command, description) if payload is not None and verdict in {"approve", "deny"}: - _a._fire_approval_hook("post_approval_response", **payload, choice=f"smart_{verdict}", decided_by="aux_llm") + _ctx._fire_approval_hook("post_approval_response", **payload, choice=f"smart_{verdict}", decided_by="aux_llm") return verdict diff --git a/tools/browser_extension_router.py b/tools/browser_extension_router.py index 70e62f9418..5e73f006c1 100644 --- a/tools/browser_extension_router.py +++ b/tools/browser_extension_router.py @@ -102,7 +102,7 @@ def route_browser_tool( def current_tool_call_id() -> str: """Active tool_call_id bound by the agent executor, or ``""`` when none.""" try: - from tools.approval import _approval_tool_call_id + from tools.approval_context import _approval_tool_call_id return _approval_tool_call_id.get() or "" except Exception: diff --git a/tools/delegate_tool_dispatch.py b/tools/delegate_tool_dispatch.py index e40d9c163a..5f883a2b8f 100644 --- a/tools/delegate_tool_dispatch.py +++ b/tools/delegate_tool_dispatch.py @@ -214,7 +214,7 @@ def _resolve_async_session_key(parent_agent: Any, origin_ui_session_id: str) -> the key resolves empty; its drain is a positive-ownership filter on the durable session_id (empty would fail closed), so stamp the parent's durable id. """ - from tools.approval import get_current_session_key + from tools.approval_context import get_current_session_key session_key = get_current_session_key(default="") agent_session_id = str(getattr(parent_agent, "session_id", "") or "") with _quiet(None): diff --git a/tools/mcp_tool_sampling.py b/tools/mcp_tool_sampling.py index 11a09c388c..2196ac0f02 100644 --- a/tools/mcp_tool_sampling.py +++ b/tools/mcp_tool_sampling.py @@ -271,7 +271,7 @@ class ElicitationHandler: """Sync consent call replaying the agent's contextvars snapshot when the owner captured one (the recv-loop task does NOT inherit them; gateway-platform detection needs them). ``Context.run`` runs a context once, so it is copied per elicitation.""" - from tools.approval import request_elicitation_consent + from tools.approval_prompt import request_elicitation_consent kwargs = {"timeout_seconds": int(self.timeout), "surface": f"mcp-elicitation/{self.server_name}"} captured = getattr(self.owner, "_pending_call_context", None) if self.owner else None diff --git a/tools/self_repo_guard.py b/tools/self_repo_guard.py index 35ab063966..a747924dff 100644 --- a/tools/self_repo_guard.py +++ b/tools/self_repo_guard.py @@ -11,7 +11,7 @@ from dataclasses import dataclass, field from pathlib import Path from typing import Callable -from tools.approval import ( +from tools.approval_detection import ( _bash_exec_payload, _deobfuscate_shell_word_for_detection, _iter_shell_command_starts, _read_shell_word) diff --git a/tools/terminal_tool_sudo.py b/tools/terminal_tool_sudo.py index 154d9b4f66..7ee30966f2 100644 --- a/tools/terminal_tool_sudo.py +++ b/tools/terminal_tool_sudo.py @@ -180,7 +180,7 @@ def _prompt_for_sudo_password(timeout_seconds: int = 45) -> str: _sudo_cb = _get_sudo_password_callback() if _sudo_cb is not None: try: - from tools.approval import human_wait_window + from tools.approval_human_wait import human_wait_window with human_wait_window(): return _sudo_cb() or "" except Exception: @@ -204,7 +204,7 @@ def _prompt_for_sudo_password(timeout_seconds: int = 45) -> str: print(" Password (hidden): ", end="", flush=True) password_thread = threading.Thread(target=_read_hidden_password, args=(result,), daemon=True) password_thread.start() - from tools.approval import human_wait_window + from tools.approval_human_wait import human_wait_window with human_wait_window(): password_thread.join(timeout=timeout_seconds) if not result["done"]: diff --git a/tui_gateway/methods_config_set.py b/tui_gateway/methods_config_set.py index d676c8f2ce..cd16baaed3 100644 --- a/tui_gateway/methods_config_set.py +++ b/tui_gateway/methods_config_set.py @@ -256,7 +256,7 @@ def _set_yolo(rid, params, key, value, session): from tools.approval import disable_session_yolo, enable_session_yolo, is_session_yolo_enabled raw = _word(value) if scope == "global": - from tools.approval import _normalize_approval_mode + from tools.approval_context import _normalize_approval_mode appr = _load_cfg().get("approvals") appr = appr if isinstance(appr, dict) else {} enable = _BOOL_WORDS.get(raw, _normalize_approval_mode(appr.get("mode", "manual")) != "off") diff --git a/tui_gateway/prompt_turn.py b/tui_gateway/prompt_turn.py index 1322bdbbf0..d098bfc7ed 100644 --- a/tui_gateway/prompt_turn.py +++ b/tui_gateway/prompt_turn.py @@ -443,7 +443,7 @@ def _prepare_turn_input(sid: str, session: dict, st: _TurnRun, text: Any, images fail-closed refusal scope). The config-model sync is skipped under a /model --once override (not pinned as model_override, the sync would clobber it); a model picked mid-turn is applied first so the explicit pick wins over a config change.""" - from tools.approval import set_current_session_key + from tools.approval_context import set_current_session_key scopes = st.scopes scopes.approval = set_current_session_key(session["session_key"]) scopes.session_tokens = _set_session_context(session["session_key"], ui_session_id=sid) @@ -738,7 +738,7 @@ def _finish_turn(sid: str, session: dict, st: _TurnRun) -> None: scopes = st.scopes with contextlib.suppress(Exception): if scopes.approval is not None: - from tools.approval import reset_current_session_key + from tools.approval_context import reset_current_session_key reset_current_session_key(scopes.approval) if scopes.home is not None: reset_hermes_home_override(scopes.home) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 5fa58c8a6e..49266de5ef 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -1673,7 +1673,7 @@ _BOOL_WORDS = { def _load_approval_mode() -> str: """Effective ``approvals.mode`` via the gate's own ``_get_approval_mode`` (a raw re-read missed the managed overlay and ``${VAR}`` expansion).""" - from tools.approval import _get_approval_mode + from tools.approval_context import _get_approval_mode mode = _get_approval_mode() return mode if mode in _APPROVAL_MODES else "manual"