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

simplify(compat): approval — drop 43 facade re-exports + _command_detection_variants late-bind seam, repoint 30 callers + 46 test files

Browse Source 
tools/approval.py no longer re-exports sibling names (approval_context/prompt/floors/detection/
human_wait/smart/gateway_wait); it imports only what it uses. Siblings reference sibling-defined
names directly (module-attribute reads on tools.approval_context so patching the defining module
still works); only facade-owned state (_lock, _gateway_queues, _permanent_approved, _denied,
_denial_breaker_addendum, _gateway_notify_cb) is still read back through tools.approval.
approval_detection calls its own _command_detection_variants instead of late-binding through the facade.
This commit is contained in:
Teknium
2026-09-03 13:49:57 -07:00
parent 7b8c11bcf7
commit 14791b4d4e
74 changed files with 438 additions and 518 deletions
Download Patch File Download Diff File Expand all files Collapse all files
acp_adapter/server.py
+1 -1
View File
@@ -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__)
agent/tool_executor.py
+3 -3
View File
@@ -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:
gateway/hosted_room_execution_policy.py
+2 -1
View File
@@ -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 {}
gateway/platforms/api_server_runs.py
+2 -3
View File
@@ -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.
gateway/run_turn_runner.py
+2 -1
View File
@@ -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)
gateway/slash_commands_goals.py
+1 -1
View File
@@ -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)
hermes_cli/approval_mode.py
+1 -1
View File
@@ -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()
hermes_cli/approvals_suggest.py
+3 -3
View File
@@ -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
hermes_cli/approvals_test.py
+8 -7
View File
@@ -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,
hermes_cli/cli_chat_turn_mixin.py
+1 -1
View File
@@ -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]
hermes_cli/cli_session_mixin.py
+3 -2
View File
@@ -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":
hermes_cli/plugins.py
+4 -5
View File
@@ -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(
hermes_cli/web_server_profiles.py
+2 -4
View File
@@ -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
model_tools.py
+1 -1
View File
@@ -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:
tests/acp/test_approval_isolation.py
+3 -1
View File
@@ -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:
tests/agent/test_auxiliary_explicit_cancellation.py
+2 -5
View File
@@ -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")
tests/cli/test_cli_status_command.py
+1 -1
View File
@@ -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()
tests/cli/test_cli_yolo_resume_persistence.py
+3 -2
View File
@@ -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
tests/cli/test_cli_yolo_toggle.py
+3 -2
View File
@@ -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)
tests/cron/test_scheduler_cron_session_isolation.py
+5 -4
View File
@@ -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)
tests/gateway/test_api_server_runs.py
+5 -4
View File
@@ -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",
})
tests/gateway/test_approve_deny_commands.py
+27 -49
View File
@@ -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 <reason> 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 <reason> 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)
tests/gateway/test_hosted_room_execution_policy.py
+3 -3
View File
@@ -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)
tests/gateway/test_plaintext_approval_routing.py
+2 -1
View File
@@ -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"})
tests/gateway/test_session_boundary_security_state.py
+2 -7
View File
@@ -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)
tests/hermes_cli/test_approval_transport.py
+13 -17
View File
@@ -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)
tests/hermes_cli/test_approvals_test.py
+13 -11
View File
@@ -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()
tests/hermes_cli/test_plugins.py
+3 -2
View File
@@ -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)
tests/run_agent/test_authorization_gate.py
+39 -37
View File
@@ -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
tests/run_agent/test_tool_executor_contextvar_propagation.py
+4 -8
View File
@@ -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 = {}
tests/tools/test_allowlist_quoted_metachars.py
+2 -4
View File
@@ -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:
tests/tools/test_approval.py
+29 -33
View File
@@ -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")
tests/tools/test_approval_config_readonly.py
+3 -7
View File
@@ -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
tests/tools/test_approval_deny_rules.py
+4 -3
View File
@@ -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):
tests/tools/test_approval_hook_session_id.py
+8 -7
View File
@@ -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",
tests/tools/test_approval_interrupt.py
+6 -4
View File
@@ -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",
tests/tools/test_approval_mode_parity.py
+9 -8
View File
@@ -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"
tests/tools/test_approval_outcome_parity.py
+3 -3
View File
@@ -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)
tests/tools/test_approval_plugin_hooks.py
+14 -15
View File
@@ -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": ""},
tests/tools/test_approval_timeout_overflow.py
+11 -10
View File
@@ -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()
tests/tools/test_async_delegation.py
+1 -1
View File
@@ -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
tests/tools/test_blocked_command_guidance.py
+5 -2
View File
@@ -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"]
tests/tools/test_cli_approval_exec_ask_leak.py
+2 -2
View File
@@ -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(
tests/tools/test_code_kernel.py
+1 -1
View File
@@ -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:
tests/tools/test_command_guards.py
+4 -9
View File
@@ -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)
tests/tools/test_cron_approval_mode.py
+23 -25
View File
@@ -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.
tests/tools/test_cronjob_run_background.py
+1 -1
View File
@@ -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():
tests/tools/test_cronjob_run_delivery_notice.py
+1 -1
View File
@@ -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:
tests/tools/test_denial_circuit_breaker.py
+12 -10
View File
@@ -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)
tests/tools/test_execution_flag_detection.py
+1 -1
View File
@@ -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()
tests/tools/test_hardline_blocklist.py
+8 -13
View File
@@ -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
tests/tools/test_mcp_elicitation.py
+10 -10
View File
@@ -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 ..."
tests/tools/test_modal_sandbox_fixes.py
+3 -1
View File
@@ -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."""
tests/tools/test_request_tool_approval.py
+5 -4
View File
@@ -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
tests/tools/test_single_query_approval_mode.py
+17 -20
View File
@@ -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"]
tests/tools/test_smart_approval_injection.py
+1 -5
View File
@@ -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 ──────────────────────────────────────────────────
tests/tools/test_smart_approval_policy.py
+9 -9
View File
@@ -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 "<command>" 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
tests/tools/test_yolo_mode.py
+4 -12
View File
@@ -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")
tools/approval.py
+24 -41
View File
@@ -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()
tools/approval_context.py
+4 -7
View File
@@ -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:
tools/approval_detection.py
+3 -8
View File
@@ -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):
tools/approval_floors.py
+8 -8
View File
@@ -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)
tools/approval_gateway_wait.py
+8 -11
View File
@@ -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,
tools/approval_human_wait.py
+4 -4
View File
@@ -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):
tools/approval_prompt.py
+13 -14
View File
@@ -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"
tools/approval_smart.py
+5 -6
View File
@@ -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
tools/browser_extension_router.py
+1 -1
View File
@@ -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:
tools/delegate_tool_dispatch.py
+1 -1
View File
@@ -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):
tools/mcp_tool_sampling.py
+1 -1
View File
@@ -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
tools/self_repo_guard.py
+1 -1
View File
@@ -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)
tools/terminal_tool_sudo.py
+2 -2
View File
@@ -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"]:
tui_gateway/methods_config_set.py
+1 -1
View File
@@ -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")
tui_gateway/prompt_turn.py
+2 -2
View File
@@ -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)
tui_gateway/server.py
+1 -1
View File
@@ -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"