From 74cc9ee3f06ab09d7f6b06103b961722a2153578 Mon Sep 17 00:00:00 2001
From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com>
Date: Mon, 6 Jul 2026 02:36:52 +0530
Subject: [PATCH] Revert "Merge pull request #58698 from
 kshitijk4poor/feat/pre-tool-call-approve-escalation"

This reverts commit 368e5f197e723ed39d40b93baae86e5522d7f22f, reversing
changes made to abf9638f4eb3dc02d4159bae5c3af86457edd323.
---
 agent/agent_runtime_helpers.py                |   8 +-
 agent/tool_executor.py                        |   8 +-
 hermes_cli/plugins.py                         | 119 +------
 model_tools.py                                |  19 +-
 tests/hermes_cli/test_plugins.py              | 109 -------
 tests/run_agent/test_run_agent.py             |  30 +-
 .../test_tool_call_guardrail_runtime.py       |   2 +-
 tests/tools/test_request_tool_approval.py     | 167 ----------
 tools/approval.py                             | 307 ++++--------------
 9 files changed, 103 insertions(+), 666 deletions(-)
 delete mode 100644 tests/tools/test_request_tool_approval.py

diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index da9c20c0510..ade4831d1b8 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -2094,12 +2094,12 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
     except Exception as _mw_err:
         logger.debug("tool_request middleware error: %s", _mw_err)
 
-    # Check plugin hooks for a block or approval directive before executing.
+    # Check plugin hooks for a block directive before executing anything.
     block_message: Optional[str] = None
     if not pre_tool_block_checked:
         try:
-            from hermes_cli.plugins import resolve_pre_tool_block
-            block_message = resolve_pre_tool_block(
+            from hermes_cli.plugins import get_pre_tool_call_block_message
+            block_message = get_pre_tool_call_block_message(
                 function_name,
                 function_args,
                 task_id=effective_task_id or "",
@@ -2110,7 +2110,7 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
                 middleware_trace=list(_tool_middleware_trace),
             )
         except Exception:
-            block_message = None
+            pass
     if block_message is not None:
         result = json.dumps({"error": block_message}, ensure_ascii=False)
         try:
diff --git a/agent/tool_executor.py b/agent/tool_executor.py
index 9688f5d3dc6..44b9a367c90 100644
--- a/agent/tool_executor.py
+++ b/agent/tool_executor.py
@@ -415,8 +415,8 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
             )
         else:
             try:
-                from hermes_cli.plugins import resolve_pre_tool_block
-                block_message = resolve_pre_tool_block(
+                from hermes_cli.plugins import get_pre_tool_call_block_message
+                block_message = get_pre_tool_call_block_message(
                     function_name,
                     function_args,
                     task_id=effective_task_id or "",
@@ -1034,8 +1034,8 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
             _block_error_type = "tool_scope_block"
         else:
             try:
-                from hermes_cli.plugins import resolve_pre_tool_block
-                _block_msg = resolve_pre_tool_block(
+                from hermes_cli.plugins import get_pre_tool_call_block_message
+                _block_msg = get_pre_tool_call_block_message(
                     function_name,
                     function_args,
                     task_id=effective_task_id or "",
diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py
index e8b1b59b798..d5e4b3ff8c1 100644
--- a/hermes_cli/plugins.py
+++ b/hermes_cli/plugins.py
@@ -2046,7 +2046,7 @@ def clear_thread_tool_whitelist() -> None:
     _thread_tool_whitelist.allowed = None
 
 
-def get_pre_tool_call_directive(
+def get_pre_tool_call_block_message(
     tool_name: str,
     args: Optional[Dict[str, Any]],
     task_id: str = "",
@@ -2055,38 +2055,22 @@ def get_pre_tool_call_directive(
     turn_id: str = "",
     api_request_id: str = "",
     middleware_trace: Optional[List[Dict[str, Any]]] = None,
-) -> tuple[Optional[str], Optional[str]]:
-    """Check ``pre_tool_call`` hooks for a blocking or approval directive.
+) -> Optional[str]:
+    """Check ``pre_tool_call`` hooks for a blocking directive.
 
     Plugins that need to enforce policy (rate limiting, security
-    restrictions, approval workflows) can return one of::
+    restrictions, approval workflows) can return::
 
-        {"action": "block",   "message": "Reason the tool was blocked"}
-        {"action": "approve", "message": "Why this needs human confirmation"}
+        {"action": "block", "message": "Reason the tool was blocked"}
 
-    from their ``pre_tool_call`` callback.
-
-    - ``block`` vetoes the tool call outright (the message becomes the tool
-      result the model sees).
-    - ``approve`` ESCALATES to the existing human-approval gate
-      (``prompt_dangerous_approval`` on CLI, the approval callback on the
-      gateway) — the same mechanism Tier-2 dangerous shell patterns use.
-      This lets a plugin require a human ``[o]nce/[s]ession/[a]lways/[d]eny``
-      decision on ANY tool, not just terminal command strings. The caller is
-      responsible for invoking the gate (see
-      :func:`tools.approval.request_tool_approval`).
-
-    The first valid directive wins. Invalid or irrelevant hook return values
-    are silently ignored so existing observer-only hooks are unaffected.
-
-    Returns:
-        ``(directive, message)`` where ``directive`` is ``"block"``,
-        ``"approve"``, or ``None``.
+    from their ``pre_tool_call`` callback.  The first valid block
+    directive wins.  Invalid or irrelevant hook return values are
+    silently ignored so existing observer-only hooks are unaffected.
     """
     allowed = getattr(_thread_tool_whitelist, "allowed", None)
     if allowed is not None and tool_name not in allowed:
         fmt = getattr(_thread_tool_whitelist, "fmt", "Tool '{tool_name}' denied")
-        return ("block", fmt.format(tool_name=tool_name))
+        return fmt.format(tool_name=tool_name)
 
     hook_results = invoke_hook(
         "pre_tool_call",
@@ -2103,91 +2087,12 @@ def get_pre_tool_call_directive(
     for result in hook_results:
         if not isinstance(result, dict):
             continue
-        action = result.get("action")
-        if action not in ("block", "approve"):
+        if result.get("action") != "block":
             continue
         message = result.get("message")
-        message = message if isinstance(message, str) and message else None
-        # A block directive requires a message (it becomes the tool result);
-        # an approve directive can carry an optional reason.
-        if action == "block" and not message:
-            continue
-        return (action, message)
+        if isinstance(message, str) and message:
+            return message
 
-    return (None, None)
-
-
-def get_pre_tool_call_block_message(
-    tool_name: str,
-    args: Optional[Dict[str, Any]],
-    task_id: str = "",
-    session_id: str = "",
-    tool_call_id: str = "",
-    turn_id: str = "",
-    api_request_id: str = "",
-    middleware_trace: Optional[List[Dict[str, Any]]] = None,
-) -> Optional[str]:
-    """Back-compat shim: return only a ``block`` message (or ``None``).
-
-    Deprecated in favor of :func:`get_pre_tool_call_directive`, which also
-    surfaces the ``approve`` escalation directive. Kept so any external caller
-    importing the old name keeps working; ``approve`` directives are invisible
-    to this shim (it only reports blocks).
-    """
-    directive, message = get_pre_tool_call_directive(
-        tool_name, args, task_id=task_id, session_id=session_id,
-        tool_call_id=tool_call_id, turn_id=turn_id,
-        api_request_id=api_request_id, middleware_trace=middleware_trace,
-    )
-    return message if directive == "block" else None
-
-
-def resolve_pre_tool_block(
-    tool_name: str,
-    args: Optional[Dict[str, Any]],
-    task_id: str = "",
-    session_id: str = "",
-    tool_call_id: str = "",
-    turn_id: str = "",
-    api_request_id: str = "",
-    middleware_trace: Optional[List[Dict[str, Any]]] = None,
-) -> Optional[str]:
-    """Resolve the pre_tool_call directive to a final block message (or None).
-
-    Single entry point for every tool-dispatch site: fetches the plugin
-    directive and, for an ``approve`` escalation, invokes the human-approval
-    gate (:func:`tools.approval.request_tool_approval`). Returns the message
-    the tool result should carry when the call is blocked, or ``None`` when
-    the call may proceed.
-
-    Centralizing this keeps the security-critical fail-closed logic in ONE
-    place instead of copy-pasted across the concurrent/sequential/helper
-    dispatch paths: an ``approve`` directive whose gate errors, denies, or
-    times out is fail-closed to a block; ``block`` blocks with its message;
-    anything else proceeds.
-    """
-    directive, message = get_pre_tool_call_directive(
-        tool_name, args, task_id=task_id, session_id=session_id,
-        tool_call_id=tool_call_id, turn_id=turn_id,
-        api_request_id=api_request_id, middleware_trace=middleware_trace,
-    )
-    if directive == "block":
-        return message
-    if directive == "approve":
-        try:
-            from tools.approval import request_tool_approval
-            result = request_tool_approval(
-                tool_name, message or "", rule_key=tool_name,
-            )
-        except Exception:
-            # Fail-closed: if the gate itself errors, block rather than
-            # silently execute an action a plugin flagged for approval.
-            return f"BLOCKED: plugin approval gate failed for {tool_name}"
-        if not result.get("approved"):
-            return str(
-                result.get("message")
-                or f"BLOCKED: plugin approval required for {tool_name}"
-            )
     return None
 
 
diff --git a/model_tools.py b/model_tools.py
index 65eff7a2e96..c3c3a11a66f 100644
--- a/model_tools.py
+++ b/model_tools.py
@@ -1158,22 +1158,21 @@ def handle_function_call(
         if function_name in _AGENT_LOOP_TOOLS:
             return json.dumps({"error": f"{function_name} must be handled by the agent loop"})
 
-        # Check plugin hooks for a block/approve directive (unless caller
-        # already checked — e.g. run_agent._invoke_tool passes skip=True to
+        # Check plugin hooks for a block directive (unless caller already
+        # checked — e.g. run_agent._invoke_tool passes skip=True to
         # avoid double-firing the hook).
         #
         # Single-fire contract: pre_tool_call fires exactly once per tool
-        # execution. resolve_pre_tool_block() internally calls
-        # invoke_hook("pre_tool_call", ...) once and returns the block message
-        # for a `block` directive OR for an `approve` directive whose human
-        # gate denied/timed-out/errored (fail-closed). Observer plugins see
-        # the hook on that same pass. When skip=True, the caller already
-        # fired it — do nothing here.
+        # execution. get_pre_tool_call_block_message() internally calls
+        # invoke_hook("pre_tool_call", ...) and returns the first block
+        # directive (if any), so observer plugins see the hook on that same
+        # pass. When skip=True, the caller already fired it — do nothing
+        # here.
         if not skip_pre_tool_call_hook:
             block_message: Optional[str] = None
             try:
-                from hermes_cli.plugins import resolve_pre_tool_block
-                block_message = resolve_pre_tool_block(
+                from hermes_cli.plugins import get_pre_tool_call_block_message
+                block_message = get_pre_tool_call_block_message(
                     function_name,
                     function_args,
                     task_id=task_id or "",
diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py
index 10d68f81ea8..0df9790c31a 100644
--- a/tests/hermes_cli/test_plugins.py
+++ b/tests/hermes_cli/test_plugins.py
@@ -859,115 +859,6 @@ class TestPreToolCallBlocking:
         assert get_pre_tool_call_block_message("terminal", {}) == "first blocker"
 
 
-class TestPreToolCallDirective:
-    """Tests for the extended (block | approve) directive helper."""
-
-    def test_approve_directive_returned(self, monkeypatch):
-        from hermes_cli.plugins import get_pre_tool_call_directive
-        monkeypatch.setattr(
-            "hermes_cli.plugins.invoke_hook",
-            lambda hook_name, **kwargs: [
-                {"action": "approve", "message": "needs human ok"}
-            ],
-        )
-        assert get_pre_tool_call_directive("write_file", {}) == (
-            "approve", "needs human ok")
-
-    def test_approve_without_message_is_valid(self, monkeypatch):
-        """approve may omit a message (block may not)."""
-        from hermes_cli.plugins import get_pre_tool_call_directive
-        monkeypatch.setattr(
-            "hermes_cli.plugins.invoke_hook",
-            lambda hook_name, **kwargs: [{"action": "approve"}],
-        )
-        assert get_pre_tool_call_directive("write_file", {}) == ("approve", None)
-
-    def test_block_still_requires_message(self, monkeypatch):
-        from hermes_cli.plugins import get_pre_tool_call_directive
-        monkeypatch.setattr(
-            "hermes_cli.plugins.invoke_hook",
-            lambda hook_name, **kwargs: [{"action": "block"}],
-        )
-        assert get_pre_tool_call_directive("terminal", {}) == (None, None)
-
-    def test_first_directive_wins_across_actions(self, monkeypatch):
-        from hermes_cli.plugins import get_pre_tool_call_directive
-        monkeypatch.setattr(
-            "hermes_cli.plugins.invoke_hook",
-            lambda hook_name, **kwargs: [
-                {"action": "approve", "message": "gate first"},
-                {"action": "block", "message": "block second"},
-            ],
-        )
-        assert get_pre_tool_call_directive("terminal", {}) == (
-            "approve", "gate first")
-
-    def test_shim_ignores_approve(self, monkeypatch):
-        """Back-compat shim only reports block, never approve."""
-        monkeypatch.setattr(
-            "hermes_cli.plugins.invoke_hook",
-            lambda hook_name, **kwargs: [
-                {"action": "approve", "message": "gate"}
-            ],
-        )
-        assert get_pre_tool_call_block_message("write_file", {}) is None
-
-
-class TestResolvePreToolBlock:
-    """Tests for the single dispatch-site chokepoint that resolves a
-    directive (incl. the approve→gate escalation) to a block message."""
-
-    def test_block_returns_message(self, monkeypatch):
-        from hermes_cli.plugins import resolve_pre_tool_block
-        monkeypatch.setattr(
-            "hermes_cli.plugins.invoke_hook",
-            lambda hook_name, **kwargs: [{"action": "block", "message": "no"}],
-        )
-        assert resolve_pre_tool_block("terminal", {}) == "no"
-
-    def test_no_directive_returns_none(self, monkeypatch):
-        from hermes_cli.plugins import resolve_pre_tool_block
-        monkeypatch.setattr(
-            "hermes_cli.plugins.invoke_hook", lambda hook_name, **kwargs: [])
-        assert resolve_pre_tool_block("terminal", {}) is None
-
-    def test_approve_denied_blocks(self, monkeypatch):
-        from hermes_cli.plugins import resolve_pre_tool_block
-        monkeypatch.setattr(
-            "hermes_cli.plugins.invoke_hook",
-            lambda hook_name, **kwargs: [{"action": "approve", "message": "why"}],
-        )
-        monkeypatch.setattr(
-            "tools.approval.request_tool_approval",
-            lambda *a, **k: {"approved": False, "message": "user denied it"},
-        )
-        assert resolve_pre_tool_block("write_file", {}) == "user denied it"
-
-    def test_approve_granted_allows(self, monkeypatch):
-        from hermes_cli.plugins import resolve_pre_tool_block
-        monkeypatch.setattr(
-            "hermes_cli.plugins.invoke_hook",
-            lambda hook_name, **kwargs: [{"action": "approve", "message": "why"}],
-        )
-        monkeypatch.setattr(
-            "tools.approval.request_tool_approval",
-            lambda *a, **k: {"approved": True, "message": None},
-        )
-        assert resolve_pre_tool_block("write_file", {}) is None
-
-    def test_approve_gate_exception_fails_closed(self, monkeypatch):
-        from hermes_cli.plugins import resolve_pre_tool_block
-        monkeypatch.setattr(
-            "hermes_cli.plugins.invoke_hook",
-            lambda hook_name, **kwargs: [{"action": "approve", "message": "why"}],
-        )
-        def _boom(*a, **k):
-            raise RuntimeError("gate crashed")
-        monkeypatch.setattr("tools.approval.request_tool_approval", _boom)
-        msg = resolve_pre_tool_block("terminal", {})
-        assert msg is not None and "gate failed" in msg  # fail-closed
-
-
 class TestGetPreVerifyContinueMessage:
     """`pre_verify` directive aggregation — mirrors the pre_tool_call block path."""
 
diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py
index 171a7c11255..da8a446bf8d 100644
--- a/tests/run_agent/test_run_agent.py
+++ b/tests/run_agent/test_run_agent.py
@@ -2966,7 +2966,7 @@ class TestConcurrentToolExecution:
         """Agent-owned tool paths should close observer tool spans."""
         hook_calls = []
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: None,
         )
         monkeypatch.setattr(
@@ -2990,7 +2990,7 @@ class TestConcurrentToolExecution:
     def test_invoke_tool_blocked_returns_error_and_skips_execution(self, agent, monkeypatch):
         """_invoke_tool should return error JSON when a plugin blocks the tool."""
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: "Blocked by test policy",
         )
         with patch("tools.todo_tool.todo_tool", side_effect=AssertionError("should not run")) as mock_todo:
@@ -3002,7 +3002,7 @@ class TestConcurrentToolExecution:
     def test_invoke_tool_blocked_skips_handle_function_call(self, agent, monkeypatch):
         """Blocked registry tools should not reach handle_function_call."""
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: "Blocked",
         )
         with patch("run_agent.handle_function_call", side_effect=AssertionError("should not run")):
@@ -3019,7 +3019,7 @@ class TestConcurrentToolExecution:
         messages = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: "Blocked by policy",
         )
         agent._checkpoint_mgr.enabled = True
@@ -3049,7 +3049,7 @@ class TestConcurrentToolExecution:
         hook_calls = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: "Blocked by policy",
         )
         monkeypatch.setattr(
@@ -3076,7 +3076,7 @@ class TestConcurrentToolExecution:
         hook_calls = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: None,
         )
         monkeypatch.setattr(
@@ -3123,7 +3123,7 @@ class TestConcurrentToolExecution:
             lambda kind, **kwargs: [request_middleware(**kwargs)] if kind == "tool_request" else [],
         )
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: None,
         )
         monkeypatch.setattr(
@@ -3161,7 +3161,7 @@ class TestConcurrentToolExecution:
             lambda kind, **kwargs: [request_middleware(**kwargs)] if kind == "tool_request" else [],
         )
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: None,
         )
         monkeypatch.setattr(
@@ -3196,7 +3196,7 @@ class TestConcurrentToolExecution:
         """Blocked memory tool should not reset the nudge counter."""
         agent._turns_since_memory = 5
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: "Blocked",
         )
         with patch("tools.memory_tool.memory_tool", side_effect=AssertionError("should not run")):
@@ -3209,7 +3209,7 @@ class TestConcurrentToolExecution:
 
     def test_invoke_tool_memory_remove_notifies_provider_with_old_text(self, agent, monkeypatch):
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: None,
         )
         calls = []
@@ -3241,7 +3241,7 @@ class TestConcurrentToolExecution:
 
     def test_invoke_tool_memory_failed_remove_skips_provider_notification(self, agent, monkeypatch):
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: None,
         )
         notify = MagicMock(side_effect=AssertionError("should not notify"))
@@ -3281,7 +3281,7 @@ class TestConcurrentToolExecution:
         messages = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: "Blocked" if args[0] == "write_file" else None,
         )
 
@@ -3308,7 +3308,7 @@ class TestConcurrentToolExecution:
         messages = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: "Blocked" if args[0] == "patch" else None,
         )
 
@@ -3335,7 +3335,7 @@ class TestConcurrentToolExecution:
         messages = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             lambda *args, **kwargs: "Blocked" if args[0] == "terminal" else None,
         )
 
@@ -3369,7 +3369,7 @@ class TestConcurrentToolExecution:
             return "Blocked" if call_count["n"] == 1 else None
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.resolve_pre_tool_block",
+            "hermes_cli.plugins.get_pre_tool_call_block_message",
             block_first_only,
         )
 
diff --git a/tests/run_agent/test_tool_call_guardrail_runtime.py b/tests/run_agent/test_tool_call_guardrail_runtime.py
index 36ced43c795..e7ab376281a 100644
--- a/tests/run_agent/test_tool_call_guardrail_runtime.py
+++ b/tests/run_agent/test_tool_call_guardrail_runtime.py
@@ -228,7 +228,7 @@ def test_plugin_pre_tool_block_wins_without_counting_as_toolguard_block():
     messages = []
 
     with (
-        patch("hermes_cli.plugins.resolve_pre_tool_block", return_value="plugin policy"),
+        patch("hermes_cli.plugins.get_pre_tool_call_block_message", return_value="plugin policy"),
         patch("run_agent.handle_function_call", return_value="SHOULD_NOT_RUN") as mock_hfc,
     ):
         agent._execute_tool_calls_sequential(msg, messages, "task-1")
diff --git a/tests/tools/test_request_tool_approval.py b/tests/tools/test_request_tool_approval.py
deleted file mode 100644
index 16414fc054c..00000000000
--- a/tests/tools/test_request_tool_approval.py
+++ /dev/null
@@ -1,167 +0,0 @@
-"""Tests for tools.approval.request_tool_approval — the plugin pre_tool_call
-``{"action": "approve"}`` escalation into the human-approval gate.
-
-These verify that a plugin-driven approval reuses the SAME machinery as a
-Tier-2 dangerous-command match: session/permanent allowlist, the CLI prompt,
-the gateway submit_pending path, cron_mode, and fail-closed timeouts.
-"""
-
-import pytest
-
-import tools.approval as approval
-from tools.approval import request_tool_approval
-
-
-@pytest.fixture(autouse=True)
-def _isolate_approval_state(monkeypatch):
-    """Give each test a clean session key and empty allowlists."""
-    monkeypatch.setattr(
-        approval, "get_current_session_key",
-        lambda default="default": "test-session",
-    )
-    # Empty session + permanent approval stores so nothing pre-approves.
-    monkeypatch.setattr(approval, "is_approved", lambda sk, pk: False)
-    # Not a yolo session (the shared gate checks this first).
-    monkeypatch.setattr(approval, "is_current_session_yolo_enabled", lambda: False)
-    monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", False, raising=False)
-    # No thread-registered CLI callback by default.
-    monkeypatch.setattr(
-        "tools.terminal_tool._get_approval_callback", lambda: None, raising=False
-    )
-    yield
-
-
-class TestRequestToolApproval:
-    def test_session_cached_approval_short_circuits(self, monkeypatch):
-        monkeypatch.setattr(approval, "is_approved", lambda sk, pk: True)
-        # Should NOT prompt at all.
-        monkeypatch.setattr(
-            approval, "prompt_dangerous_approval",
-            lambda *a, **k: pytest.fail("should not prompt when already approved"),
-        )
-        res = request_tool_approval("write_file", "sensitive path", rule_key="ssh")
-        assert res == {"approved": True, "message": None}
-
-    def test_cli_approve_once(self, monkeypatch):
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "once")
-        res = request_tool_approval("write_file", "writing ~/.ssh/authorized_keys")
-        assert res["approved"] is True
-
-    def test_cli_deny_blocks(self, monkeypatch):
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "deny")
-        res = request_tool_approval("terminal", "curl PUT to external API")
-        assert res["approved"] is False
-        assert "denied" in res["message"].lower()
-        assert res["pattern_key"].startswith("plugin_rule:")
-
-    def test_cli_session_persists_session_only(self, monkeypatch):
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "session")
-        calls = {"session": [], "permanent": []}
-        monkeypatch.setattr(approval, "approve_session",
-                            lambda sk, pk: calls["session"].append(pk))
-        monkeypatch.setattr(approval, "approve_permanent",
-                            lambda pk: calls["permanent"].append(pk))
-        monkeypatch.setattr(approval, "save_permanent_allowlist", lambda x: None)
-        res = request_tool_approval("write_file", "reason", rule_key="ssh-writes")
-        assert res["approved"] is True
-        assert calls["session"] == ["plugin_rule:ssh-writes"]
-        assert calls["permanent"] == []  # session != always
-
-    def test_cli_always_persists_permanent(self, monkeypatch):
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "always")
-        persisted = {}
-        monkeypatch.setattr(approval, "approve_session", lambda sk, pk: None)
-        monkeypatch.setattr(approval, "approve_permanent",
-                            lambda pk: persisted.setdefault("key", pk))
-        monkeypatch.setattr(approval, "save_permanent_allowlist",
-                            lambda x: persisted.setdefault("saved", True))
-        res = request_tool_approval("write_file", "reason", rule_key="ssh-writes")
-        assert res["approved"] is True
-        assert persisted["key"] == "plugin_rule:ssh-writes"
-        assert persisted["saved"] is True
-
-    def test_gateway_path_submits_pending_and_defers(self, monkeypatch):
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: True)
-        submitted = {}
-        monkeypatch.setattr(approval, "submit_pending",
-                            lambda sk, data: submitted.update(data))
-        res = request_tool_approval("browser_navigate", "external URL",
-                                    rule_key="ext-nav")
-        assert res["approved"] is False
-        assert res["status"] == "approval_required"
-        assert submitted["pattern_key"] == "plugin_rule:ext-nav"
-
-    def test_cron_deny_mode_blocks(self, monkeypatch):
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "env_var_enabled",
-                            lambda v: v == "HERMES_CRON_SESSION")
-        monkeypatch.setattr(approval, "_get_cron_approval_mode", lambda: "deny")
-        res = request_tool_approval("terminal", "smtp send")
-        assert res["approved"] is False
-        assert "cron" in res["message"].lower()
-
-    def test_cron_approve_mode_allows(self, monkeypatch):
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "env_var_enabled",
-                            lambda v: v == "HERMES_CRON_SESSION")
-        monkeypatch.setattr(approval, "_get_cron_approval_mode", lambda: "approve")
-        res = request_tool_approval("terminal", "smtp send")
-        assert res["approved"] is True
-
-    def test_rule_key_derived_from_tool_and_reason(self, monkeypatch):
-        """With no explicit rule_key, the pattern key is derived from
-        tool + a hash of the reason (so distinct reasons persist apart)."""
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "deny")
-        res = request_tool_approval("patch", "reason")  # no rule_key
-        assert res["pattern_key"].startswith("plugin_rule:patch:")
-
-    def test_distinct_reasons_get_distinct_keys(self, monkeypatch):
-        """Two different reasons on the SAME tool must not share an [a]lways
-        allowlist entry (Finding 3: tool_name alone was too coarse)."""
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "deny")
-        k1 = request_tool_approval("write_file", "write to ~/.ssh")["pattern_key"]
-        k2 = request_tool_approval("write_file", "send an email")["pattern_key"]
-        assert k1 != k2
-
-    def test_explicit_rule_key_overrides_derivation(self, monkeypatch):
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "deny")
-        res = request_tool_approval("terminal", "any", rule_key="my-rule")
-        assert res["pattern_key"] == "plugin_rule:my-rule"
-
-    def test_no_human_non_cron_fails_closed(self, monkeypatch):
-        """Non-interactive, non-gateway, NON-cron context blocks (fail-closed)
-        — a plugin-flagged action never runs ungated without a human."""
-        monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False)
-        monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False)
-        monkeypatch.setattr(approval, "env_var_enabled", lambda v: False)  # not cron
-        res = request_tool_approval("terminal", "smtp send")
-        assert res["approved"] is False
-        assert "no interactive user or gateway" in res["message"].lower()
-
-    def test_yolo_session_bypasses_gate(self, monkeypatch):
-        """A --yolo session skips the plugin approval gate (parity with the
-        dangerous-command path, via the shared _run_approval_gate)."""
-        monkeypatch.setattr(approval, "is_current_session_yolo_enabled", lambda: True)
-        monkeypatch.setattr(
-            approval, "prompt_dangerous_approval",
-            lambda *a, **k: pytest.fail("yolo must not prompt"),
-        )
-        res = request_tool_approval("terminal", "curl PUT", rule_key="ext")
-        assert res == {"approved": True, "message": None}
diff --git a/tools/approval.py b/tools/approval.py
index 616504569ec..e01db823139 100644
--- a/tools/approval.py
+++ b/tools/approval.py
@@ -11,7 +11,6 @@ This module is the single source of truth for the dangerous command system:
 import contextvars
 import fnmatch
 import functools
-import hashlib
 import logging
 import os
 import re
@@ -1949,158 +1948,6 @@ def _smart_approve(command: str, description: str) -> str:
         return "escalate"
 
 
-def _run_approval_gate(
-    *,
-    pattern_key: str,
-    description: str,
-    display_target: str,
-    approval_callback=None,
-    cron_deny_message: str,
-    autoapprove_log_prefix: str,
-    fail_closed_when_no_human: bool = False,
-    no_human_block_message: str = "",
-) -> dict:
-    """Shared human-approval gate for a flagged action (command or tool).
-
-    This is the single decision core reused by both
-    :func:`check_dangerous_command` (dangerous shell patterns) and
-    :func:`request_tool_approval` (plugin ``pre_tool_call`` ``approve``
-    escalations). Extracting it keeps the fail-closed / cron / gateway /
-    persist policy in ONE place so the two entry points can never drift.
-
-    Ordering mirrors the historical ``check_dangerous_command`` tail:
-    yolo bypass → session-cache short-circuit → interactive/gateway/cron
-    branch → prompt → ``deny/session/always`` persistence. The caller is
-    responsible for the checks that are specific to its input shape
-    (hardline detection, command-string permanent allowlist, dangerous-
-    pattern detection) BEFORE calling this gate.
-
-    Args:
-        pattern_key: Allowlist/session key this decision is stored under.
-        description: Human-facing reason shown in the prompt.
-        display_target: The command string or synthetic tool label shown
-            to the user (redacted by ``prompt_dangerous_approval``).
-        approval_callback: Optional CLI prompt callback. When ``None`` the
-            per-thread callback registered via
-            ``tools.terminal_tool.set_approval_callback`` is used.
-        cron_deny_message: Message returned when a cron job hits this gate
-            under ``cron_mode: deny``.
-        autoapprove_log_prefix: Log line prefix for the non-interactive
-            auto-approve warning (identifies command vs plugin origin).
-        fail_closed_when_no_human: When True, a non-interactive non-gateway
-            context that is NOT a cron session (e.g. a bare script with
-            HERMES_INTERACTIVE unset) BLOCKS instead of auto-approving. The
-            dangerous-command path keeps its historical fail-open default
-            (False); the plugin-escalation path opts in to fail-closed so a
-            plugin-flagged action never runs ungated without a human.
-        no_human_block_message: Message returned when
-            ``fail_closed_when_no_human`` blocks.
-
-    Returns:
-        ``{"approved": bool, "message": str|None, ...}`` — shape shared with
-        ``check_dangerous_command`` so all callers handle it uniformly.
-    """
-    # --yolo bypasses all approval prompts (session- or process-scoped).
-    # Hardline blocks are handled by the caller BEFORE this gate, so yolo
-    # here only skips the recoverable approval layer.
-    if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled():
-        return {"approved": True, "message": None}
-
-    session_key = get_current_session_key()
-    if is_approved(session_key, pattern_key):
-        return {"approved": True, "message": None}
-
-    if approval_callback is None:
-        try:
-            from tools.terminal_tool import _get_approval_callback
-            approval_callback = _get_approval_callback()
-        except Exception:
-            approval_callback = None
-
-    is_cli = _is_interactive_cli()
-    is_gateway = _is_gateway_approval_context()
-
-    if not is_cli and not is_gateway:
-        # Cron sessions: respect cron_mode config
-        if env_var_enabled("HERMES_CRON_SESSION"):
-            if _get_cron_approval_mode() == "deny":
-                return {
-                    "approved": False,
-                    "message": cron_deny_message,
-                    "pattern_key": pattern_key,
-                    "description": description,
-                }
-            # cron_mode: approve — fall through to auto-approve below.
-        elif fail_closed_when_no_human:
-            # Non-cron, non-interactive, no gateway: no human can answer.
-            # The plugin-escalation path opts in to fail-closed here so a
-            # plugin-flagged action never runs ungated. (The dangerous-
-            # command path keeps the historical fail-open default.)
-            logger.warning(
-                "%s (pattern: %s): %s — no interactive user/gateway present; "
-                "BLOCKED (fail-closed). Set HERMES_INTERACTIVE or "
-                "HERMES_GATEWAY_SESSION to answer the prompt.",
-                autoapprove_log_prefix, pattern_key, description,
-            )
-            return {
-                "approved": False,
-                "message": no_human_block_message or (
-                    f"BLOCKED: approval required ({description}) but no "
-                    "interactive user or gateway is present to approve it."
-                ),
-                "pattern_key": pattern_key,
-                "description": description,
-            }
-        logger.warning(
-            "%s (pattern: %s): %s — set HERMES_INTERACTIVE or "
-            "HERMES_GATEWAY_SESSION to require approval.",
-            autoapprove_log_prefix, pattern_key, description,
-        )
-        return {"approved": True, "message": None}
-
-    if is_gateway or env_var_enabled("HERMES_EXEC_ASK"):
-        submit_pending(session_key, {
-            "command": display_target,
-            "pattern_key": pattern_key,
-            "description": description,
-        })
-        return {
-            "approved": False,
-            "pattern_key": pattern_key,
-            "status": "approval_required",
-            "command": display_target,
-            "description": description,
-            "message": (
-                f"⚠️ This action is potentially dangerous ({description}). "
-                f"Asking the user for approval.\n\n**Target:**\n```\n{display_target}\n```"
-            ),
-        }
-
-    choice = prompt_dangerous_approval(display_target, description,
-                                       approval_callback=approval_callback)
-
-    if choice == "deny":
-        return {
-            "approved": False,
-            "message": (
-                f"BLOCKED: User denied this potentially dangerous action "
-                f"(matched '{description}'). Do NOT retry — the user has "
-                "explicitly rejected it."
-            ),
-            "pattern_key": pattern_key,
-            "description": description,
-        }
-
-    if choice == "session":
-        approve_session(session_key, pattern_key)
-    elif choice == "always":
-        approve_session(session_key, pattern_key)
-        approve_permanent(pattern_key)
-        save_permanent_allowlist(_permanent_approved)
-
-    return {"approved": True, "message": None}
-
-
 def _should_skip_container_guards(env_type: str, has_host_access: bool = False) -> bool:
     """Return True when the backend is isolated enough to skip dangerous-command prompts.
 
@@ -2158,109 +2005,71 @@ def check_dangerous_command(command: str, env_type: str,
     if not is_dangerous:
         return {"approved": True, "message": None}
 
-    return _run_approval_gate(
-        pattern_key=pattern_key,
-        description=description,
-        display_target=command,
-        approval_callback=approval_callback,
-        cron_deny_message=(
-            f"BLOCKED: Command flagged as dangerous ({description}) "
-            "but cron jobs run without a user present to approve it. "
-            "Find an alternative approach that avoids this command. "
-            "To allow dangerous commands in cron jobs, set "
-            "approvals.cron_mode: approve in config.yaml."
-        ),
-        autoapprove_log_prefix=(
-            "AUTO-APPROVED dangerous command in non-interactive non-gateway context"
-        ),
-    )
+    session_key = get_current_session_key()
+    if is_approved(session_key, pattern_key):
+        return {"approved": True, "message": None}
 
+    is_cli = _is_interactive_cli()
+    is_gateway = _is_gateway_approval_context()
 
-def request_tool_approval(
-    tool_name: str,
-    reason: str,
-    *,
-    rule_key: str = "",
-    approval_callback=None,
-) -> dict:
-    """Escalate an arbitrary tool call to the human-approval gate.
+    if not is_cli and not is_gateway:
+        # Cron sessions: respect cron_mode config
+        if env_var_enabled("HERMES_CRON_SESSION"):
+            if _get_cron_approval_mode() == "deny":
+                return {
+                    "approved": False,
+                    "message": (
+                        f"BLOCKED: Command flagged as dangerous ({description}) "
+                        "but cron jobs run without a user present to approve it. "
+                        "Find an alternative approach that avoids this command. "
+                        "To allow dangerous commands in cron jobs, set "
+                        "approvals.cron_mode: approve in config.yaml."
+                    ),
+                }
+        logger.warning(
+            "AUTO-APPROVED dangerous command in non-interactive non-gateway context "
+            "(pattern: %s): %s — set HERMES_INTERACTIVE or HERMES_GATEWAY_SESSION to require approval.",
+            description, command[:200],
+        )
+        return {"approved": True, "message": None}
 
-    This is the entry point for a plugin ``pre_tool_call`` hook that returns
-    ``{"action": "approve", "message": ...}``: instead of the plugin vetoing
-    the call (``action: block``) or silently allowing it, it asks the SAME
-    human gate that Tier-2 dangerous shell patterns use. The LLM cannot skip
-    or bypass this — the tool call is intercepted before execution.
+    if is_gateway or env_var_enabled("HERMES_EXEC_ASK"):
+        submit_pending(session_key, {
+            "command": command,
+            "pattern_key": pattern_key,
+            "description": description,
+        })
+        return {
+            "approved": False,
+            "pattern_key": pattern_key,
+            "status": "approval_required",
+            "command": command,
+            "description": description,
+            "message": (
+                f"⚠️ This command is potentially dangerous ({description}). "
+                f"Asking the user for approval.\n\n**Command:**\n```\n{command}\n```"
+            ),
+        }
 
-    It reuses the existing approval primitives (session/permanent allowlist,
-    ``prompt_dangerous_approval`` for CLI, ``submit_pending`` for the gateway
-    callback, ``[o]nce/[s]ession/[a]lways/[d]eny``, timeout fail-closed) so
-    behavior is identical to a dangerous-command match — only the trigger
-    (a plugin rule on any tool) differs.
+    choice = prompt_dangerous_approval(command, description,
+                                       approval_callback=approval_callback)
 
-    Args:
-        tool_name: The tool being gated (e.g. ``"write_file"``, ``"terminal"``).
-        reason: Human-facing message from the plugin explaining why approval
-            is needed (rendered in the prompt).
-        rule_key: Optional stable identifier the plugin can supply to control
-            the ``[a]lways`` allowlist grain. When empty, the key is derived
-            from ``tool_name`` + a hash of ``reason`` so that DISTINCT reasons
-            on the same tool persist independently (answering ``[a]lways`` to
-            "write to ~/.ssh" does NOT auto-approve a later "send email" rule
-            on the same tool).
-        approval_callback: Optional CLI callback for interactive prompts
-            (same contract as ``check_dangerous_command``).
+    if choice == "deny":
+        return {
+            "approved": False,
+            "message": f"BLOCKED: User denied this potentially dangerous command (matched '{description}' pattern). Do NOT retry this command - the user has explicitly rejected it.",
+            "pattern_key": pattern_key,
+            "description": description,
+        }
 
-    Returns:
-        ``{"approved": True, "message": None}`` when allowed, or
-        ``{"approved": False, "message": <reason>, ...}`` when denied /
-        blocked. Shape matches ``check_dangerous_command`` so callers handle
-        both paths identically.
+    if choice == "session":
+        approve_session(session_key, pattern_key)
+    elif choice == "always":
+        approve_session(session_key, pattern_key)
+        approve_permanent(pattern_key)
+        save_permanent_allowlist(_permanent_approved)
 
-    Non-interactive contexts: cron jobs honor ``approvals.cron_mode`` (parity
-    with dangerous commands); any OTHER non-interactive non-gateway context
-    (a bare script with no ``HERMES_INTERACTIVE``) fails CLOSED — a plugin-
-    flagged action never runs ungated without a human.
-    """
-    description = reason or f"Plugin requires approval for {tool_name}"
-    # Allowlist grain: an explicit plugin rule_key wins; otherwise derive from
-    # tool + a short hash of the reason so distinct reasons on the same tool
-    # get independent [a]lways entries (Finding: rule_key=tool_name alone was
-    # too coarse — one "always" would blanket every rule on that tool).
-    if rule_key:
-        key_suffix = rule_key
-    else:
-        _reason_hash = hashlib.sha256(description.encode("utf-8")).hexdigest()[:12]
-        key_suffix = f"{tool_name}:{_reason_hash}"
-    # Synthetic pattern key so plugin-rule approvals live in the same
-    # session/permanent allowlist machinery as command patterns, namespaced
-    # to avoid ever colliding with a real command pattern key.
-    pattern_key = f"plugin_rule:{key_suffix}"
-    # A synthetic "command" string for the display/allowlist layer. It never
-    # executes; it only labels the gate. Namespaced identically.
-    display_target = f"<{tool_name}> (plugin approval rule)"
-
-    return _run_approval_gate(
-        pattern_key=pattern_key,
-        description=description,
-        display_target=display_target,
-        approval_callback=approval_callback,
-        cron_deny_message=(
-            f"BLOCKED: Tool '{tool_name}' requires approval ({description}) "
-            "but cron jobs run without a user present to approve it. Find an "
-            "alternative approach. To allow flagged actions in cron jobs, set "
-            "approvals.cron_mode: approve in config.yaml."
-        ),
-        autoapprove_log_prefix=(
-            f"plugin-escalated tool call '{tool_name}' in "
-            "non-interactive non-gateway context"
-        ),
-        fail_closed_when_no_human=True,
-        no_human_block_message=(
-            f"BLOCKED: Tool '{tool_name}' requires approval ({description}) "
-            "but no interactive user or gateway is present to approve it. "
-            "A plugin flagged this action for human confirmation."
-        ),
-    )
+    return {"approved": True, "message": None}
 
 
 # =========================================================================