1  """hermes hooks — inspect and manage shell-script hooks.
  2  
  3  Usage::
  4  
  5      hermes hooks list
  6      hermes hooks test <event> [--for-tool X] [--payload-file F]
  7      hermes hooks revoke <command>
  8      hermes hooks doctor
  9  
 10  Consent records live under ``~/.hermes/shell-hooks-allowlist.json`` and
 11  hook definitions come from the ``hooks:`` block in ``~/.hermes/config.yaml``
 12  (the same config read by the CLI / gateway at startup).
 13  
 14  This module is a thin CLI shell over :mod:`agent.shell_hooks`; every
 15  shared concern (payload serialisation, response parsing, allowlist
 16  format) lives there.
 17  """
 18  
 19  from __future__ import annotations
 20  
 21  import json
 22  from pathlib import Path
 23  from typing import Any, Dict, List
 24  
 25  
 26  def hooks_command(args) -> None:
 27      """Entry point for ``hermes hooks`` — dispatches to the requested action."""
 28      sub = getattr(args, "hooks_action", None)
 29  
 30      if not sub:
 31          print("Usage: hermes hooks {list|test|revoke|doctor}")
 32          print("Run 'hermes hooks --help' for details.")
 33          return
 34  
 35      if sub in ("list", "ls"):
 36          _cmd_list(args)
 37      elif sub == "test":
 38          _cmd_test(args)
 39      elif sub in ("revoke", "remove", "rm"):
 40          _cmd_revoke(args)
 41      elif sub == "doctor":
 42          _cmd_doctor(args)
 43      else:
 44          print(f"Unknown hooks subcommand: {sub}")
 45  
 46  
 47  # ---------------------------------------------------------------------------
 48  # list
 49  # ---------------------------------------------------------------------------
 50  
 51  def _cmd_list(_args) -> None:
 52      from hermes_cli.config import load_config
 53      from agent import shell_hooks
 54  
 55      specs = shell_hooks.iter_configured_hooks(load_config())
 56  
 57      if not specs:
 58          print("No shell hooks configured in ~/.hermes/config.yaml.")
 59          print("See `hermes hooks --help` or")
 60          print("    website/docs/user-guide/features/hooks.md")
 61          print("for the config schema and worked examples.")
 62          return
 63  
 64      by_event: Dict[str, List] = {}
 65      for spec in specs:
 66          by_event.setdefault(spec.event, []).append(spec)
 67  
 68      allowlist = shell_hooks.load_allowlist()
 69      approved = {
 70          (e.get("event"), e.get("command"))
 71          for e in allowlist.get("approvals", [])
 72          if isinstance(e, dict)
 73      }
 74  
 75      print(f"Configured shell hooks ({len(specs)} total):\n")
 76  
 77      for event in sorted(by_event.keys()):
 78          print(f"  [{event}]")
 79          for spec in by_event[event]:
 80              is_approved = (spec.event, spec.command) in approved
 81              status = "✓ allowed" if is_approved else "✗ not allowlisted"
 82              matcher_part = f" matcher={spec.matcher!r}" if spec.matcher else ""
 83              print(
 84                  f"    - {spec.command}{matcher_part} "
 85                  f"(timeout={spec.timeout}s, {status})"
 86              )
 87  
 88              if is_approved:
 89                  entry = shell_hooks.allowlist_entry_for(spec.event, spec.command)
 90                  if entry and entry.get("approved_at"):
 91                      print(f"      approved_at: {entry['approved_at']}")
 92                      mtime_now = shell_hooks.script_mtime_iso(spec.command)
 93                      mtime_at = entry.get("script_mtime_at_approval")
 94                      if mtime_now and mtime_at and mtime_now > mtime_at:
 95                          print(
 96                              f"      ⚠ script modified since approval "
 97                              f"(was {mtime_at}, now {mtime_now}) — "
 98                              f"run `hermes hooks doctor` to re-validate"
 99                          )
100          print()
101  
102  
103  # ---------------------------------------------------------------------------
104  # test
105  # ---------------------------------------------------------------------------
106  
107  # Synthetic kwargs matching the real invoke_hook() call sites — these are
108  # passed verbatim to agent.shell_hooks.run_once(), which routes them through
109  # the same _serialize_payload() that production firings use.  That way the
110  # stdin a script sees under `hermes hooks test` and `hermes hooks doctor`
111  # is identical in shape to what it will see at runtime.
112  _DEFAULT_PAYLOADS = {
113      "pre_tool_call": {
114          "tool_name": "terminal",
115          "args": {"command": "echo hello"},
116          "session_id": "test-session",
117          "task_id": "test-task",
118          "tool_call_id": "test-call",
119      },
120      "post_tool_call": {
121          "tool_name": "terminal",
122          "args": {"command": "echo hello"},
123          "session_id": "test-session",
124          "task_id": "test-task",
125          "tool_call_id": "test-call",
126          "result": '{"output": "hello"}',
127          "duration_ms": 42,
128      },
129      "pre_llm_call": {
130          "session_id": "test-session",
131          "user_message": "What is the weather?",
132          "conversation_history": [],
133          "is_first_turn": True,
134          "model": "gpt-4",
135          "platform": "cli",
136      },
137      "post_llm_call": {
138          "session_id": "test-session",
139          "model": "gpt-4",
140          "platform": "cli",
141      },
142      "on_session_start": {"session_id": "test-session"},
143      "on_session_end": {"session_id": "test-session"},
144      "on_session_finalize": {"session_id": "test-session"},
145      "on_session_reset": {"session_id": "test-session"},
146      "pre_api_request": {
147          "session_id": "test-session",
148          "task_id": "test-task",
149          "platform": "cli",
150          "model": "claude-sonnet-4-6",
151          "provider": "anthropic",
152          "base_url": "https://api.anthropic.com",
153          "api_mode": "anthropic_messages",
154          "api_call_count": 1,
155          "message_count": 4,
156          "tool_count": 12,
157          "approx_input_tokens": 2048,
158          "request_char_count": 8192,
159          "max_tokens": 4096,
160      },
161      "post_api_request": {
162          "session_id": "test-session",
163          "task_id": "test-task",
164          "platform": "cli",
165          "model": "claude-sonnet-4-6",
166          "provider": "anthropic",
167          "base_url": "https://api.anthropic.com",
168          "api_mode": "anthropic_messages",
169          "api_call_count": 1,
170          "api_duration": 1.234,
171          "finish_reason": "stop",
172          "message_count": 4,
173          "response_model": "claude-sonnet-4-6",
174          "usage": {"input_tokens": 2048, "output_tokens": 512},
175          "assistant_content_chars": 1200,
176          "assistant_tool_call_count": 0,
177      },
178      "subagent_stop": {
179          "parent_session_id": "parent-sess",
180          "child_role": None,
181          "child_summary": "Synthetic summary for hooks test",
182          "child_status": "completed",
183          "duration_ms": 1234,
184      },
185  }
186  
187  
188  def _cmd_test(args) -> None:
189      from hermes_cli.config import load_config
190      from hermes_cli.plugins import VALID_HOOKS
191      from agent import shell_hooks
192  
193      event = args.event
194      if event not in VALID_HOOKS:
195          print(f"Unknown event: {event!r}")
196          print(f"Valid events: {', '.join(sorted(VALID_HOOKS))}")
197          return
198  
199      # Synthetic kwargs in the same shape invoke_hook() would pass.  Merged
200      # with --for-tool (overrides tool_name) and --payload-file (extra kwargs).
201      payload = dict(_DEFAULT_PAYLOADS.get(event, {"session_id": "test-session"}))
202  
203      if getattr(args, "for_tool", None):
204          payload["tool_name"] = args.for_tool
205  
206      if getattr(args, "payload_file", None):
207          try:
208              custom = json.loads(Path(args.payload_file).read_text())
209              if isinstance(custom, dict):
210                  payload.update(custom)
211              else:
212                  print(f"Warning: {args.payload_file} is not a JSON object; ignoring")
213          except Exception as exc:
214              print(f"Error reading payload file: {exc}")
215              return
216  
217      specs = shell_hooks.iter_configured_hooks(load_config())
218      specs = [s for s in specs if s.event == event]
219  
220      if getattr(args, "for_tool", None):
221          specs = [
222              s for s in specs
223              if s.event not in ("pre_tool_call", "post_tool_call")
224              or s.matches_tool(args.for_tool)
225          ]
226  
227      if not specs:
228          print(f"No shell hooks configured for event: {event}")
229          if getattr(args, "for_tool", None):
230              print(f"(with matcher filter --for-tool={args.for_tool})")
231          return
232  
233      print(f"Firing {len(specs)} hook(s) for event '{event}':\n")
234      for spec in specs:
235          print(f"  → {spec.command}")
236          result = shell_hooks.run_once(spec, payload)
237          _print_run_result(result)
238          print()
239  
240  
241  def _print_run_result(result: Dict[str, Any]) -> None:
242      if result.get("error"):
243          print(f"      ✗ error: {result['error']}")
244          return
245      if result.get("timed_out"):
246          print(f"      ✗ timed out after {result['elapsed_seconds']}s")
247          return
248  
249      rc = result.get("returncode")
250      elapsed = result.get("elapsed_seconds", 0)
251      print(f"      exit={rc}  elapsed={elapsed}s")
252  
253      stdout = (result.get("stdout") or "").strip()
254      stderr = (result.get("stderr") or "").strip()
255      if stdout:
256          print(f"      stdout: {_truncate(stdout, 400)}")
257      if stderr:
258          print(f"      stderr: {_truncate(stderr, 400)}")
259  
260      parsed = result.get("parsed")
261      if parsed:
262          print(f"      parsed (Hermes wire shape): {json.dumps(parsed)}")
263      else:
264          print("      parsed: <none — hook contributed nothing to the dispatcher>")
265  
266  
267  def _truncate(s: str, n: int) -> str:
268      return s if len(s) <= n else s[: n - 3] + "..."
269  
270  
271  # ---------------------------------------------------------------------------
272  # revoke
273  # ---------------------------------------------------------------------------
274  
275  def _cmd_revoke(args) -> None:
276      from agent import shell_hooks
277  
278      removed = shell_hooks.revoke(args.command)
279      if removed == 0:
280          print(f"No allowlist entry found for command: {args.command}")
281          return
282      print(f"Removed {removed} allowlist entry/entries for: {args.command}")
283      print(
284          "Note: currently running CLI / gateway processes keep their "
285          "already-registered callbacks until they restart."
286      )
287  
288  
289  # ---------------------------------------------------------------------------
290  # doctor
291  # ---------------------------------------------------------------------------
292  
293  def _cmd_doctor(_args) -> None:
294      from hermes_cli.config import load_config
295      from agent import shell_hooks
296  
297      specs = shell_hooks.iter_configured_hooks(load_config())
298  
299      if not specs:
300          print("No shell hooks configured — nothing to check.")
301          return
302  
303      print(f"Checking {len(specs)} configured shell hook(s)...\n")
304  
305      problems = 0
306      for spec in specs:
307          print(f"  [{spec.event}] {spec.command}")
308          problems += _doctor_one(spec, shell_hooks)
309          print()
310  
311      if problems:
312          print(f"{problems} issue(s) found.  Fix before relying on these hooks.")
313      else:
314          print("All shell hooks look healthy.")
315  
316  
317  def _doctor_one(spec, shell_hooks) -> int:
318      problems = 0
319  
320      # 1. Script exists and is executable
321      if shell_hooks.script_is_executable(spec.command):
322          print("      ✓ script exists and is executable")
323      else:
324          problems += 1
325          print("      ✗ script missing or not executable "
326                "(chmod +x the file, or fix the path)")
327  
328      # 2. Allowlist status
329      entry = shell_hooks.allowlist_entry_for(spec.event, spec.command)
330      if entry:
331          print(f"      ✓ allowlisted (approved {entry.get('approved_at', '?')})")
332      else:
333          problems += 1
334          print("      ✗ not allowlisted — hook will NOT fire at runtime "
335                "(run with --accept-hooks once, or confirm at the TTY prompt)")
336  
337      # 3. Mtime drift
338      if entry and entry.get("script_mtime_at_approval"):
339          mtime_now = shell_hooks.script_mtime_iso(spec.command)
340          mtime_at = entry["script_mtime_at_approval"]
341          if mtime_now and mtime_at and mtime_now > mtime_at:
342              problems += 1
343              print(f"      ⚠ script modified since approval "
344                    f"(was {mtime_at}, now {mtime_now}) — review changes, "
345                    f"then `hermes hooks revoke` + re-approve to refresh")
346          elif mtime_now and mtime_at and mtime_now == mtime_at:
347              print("      ✓ script unchanged since approval")
348  
349      # 4. Produces valid JSON for a synthetic payload — only when the entry
350      # is already allowlisted.  Otherwise `hermes hooks doctor` would execute
351      # every script listed in a freshly-pulled config before the user has
352      # reviewed them, which directly contradicts the documented workflow
353      # ("spot newly-added hooks *before they register*").
354      if not entry:
355          print("      ℹ skipped JSON smoke test — not allowlisted yet. "
356                "Approve the hook first (via TTY prompt or --accept-hooks), "
357                "then re-run `hermes hooks doctor`.")
358      elif shell_hooks.script_is_executable(spec.command):
359          payload = _DEFAULT_PAYLOADS.get(spec.event, {"extra": {}})
360          result = shell_hooks.run_once(spec, payload)
361          if result.get("timed_out"):
362              problems += 1
363              print(f"      ✗ timed out after {result['elapsed_seconds']}s "
364                    f"on synthetic payload (timeout={spec.timeout}s)")
365          elif result.get("error"):
366              problems += 1
367              print(f"      ✗ execution error: {result['error']}")
368          else:
369              rc = result.get("returncode")
370              elapsed = result.get("elapsed_seconds", 0)
371              stdout = (result.get("stdout") or "").strip()
372              if stdout:
373                  try:
374                      json.loads(stdout)
375                      print(f"      ✓ produced valid JSON on synthetic payload "
376                            f"(exit={rc}, {elapsed}s)")
377                  except json.JSONDecodeError:
378                      problems += 1
379                      print(f"      ✗ stdout was not valid JSON (exit={rc}, "
380                            f"{elapsed}s): {_truncate(stdout, 120)}")
381              else:
382                  print(f"      ✓ ran clean with empty stdout "
383                        f"(exit={rc}, {elapsed}s) — hook is observer-only")
384  
385      return problems