Route pre_tool_call approvals through human gate with rule keys

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #60504 feat(plugins): re-land pre_tool_call approve escalation with rule keys + working gateway notify

  merged to main

  files: agent/agent_runtime_helpers.py, agent/tool_executor.py, hermes_cli/plugins.py, model_tools.py, scripts/release.py, tests/hermes_cli/test_plugins.py, tests/run_agent/test_run_agent.py, tests/run_agent/test_tool_call_guardrail_runtime.py, tests/tools/test_request_tool_approval.py, tools/approval.py

  Show the diff

  • agent/agent_runtime_helpers.py

    @@ -2115,12 +2115,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 directive before executing anything.
    +    # Check plugin hooks for a block or approval directive before executing.
         block_message: Optional[str] = None
         if not pre_tool_block_checked:
             try:
    -            from hermes_cli.plugins import get_pre_tool_call_block_message
    -            block_message = get_pre_tool_call_block_message(
    +            from hermes_cli.plugins import resolve_pre_tool_block
    +            block_message = resolve_pre_tool_block(
                     function_name,
                     function_args,
                     task_id=effective_task_id or "",
    @@ -2131,7 +2131,7 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
                     middleware_trace=list(_tool_middleware_trace),
                 )
             except Exception:
    -            pass
    +            block_message = None
         if block_message is not None:
             result = json.dumps({"error": block_message}, ensure_ascii=False)
             try:

  • 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 get_pre_tool_call_block_message
    -                block_message = get_pre_tool_call_block_message(
    +                from hermes_cli.plugins import resolve_pre_tool_block
    +                block_message = resolve_pre_tool_block(
                         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 get_pre_tool_call_block_message
    -                _block_msg = get_pre_tool_call_block_message(
    +                from hermes_cli.plugins import resolve_pre_tool_block
    +                _block_msg = resolve_pre_tool_block(
                         function_name,
                         function_args,
                         task_id=effective_task_id or "",

  • hermes_cli/plugins.py

    @@ -2077,6 +2077,13 @@ def has_hook(hook_name: str) -> bool:
     _thread_tool_whitelist = threading.local()
     
     
    +@dataclass(frozen=True)
    +class _PreToolCallDirective:
    +    action: Optional[str] = None
    +    message: Optional[str] = None
    +    rule_key: Optional[str] = None
    +
    +
     def set_thread_tool_whitelist(
         allowed: Optional[Set[str]],
         deny_msg_fmt: str = "Tool '{tool_name}' denied: not in this thread's tool whitelist",
    @@ -2089,7 +2096,7 @@ def clear_thread_tool_whitelist() -> None:
         _thread_tool_whitelist.allowed = None
     
     
    -def get_pre_tool_call_block_message(
    +def _get_pre_tool_call_directive_details(
         tool_name: str,
         args: Optional[Dict[str, Any]],
         task_id: str = "",
    @@ -2098,22 +2105,40 @@ def get_pre_tool_call_block_message(
         turn_id: str = "",
         api_request_id: str = "",
         middleware_trace: Optional[List[Dict[str, Any]]] = None,
    -) -> Optional[str]:
    -    """Check ``pre_tool_call`` hooks for a blocking directive.
    +) -> _PreToolCallDirective:
    +    """Check ``pre_tool_call`` hooks for a blocking or approval directive.
     
         Plugins that need to enforce policy (rate limiting, security
    -    restrictions, approval workflows) can return::
    -
    -        {"action": "block", "message": "Reason the tool was blocked"}
    -
    -    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.
    +    restrictions, approval workflows) can return one of::
    +
    +        {"action": "block",   "message": "Reason the tool was blocked"}
    +        {"action": "approve", "message": "Why this needs human confirmation"}
    +        {"action": "approve", "message": "...", "rule_key": "write_file:ssh"}
    +
    +    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`).
    +    - ``rule_key`` is optional and only honored for ``approve`` directives. It
    +      lets plugins choose the allowlist grain for `[a]lways` approvals.
    +
    +    The first valid 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 fmt.format(tool_name=tool_name)
    +        return _PreToolCallDirective(
    +            action="block",
    +            message=fmt.format(tool_name=tool_name),
    +        )
     
         hook_results = invoke_hook(
             "pre_tool_call",
    @@ -2130,12 +2155,122 @@ def get_pre_tool_call_block_message(
         for result in hook_results:
             if not isinstance(result, dict):
                 continue
    -        if result.get("action") != "block":
    +        action = result.get("action")
    +        if action not in ("block", "approve"):
                 continue
             message = result.get("message")
    -        if isinstance(message, str) and message:
    -            return 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
    +        rule_key = result.get("rule_key") if action == "approve" else None
    +        rule_key = rule_key.strip() if isinstance(rule_key, str) else None
    +        if not rule_key:
    +            rule_key = None
    +        return _PreToolCallDirective(action=action, message=message, rule_key=rule_key)
    +
    +    return _PreToolCallDirective()
     
    +
    +def get_pre_tool_call_directive(
    +    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,
    +) -> tuple[Optional[str], Optional[str]]:
    +    """Check ``pre_tool_call`` hooks for a blocking or approval directive.
    +
    +    Backward-compatible public helper: returns ``(directive, message)`` where
    +    ``directive`` is ``"block"``, ``"approve"``, or ``None``. Internal callers
    +    that need approve-specific metadata use
    +    :func:`_get_pre_tool_call_directive_details`.
    +    """
    +    details = _get_pre_tool_call_directive_details(
    +        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 (details.action, details.message)
    +
    +
    +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.
    +    """
    +    details = _get_pre_tool_call_directive_details(
    +        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 details.action == "block":
    +        return details.message
    +    if details.action == "approve":
    +        try:
    +            from tools.approval import request_tool_approval
    +            result = request_tool_approval(
    +                tool_name,
    +                details.message or "",
    +                rule_key=details.rule_key or 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
     
     

  • model_tools.py

    @@ -1161,21 +1161,22 @@ 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 directive (unless caller already
    -        # checked — e.g. run_agent._invoke_tool passes skip=True to
    +        # Check plugin hooks for a block/approve 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. 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.
    +        # 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.
             if not skip_pre_tool_call_hook:
                 block_message: Optional[str] = None
                 try:
    -                from hermes_cli.plugins import get_pre_tool_call_block_message
    -                block_message = get_pre_tool_call_block_message(
    +                from hermes_cli.plugins import resolve_pre_tool_block
    +                block_message = resolve_pre_tool_block(
                         function_name,
                         function_args,
                         task_id=task_id or "",

  • scripts/release.py

    @@ -291,6 +291,7 @@
         "[email protected]": "synapsesx",
         "[email protected]": "itsflownium",
         "[email protected]": "dirtyren",
    +    "[redacted:email]": "doncazper",
         "[email protected]": "alex107ivanov",
         "[email protected]": "rungmc357",
         "[redacted:email]": "flo1t",

  • tests/hermes_cli/test_plugins.py

    @@ -859,6 +859,171 @@ def test_first_valid_block_wins(self, monkeypatch):
             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_passes_plugin_rule_key_to_gate(self, monkeypatch):
    +        from hermes_cli.plugins import resolve_pre_tool_block
    +
    +        seen = {}
    +
    +        monkeypatch.setattr(
    +            "hermes_cli.plugins.invoke_hook",
    +            lambda hook_name, **kwargs: [
    +                {
    +                    "action": "approve",
    +                    "message": "why",
    +                    "rule_key": "write_file:ssh",
    +                }
    +            ],
    +        )
    +
    +        def _approve(tool_name, reason, **kwargs):
    +            seen["tool_name"] = tool_name
    +            seen["reason"] = reason
    +            seen["rule_key"] = kwargs.get("rule_key")
    +            return {"approved": True, "message": None}
    +
    +        monkeypatch.setattr("tools.approval.request_tool_approval", _approve)
    +
    +        assert resolve_pre_tool_block("write_file", {}) is None
    +        assert seen == {
    +            "tool_name": "write_file",
    +            "reason": "why",
    +            "rule_key": "write_file:ssh",
    +        }
    +
    +    @pytest.mark.parametrize("rule_key", [None, "", "   ", 123, object()])
    +    def test_approve_falls_back_to_tool_name_without_valid_rule_key(
    +        self, monkeypatch, rule_key
    +    ):
    +        from hermes_cli.plugins import resolve_pre_tool_block
    +
    +        seen = {}
    +        directive = {"action": "approve", "message": "why"}
    +        if rule_key is not None:
    +            directive["rule_key"] = rule_key
    +
    +        monkeypatch.setattr(
    +            "hermes_cli.plugins.invoke_hook",
    +            lambda hook_name, **kwargs: [directive],
    +        )
    +
    +        def _approve(tool_name, reason, **kwargs):
    +            seen["rule_key"] = kwargs.get("rule_key")
    +            return {"approved": True, "message": None}
    +
    +        monkeypatch.setattr("tools.approval.request_tool_approval", _approve)
    +
    +        assert resolve_pre_tool_block("write_file", {}) is None
    +        assert seen["rule_key"] == "write_file"
    +
    +    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 @@ def test_invoke_tool_agent_level_tool_emits_terminal_post_tool_hook(self, agent,
             """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 @@ def test_invoke_tool_agent_level_tool_emits_terminal_post_tool_hook(self, agent,
         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 @@ def test_invoke_tool_blocked_returns_error_and_skips_execution(self, agent, monk
         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 @@ def test_sequential_blocked_tool_skips_checkpoints_and_callbacks(self, agent, mo
             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 @@ def test_sequential_blocked_tool_emits_terminal_post_tool_hook(self, agent, monk
             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 @@ def test_sequential_agent_level_tool_emits_terminal_post_tool_hook(self, agent,
             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 @@ def execution_middleware(**kwargs):
                 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 @@ def request_middleware(**kwargs):
                 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 @@ def test_blocked_memory_tool_does_not_reset_counter(self, agent, monkeypatch):
             """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 @@ def test_blocked_memory_tool_does_not_reset_counter(self, agent, monkeypatch):
     
         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 @@ def on_memory_write(self, action, target, content, metadata=None):
     
         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 @@ def test_concurrent_blocked_write_skips_checkpoint(self, agent, monkeypatch):
             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 @@ def test_concurrent_blocked_patch_skips_checkpoint(self, agent, monkeypatch):
             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 @@ def test_concurrent_blocked_terminal_skips_checkpoint(self, agent, monkeypatch):
             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 @@ def block_first_only(*args, **kwargs):
                 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
    +
    +
    [redacted:email](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}

  • tools/approval.py

    @@ -11,6 +11,7 @@
     import contextvars
     import fnmatch
     import functools
    +import hashlib
     import logging
     import os
     import re
    @@ -1995,6 +1996,224 @@ 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"):
    +        # Interactive gateway round-trip when a notify callback is
    +        # registered for this session (Discord/Telegram/Slack embed +
    +        # buttons, same mechanism as check_dangerous_command). Blocks the
    +        # agent thread until the user answers; the agent never sees
    +        # "approval_required" on this path — it gets a definitive
    +        # approved/BLOCKED outcome.
    +        notify_cb = None
    +        with _lock:
    +            notify_cb = _gateway_notify_cbs.get(session_key)
    +
    +        if notify_cb is not None:
    +            from agent.redact import redact_sensitive_text
    +            approval_data = {
    +                "command": redact_sensitive_text(display_target),
    +                "pattern_key": pattern_key,
    +                "pattern_keys": [pattern_key],
    +                "description": redact_sensitive_text(description),
    +                "allow_permanent": True,
    +            }
    +            decision = _await_gateway_decision(
    +                session_key, notify_cb, approval_data, surface="gateway"
    +            )
    +            if decision.get("notify_failed"):
    +                return {
    +                    "approved": False,
    +                    "message": "BLOCKED: Failed to send approval request to user. Do NOT retry.",
    +                    "pattern_key": pattern_key,
    +                    "description": description,
    +                }
    +            resolved = decision["resolved"]
    +            choice = decision["choice"]
    +            deny_reason = decision.get("reason")
    +
    +            if not resolved or choice is None or choice == "deny":
    +                if not resolved:
    +                    reason = "timed out without user response"
    +                    timeout_addendum = " Silence is not consent."
    +                else:
    +                    reason = "denied by user"
    +                    timeout_addendum = ""
    +                reason_addendum = ""
    +                if resolved and deny_reason:
    +                    reason_addendum = f' Reason given by the user: "{deny_reason}".'
    +                return {
    +                    "approved": False,
    +                    "message": (
    +                        f"BLOCKED: Action {reason}.{reason_addendum} The user "
    +                        f"has NOT consented to this action. Do NOT retry it, "
    +                        f"do NOT rephrase it, and do NOT attempt the same "
    +                        f"outcome via a different path.{timeout_addendum}"
    +                    ),
    +                    "pattern_key": pattern_key,
    +                    "description": description,
    +                    "user_consent": False,
    +                }
    +
    +            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}
    +
    +        # No notify callback (e.g. API server without an attached chat):
    +        # queue for /approve /deny review, agent sees approval_required.
    +        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.
     
    @@ -2061,71 +2280,109 @@ def check_dangerous_command(command: str, env_type: str,
         if not is_dangerous:
             return {"approved": True, "message": None}
     
    -    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()
    -
    -    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}
    -
    -    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```"
    -            ),
    -        }
    +    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"
    +        ),
    +    )
     
    -    choice = prompt_dangerous_approval(command, description,
    -                                       approval_callback=approval_callback)
     
    -    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,
    -        }
    +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.
    +
    +    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.
    +
    +    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.
     
    -    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)
    +    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``).
     
    -    return {"approved": True, "message": None}
    +    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.
    +
    +    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."
    +        ),
    +    )
     
     
     # =========================================================================