feat(plugins): pre_tool_call approve action escalates to human gate

Extend the pre_tool_call plugin hook return contract with a new directive:

    {"action": "approve", "message": "why this needs human confirmation"}

Previously a pre_tool_call hook could only veto a tool call (action: block)
or allow it silently. It could not escalate to the existing human-approval
flow. This unlocks user-defined runtime approval rules on ANY tool (HTTP
writes, file writes to sensitive paths, email sends), enforced at runtime —
resolving #51221 as a pure plugin, with no core approval.py rule schema.

Mechanism:
- get_pre_tool_call_directive() returns (action, message) for block|approve;
  get_pre_tool_call_block_message() kept as a block-only back-compat shim.
- resolve_pre_tool_block() is the single dispatch-site chokepoint: fetches
  the directive and, for approve, invokes the human gate; fail-closed to a
  block on denial, timeout, or gate exception. ALL FOUR tool-dispatch sites
  now call it: tool_executor (concurrent + sequential), agent_runtime_helpers,
  and model_tools.handle_function_call.
- request_tool_approval() escalates via the SAME machinery as Tier-2
  dangerous commands: session/permanent allowlist, prompt_dangerous_approval
  (CLI) / submit_pending (gateway), [o]nce/[s]ession/[a]lways/[d]eny,
  timeout fail-closed, approvals.cron_mode for cron contexts.

Architecture: extracted the shared decision core into _run_approval_gate(),
called by BOTH check_dangerous_command() and request_tool_approval() so the
fail-closed / cron / gateway / yolo / persist policy lives in ONE place and
cannot drift. Fixed a latent divergence — the plugin path now honors --yolo.

Approval grain: [a]lways is keyed on tool_name + a hash of the reason (an
explicit plugin rule_key overrides), so distinct reasons on the same tool
persist independently instead of one 'always' blanketing the whole tool.

Non-interactive: cron honors approvals.cron_mode (parity with commands); any
other non-interactive non-gateway context fails CLOSED for the plugin path
(the command path keeps its historical fail-open default, unchanged).

No new config schema, no new env vars, no new hook events.

tests/hermes_cli/test_plugins.py

@@ -859,6 +859,115 @@ 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."""
 

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.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             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.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             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.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             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.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             lambda *args, **kwargs: "Blocked by policy",
         )
         agent._checkpoint_mgr.enabled = True
@@ -3049,7 +3049,7 @@ class TestConcurrentToolExecution:
         hook_calls = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             lambda *args, **kwargs: "Blocked by policy",
         )
         monkeypatch.setattr(
@@ -3076,7 +3076,7 @@ class TestConcurrentToolExecution:
         hook_calls = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             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.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             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.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             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.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             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.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             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.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             lambda *args, **kwargs: None,
         )
         notify = MagicMock(side_effect=AssertionError("should not notify"))
@@ -3281,7 +3281,7 @@ class TestConcurrentToolExecution:
         messages = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             lambda *args, **kwargs: "Blocked" if args[0] == "write_file" else None,
         )
 
@@ -3308,7 +3308,7 @@ class TestConcurrentToolExecution:
         messages = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             lambda *args, **kwargs: "Blocked" if args[0] == "patch" else None,
         )
 
@@ -3335,7 +3335,7 @@ class TestConcurrentToolExecution:
         messages = []
 
         monkeypatch.setattr(
-            "hermes_cli.plugins.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             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.get_pre_tool_call_block_message",
+            "hermes_cli.plugins.resolve_pre_tool_block",
             block_first_only,
         )
 

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.get_pre_tool_call_block_message", return_value="plugin policy"),
+        patch("hermes_cli.plugins.resolve_pre_tool_block", 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")

tests/tools/test_request_tool_approval.py

@@ -0,0 +1,167 @@
+"""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}