feat(plugins): add pre_approval_request / post_approval_response hooks (#16776)

Plugins can now observe dangerous-command approval events in real time,
on both the CLI-interactive path and the async gateway path. This is the
missing hook surface external tools need to build approval notifiers
(macOS menu-bar allow/deny, Slack alerts, audit logs, etc.) without
forking Hermes or running a parallel gateway adapter.

Changes:
- hermes_cli/plugins.py: add two entries to VALID_HOOKS
- tools/approval.py: fire both hooks from check_all_command_guards --
  around prompt_dangerous_approval (CLI surface) and around the
  notify_cb + blocking event.wait loop (gateway surface)
- website/docs/user-guide/features/hooks.md: document both hooks with
  a macOS-notification example
- tests/tools/test_approval_plugin_hooks.py: 5 tests covering CLI once,
  CLI deny, plugin-crash resilience, gateway approve, gateway timeout

Hooks are observer-only: return values are ignored, so plugins cannot
veto or pre-answer an approval (use pre_tool_call for that). A crashing
plugin cannot break the approval flow -- invoke_hook swallows per-
callback errors, and the wrapper logs and swallows dispatch-layer
errors too.

Surface kwarg distinguishes "cli" from "gateway"; post hook reports
choice as one of once/session/always/deny/timeout.
This commit is contained in:
Teknium 2026-04-27 20:08:33 -07:00 committed by GitHub
parent 6ea5699e3f
commit 30307a9802
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 431 additions and 0 deletions

View file

@ -30,6 +30,32 @@ _approval_session_key: contextvars.ContextVar[str] = contextvars.ContextVar(
)
def _fire_approval_hook(hook_name: str, **kwargs) -> None:
"""Invoke a plugin lifecycle hook for the approval system.
Lazy-imports the plugin manager to avoid circular imports (approval.py is
imported very early, long before plugins are discovered). Never raises --
plugin errors are logged and swallowed.
Only fires for the two approval-specific hooks in VALID_HOOKS:
pre_approval_request, post_approval_response.
"""
try:
from hermes_cli.plugins import invoke_hook
except Exception:
# Plugin system not available in this execution context
# (e.g. bare tool-only imports, minimal test environments).
return
try:
invoke_hook(hook_name, **kwargs)
except Exception as exc:
# invoke_hook() already swallows per-callback errors, so reaching here
# means the dispatch layer itself failed. Log and move on -- approval
# flow is safety-critical, plugin observability is not.
logger.debug("Approval hook %s dispatch failed: %s", hook_name, exc)
def set_current_session_key(session_key: str) -> contextvars.Token[str]:
"""Bind the active approval session key to the current context."""
return _approval_session_key.set(session_key or "")
@ -1002,6 +1028,19 @@ def check_all_command_guards(command: str, env_type: str,
with _lock:
_gateway_queues.setdefault(session_key, []).append(entry)
# Notify plugins that an approval is being requested. Fires before
# the gateway notify callback so observers (e.g. macOS notifier
# plugins, audit logs, Slack alerts) get the event in real time.
_fire_approval_hook(
"pre_approval_request",
command=command,
description=combined_desc,
pattern_key=primary_key,
pattern_keys=list(all_keys),
session_key=session_key,
surface="gateway",
)
# Notify the user (bridges sync agent thread → async gateway)
try:
notify_cb(approval_data)
@ -1067,6 +1106,24 @@ def check_all_command_guards(command: str, env_type: str,
_gateway_queues.pop(session_key, None)
choice = entry.result
# Normalize outcome for the post hook. Unresolved (timeout) and
# None both mean the user never responded; report that explicitly
# so plugins can distinguish timeout from explicit deny.
_outcome = (
"timeout" if not resolved
else (choice if choice else "timeout")
)
_fire_approval_hook(
"post_approval_response",
command=command,
description=combined_desc,
pattern_key=primary_key,
pattern_keys=list(all_keys),
session_key=session_key,
surface="gateway",
choice=_outcome,
)
if not resolved or choice is None or choice == "deny":
reason = "timed out" if not resolved else "denied by user"
return {
@ -1111,9 +1168,28 @@ def check_all_command_guards(command: str, env_type: str,
# CLI interactive: single combined prompt
# Hide [a]lways when any tirith warning is present
_fire_approval_hook(
"pre_approval_request",
command=command,
description=combined_desc,
pattern_key=primary_key,
pattern_keys=list(all_keys),
session_key=session_key,
surface="cli",
)
choice = prompt_dangerous_approval(command, combined_desc,
allow_permanent=not has_tirith,
approval_callback=approval_callback)
_fire_approval_hook(
"post_approval_response",
command=command,
description=combined_desc,
pattern_key=primary_key,
pattern_keys=list(all_keys),
session_key=session_key,
surface="cli",
choice=choice,
)
if choice == "deny":
return {