fix(hooks): stop shell hook approval requests from bypassing confirmation

agent/shell_hooks.py

@@ -720,21 +720,14 @@ def _evaluate_result(
         )
 
     stdout = (r["stdout"] or "").strip()
-    parsed = _parse_response(spec.event, stdout)
-
-    if parsed is None and fail_closed and stdout:
-        # The hook produced output we could not turn into a directive.
-        # A fail-closed gate must not silently allow the action on
-        # garbage output (e.g. a stack trace on stdout).
-        try:
-            data = json.loads(stdout)
-            valid_json = isinstance(data, dict)
-        except json.JSONDecodeError:
-            valid_json = False
-        if not valid_json:
-            return _fail_closed_block(
-                spec, "unparseable stdout (expected a JSON object)",
-            )
+    parsed, response_error = _parse_response_with_error(spec.event, stdout)
+    if response_error:
+        logger.warning(
+            "shell hook response rejected (event=%s command=%s): %s",
+            spec.event, spec.command, response_error,
+        )
+        if fail_closed:
+            return _fail_closed_block(spec, response_error)
 
     return parsed
 
@@ -786,42 +779,81 @@ def _parse_response(event: str, stdout: str) -> Optional[Dict[str, Any]]:
     ``{"action": "modify", "args": {...}}`` so callers can merge the
     returned fields into the tool's ``args`` before dispatch.
 
+    The canonical ``{"action": "approve"}`` shape is passed through with
+    optional ``message`` and ``rule_key`` fields so it reaches the shared
+    human-approval gate.
+
     For ``pre_llm_call``, ``{"context": "..."}`` is passed through
     unchanged to match the existing plugin-hook contract.
 
-    Anything else returns ``None``.
+    Valid no-op objects return ``None``. Unsupported directive values are
+    logged, and callers configured with ``fail_closed`` turn them into blocks.
     """
+    parsed, response_error = _parse_response_with_error(event, stdout)
+    if response_error:
+        logger.warning("shell hook response rejected (event=%s): %s", event, response_error)
+    return parsed
+
+
+def response_validation_error(event: str, stdout: str) -> Optional[str]:
+    """Return why non-empty shell-hook stdout cannot be honored, if any."""
+    _, error = _parse_response_with_error(event, stdout)
+    return error
+
+
+def _parse_response_with_error(
+    event: str, stdout: str,
+) -> tuple[Optional[Dict[str, Any]], Optional[str]]:
     stdout = (stdout or "").strip()
     if not stdout:
-        return None
+        return None, None
 
     try:
         data = json.loads(stdout)
     except json.JSONDecodeError:
-        logger.warning(
-            "shell hook stdout was not valid JSON (event=%s): %s",
-            event, stdout[:200],
-        )
-        return None
+        return None, "unparseable stdout (expected a JSON object)"
 
     if not isinstance(data, dict):
-        return None
+        return None, "unparseable stdout (expected a JSON object)"
 
     if event == "pre_tool_call":
         if data.get("action") == "block":
-            return {"action": "block", "message": _block_message(data.get("message"), data.get("reason"))}
+            return {
+                "action": "block",
+                "message": _block_message(data.get("message"), data.get("reason")),
+            }, None
         if data.get("decision") == "block":
-            return {"action": "block", "message": _block_message(data.get("reason"), data.get("message"))}
+            return {
+                "action": "block",
+                "message": _block_message(data.get("reason"), data.get("message")),
+            }, None
+        if data.get("action") == "approve":
+            directive: Dict[str, Any] = {"action": "approve"}
+            message = data.get("message")
+            if isinstance(message, str) and message:
+                directive["message"] = message
+            rule_key = data.get("rule_key")
+            if isinstance(rule_key, str) and rule_key.strip():
+                directive["rule_key"] = rule_key.strip()
+            return directive, None
         # "modify" action — transform tool_input before dispatch
         if data.get("action") == "modify":
             new_args = data.get("args")
             if isinstance(new_args, dict):
-                return {"action": "modify", "args": new_args}
+                return {"action": "modify", "args": new_args}, None
+            return None, "invalid pre_tool_call action 'modify': args must be an object"
         if data.get("decision") == "modify":
             new_args = data.get("tool_input")
             if isinstance(new_args, dict):
-                return {"action": "modify", "args": new_args}
-        return None
+                return {"action": "modify", "args": new_args}, None
+            return None, (
+                "invalid pre_tool_call decision 'modify': tool_input must be an object"
+            )
+        if "action" in data:
+            return None, f"unsupported pre_tool_call action: {data['action']!r}"
+        if "decision" in data:
+            return None, f"unsupported pre_tool_call decision: {data['decision']!r}"
+        return None, None
 
     if event == "pre_verify":
         # "continue" (Hermes) / "block" (Claude-Code Stop: block the stop) both
@@ -831,14 +863,14 @@ def _parse_response(event: str, stdout: str) -> Optional[Dict[str, Any]]:
         if action in {"continue", "block"}:
             message = data.get("message") or data.get("reason")
             if isinstance(message, str) and message.strip():
-                return {"action": "continue", "message": message.strip()}
-        return None
+                return {"action": "continue", "message": message.strip()}, None
+        return None, None
 
     context = data.get("context")
     if isinstance(context, str) and context.strip():
-        return {"context": context}
+        return {"context": context}, None
 
-    return None
+    return None, None
 
 
 # ---------------------------------------------------------------------------

hermes_cli/hooks.py

@@ -423,14 +423,16 @@ def _doctor_one(spec, shell_hooks) -> int:
             elapsed = result.get("elapsed_seconds", 0)
             stdout = (result.get("stdout") or "").strip()
             if stdout:
-                try:
-                    json.loads(stdout)
+                response_error = shell_hooks.response_validation_error(
+                    spec.event, stdout,
+                )
+                if response_error:
+                    problems += 1
+                    print(f"      ✗ unsupported response (exit={rc}, "
+                          f"{elapsed}s): {response_error}")
+                else:
                     print(f"      ✓ produced valid JSON on synthetic payload "
                           f"(exit={rc}, {elapsed}s)")
-                except json.JSONDecodeError:
-                    problems += 1
-                    print(f"      ✗ stdout was not valid JSON (exit={rc}, "
-                          f"{elapsed}s): {_truncate(stdout, 120)}")
             else:
                 print(f"      ✓ ran clean with empty stdout "
                       f"(exit={rc}, {elapsed}s) — hook is observer-only")

hermes_cli/plugins.py

@@ -5983,6 +5983,7 @@ class _PreToolCallDirective:
     action: Optional[str] = None
     message: Optional[str] = None
     rule_key: Optional[str] = None
+    approval_requests: Tuple[Tuple[Optional[str], Optional[str]], ...] = ()
     modified_args: Optional[Dict[str, Any]] = None
 
 
@@ -6031,8 +6032,11 @@ def _get_pre_tool_call_directive_details(
     - ``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.
+    All hooks run once before control directives are resolved. A valid block
+    dominates every approval regardless of registration order. Otherwise each
+    approval is resolved independently so one rule's grant cannot authorize a
+    different rule. Invalid or irrelevant 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:
@@ -6057,6 +6061,7 @@ def _get_pre_tool_call_directive_details(
     )
 
     block_msg: Optional[str] = None
+    approval_requests: List[Tuple[Optional[str], Optional[str]]] = []
     modified_args: Optional[Dict[str, Any]] = None
 
     for result in hook_results:
@@ -6087,8 +6092,23 @@ def _get_pre_tool_call_directive_details(
         rule_key = rule_key.strip() if isinstance(rule_key, str) else None
         if not rule_key:
             rule_key = None
+        if action == "block":
+            if block_msg is None:
+                block_msg = message
+        else:
+            approval_requests.append((message, rule_key))
+
+    if block_msg is not None:
         return _PreToolCallDirective(
-            action=action, message=message, rule_key=rule_key,
+            action="block", message=block_msg, modified_args=modified_args,
+        )
+    if approval_requests:
+        message, rule_key = approval_requests[0]
+        return _PreToolCallDirective(
+            action="approve",
+            message=message,
+            rule_key=rule_key,
+            approval_requests=tuple(approval_requests),
             modified_args=modified_args,
         )
 
@@ -6216,12 +6236,21 @@ def _resolve_block_from_details(
                 )
             except Exception:
                 pass
+            requests = details.approval_requests or (
+                (details.message, details.rule_key),
+            )
             try:
-                result = request_tool_approval(
-                    tool_name,
-                    details.message or "",
-                    rule_key=details.rule_key or tool_name,
-                )
+                for message, rule_key in requests:
+                    result = request_tool_approval(
+                        tool_name,
+                        message or "",
+                        rule_key=rule_key or "",
+                    )
+                    if not result.get("approved"):
+                        return str(
+                            result.get("message")
+                            or f"BLOCKED: plugin approval required for {tool_name}"
+                        )
             finally:
                 if approval_tokens is not None:
                     try:
@@ -6232,11 +6261,6 @@ def _resolve_block_from_details(
             # 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
 
 

tests/agent/test_shell_hooks.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import json
+import sys
 from pathlib import Path
 
 import pytest
@@ -49,6 +50,22 @@ def test_block_claude_code_style(self):
         )
         assert r == {"action": "block", "message": "nope"}
 
+    def test_approve_canonical_preserves_gate_metadata(self):
+        r = shell_hooks._parse_response(
+            "pre_tool_call",
+            json.dumps({
+                "action": "approve",
+                "message": "confirm sensitive write",
+                "rule_key": "write_file:ssh",
+            }),
+        )
+
+        assert r == {
+            "action": "approve",
+            "message": "confirm sensitive write",
+            "rule_key": "write_file:ssh",
+        }
+
 
 
     def test_empty_stdout_returns_none(self):
@@ -201,6 +218,64 @@ def test_block_aggregation_through_plugin_manager(self, tmp_path, monkeypatch):
         )
         assert msg == "blocked-by-shell"
 
+    def test_approve_escalates_through_plugin_manager(self, tmp_path, monkeypatch):
+        from hermes_cli import plugins
+
+        script = _write_script(
+            tmp_path, "approve.py",
+            'print(\'{"action": "approve", "message": "confirm write", '
+            '"rule_key": "write_file:ssh"}\')\n',
+        )
+        command = f'"{sys.executable}" "{script}"'
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "home"))
+        plugins._plugin_manager = plugins.PluginManager()
+        shell_hooks.register_from_config({
+            "hooks": {"pre_tool_call": [{"command": command}]},
+        }, accept_hooks=True)
+
+        seen = {}
+
+        def approve(tool_name, reason, **kwargs):
+            seen.update(
+                tool_name=tool_name,
+                reason=reason,
+                rule_key=kwargs.get("rule_key"),
+            )
+            return {"approved": True, "message": None}
+
+        monkeypatch.setattr("tools.approval.request_tool_approval", approve)
+
+        assert plugins.resolve_pre_tool_block("write_file", {"path": "id_rsa"}) is None
+        assert seen == {
+            "tool_name": "write_file",
+            "reason": "confirm write",
+            "rule_key": "write_file:ssh",
+        }
+
+    def test_unknown_directive_fails_closed(self, caplog):
+        spec = shell_hooks.ShellHookSpec(
+            event="pre_tool_call",
+            command="policy-hook",
+            fail_closed=True,
+        )
+        result = shell_hooks._evaluate_result(spec, {
+            "error": None,
+            "timed_out": False,
+            "elapsed_seconds": 0.01,
+            "stderr": "",
+            "stdout": '{"action": "permit"}',
+            "returncode": 0,
+        })
+
+        assert result == {
+            "action": "block",
+            "message": (
+                "hook policy-hook failed closed: unsupported pre_tool_call "
+                "action: 'permit'"
+            ),
+        }
+        assert "unsupported pre_tool_call action" in caplog.text
+
     def test_matcher_regex_filters_callback(self, tmp_path, monkeypatch):
         """A matcher set to 'terminal' must not fire for 'web_search'."""
         calls = tmp_path / "calls.log"

tests/hermes_cli/test_hooks_cli.py

@@ -4,6 +4,7 @@
 
 import io
 import json
+import sys
 from contextlib import redirect_stdout
 from pathlib import Path
 from types import SimpleNamespace
@@ -203,3 +204,25 @@ def test_unallowlisted_script_is_not_executed(self, tmp_path):
         )
         assert "not allowlisted" in out.lower()
         assert "skipped JSON smoke test" in out
+
+    def test_flags_unsupported_pre_tool_call_directive(self, tmp_path):
+        script = _hook_script(
+            tmp_path,
+            'print(\'{"action": "permit"}\')\n',
+            name="hook.py",
+        )
+        command = f'"{sys.executable}" "{script}"'
+        shell_hooks._record_approval("pre_tool_call", command)
+        cfg = {
+            "hooks": {
+                "pre_tool_call": [
+                    {"command": command, "fail_closed": True},
+                ],
+            },
+        }
+
+        with patch("hermes_cli.config.load_config", return_value=cfg):
+            out = _run(SimpleNamespace(hooks_action="doctor"))
+
+        assert "unsupported pre_tool_call action: 'permit'" in out
+        assert "1 issue(s) found" in out