feat(hooks): pre_tool_call content transformation

agent/agent_runtime_helpers.py

@@ -2992,17 +2992,17 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
     block_message: Optional[str] = None
     if not pre_tool_block_checked:
         try:
-            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 "",
+            from hermes_cli.plugins import _dispatch_pre_tool_call_hooks
+            block_message, modified_args = _dispatch_pre_tool_call_hooks(
+                function_name, function_args, task_id=effective_task_id or "",
                 session_id=getattr(agent, "session_id", "") or "",
                 tool_call_id=tool_call_id or "",
                 turn_id=getattr(agent, "_current_turn_id", "") or "",
                 api_request_id=getattr(agent, "_current_api_request_id", "") or "",
                 middleware_trace=list(_tool_middleware_trace),
             )
+            if modified_args is not None:
+                function_args = modified_args
         except Exception:
             block_message = None
     if block_message is not None:

agent/shell_hooks.py

@@ -47,6 +47,12 @@
     # Inject context for pre_llm_call:
     {"context": "Today is Friday"}
 
+    # Modify tool input for pre_tool_call (Hermes-canonical):
+    {"action": "modify", "args": {"new_string": "fixed content"}}
+
+    # Modify tool input for pre_tool_call (Claude-Code-style):
+    {"decision": "modify", "tool_input": {"new_string": "fixed content"}}
+
     # Silent no-op:
     <empty or any non-matching JSON object>
 
@@ -774,6 +780,12 @@ def _parse_response(event: str, stdout: str) -> Optional[Dict[str, Any]]:
     skipping the translation silently breaks every ``pre_tool_call``
     block directive.
 
+    For ``pre_tool_call`` the ``modify`` action (canonical: ``{"action":
+    "modify", "args": {...}}``, Claude-Code-style: ``{"decision":
+    "modify", "tool_input": {...}}``) is translated to
+    ``{"action": "modify", "args": {...}}`` so callers can merge the
+    returned fields into the tool's ``args`` before dispatch.
+
     For ``pre_llm_call``, ``{"context": "..."}`` is passed through
     unchanged to match the existing plugin-hook contract.
 
@@ -800,6 +812,15 @@ def _parse_response(event: str, stdout: str) -> Optional[Dict[str, Any]]:
             return {"action": "block", "message": _block_message(data.get("message"), data.get("reason"))}
         if data.get("decision") == "block":
             return {"action": "block", "message": _block_message(data.get("reason"), data.get("message"))}
+        # "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}
+        if data.get("decision") == "modify":
+            new_args = data.get("tool_input")
+            if isinstance(new_args, dict):
+                return {"action": "modify", "args": new_args}
         return None
 
     if event == "pre_verify":

agent/tool_executor.py

@@ -590,10 +590,11 @@ def _advance_start_order(callback=None) -> None:
             block_error_type = "plugin_block"
 
             def _resolve_pre_tool_block():
+                nonlocal final_args
                 try:
-                    from hermes_cli.plugins import resolve_pre_tool_block
+                    from hermes_cli.plugins import _dispatch_pre_tool_call_hooks
 
-                    return resolve_pre_tool_block(
+                    block_msg, modified_args = _dispatch_pre_tool_call_hooks(
                         function_name,
                         final_args,
                         task_id=effective_task_id or "",
@@ -604,6 +605,10 @@ def _resolve_pre_tool_block():
                         or "",
                         middleware_trace=list(state["middleware_trace"]),
                     )
+                    if modified_args is not None:
+                        final_args = modified_args
+                        state["args"] = modified_args
+                    return block_msg
                 except Exception:
                     return None
 

docs/observability/README.md

@@ -55,7 +55,7 @@ behavior-affecting hooks:
 | Hook | Return behavior |
 | --- | --- |
 | `pre_llm_call` | May return a string or `{"context": "..."}` to inject ephemeral context into the current user message. |
-| `pre_tool_call` | May return `{"action": "block", "message": "..."}` to block a tool before execution. |
+| `pre_tool_call` | May return `{"action": "block", "message": "..."}` to block a tool before execution, or `{"action": "modify", "args": {...}}` to transform the tool's input arguments. |
 | `transform_tool_result` | May return a replacement tool result string after `post_tool_call`. |
 | `transform_llm_output` | May return a replacement final assistant text string. |
 

hermes_cli/plugins.py

@@ -51,7 +51,7 @@
 from dataclasses import dataclass, field
 from functools import wraps
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Mapping, Optional, Set, Union
+from typing import (Any, Callable, Dict, Iterable, List, Mapping, Optional, Set, Tuple, Type, Union)
 
 from hermes_constants import (
     get_hermes_home,
@@ -5950,6 +5950,7 @@ class _PreToolCallDirective:
     action: Optional[str] = None
     message: Optional[str] = None
     rule_key: Optional[str] = None
+    modified_args: Optional[Dict[str, Any]] = None
 
 
 def set_thread_tool_whitelist(
@@ -6022,9 +6023,24 @@ def _get_pre_tool_call_directive_details(
         middleware_trace=list(middleware_trace or []),
     )
 
+    block_msg: Optional[str] = None
+    modified_args: Optional[Dict[str, Any]] = None
+
     for result in hook_results:
         if not isinstance(result, dict):
             continue
+        # "modify" action — transform tool_input before dispatch.
+        # Processed before the block/approve gate so modify directives
+        # are visible even when a later hook blocks. Hooks accumulate:
+        # each modify directive shallow-merges its keys into one
+        # accumulated dict built from the original args on first hit.
+        if result.get("action") == "modify":
+            partial = result.get("args")
+            if isinstance(partial, dict) and partial:
+                if modified_args is None:
+                    modified_args = dict(args) if isinstance(args, dict) else {}
+                modified_args.update(partial)
+            continue
         action = result.get("action")
         if action not in ("block", "approve"):
             continue
@@ -6038,9 +6054,12 @@ 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
-        return _PreToolCallDirective(action=action, message=message, rule_key=rule_key)
+        return _PreToolCallDirective(
+            action=action, message=message, rule_key=rule_key,
+            modified_args=modified_args,
+        )
 
-    return _PreToolCallDirective()
+    return _PreToolCallDirective(modified_args=modified_args)
 
 
 def get_pre_tool_call_directive(
@@ -6165,6 +6184,59 @@ def resolve_pre_tool_block(
     return None
 
 
+def _dispatch_pre_tool_call_hooks(
+    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[Dict[str, Any]]]:
+    """Invoke ``pre_tool_call`` hooks once and process all response types.
+
+    Returns a ``(block_message, modified_args)`` tuple:
+    - ``block_message`` — the first block/approve directive's resolved message
+      (or ``None`` when the call may proceed).  Uses the same approval-gate
+      logic as :func:`resolve_pre_tool_block`.
+    - ``modified_args`` — merged args from the first ``modify`` directive
+      (or ``None`` when no hook requested modification).
+
+    This is the single invocation point for ``pre_tool_call`` hooks.
+    Callers that only need block detection should keep using
+    :func:`get_pre_tool_call_block_message` or
+    :func:`resolve_pre_tool_block` for backward compat.
+    Callers that also need input transformation should call this
+    function and apply ``modified_args`` if not ``None``.
+    """
+    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,
+    )
+    block_msg: Optional[str] = None
+    if details.action == "block":
+        block_msg = details.message
+    elif 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:
+            block_msg = f"BLOCKED: plugin approval gate failed for {tool_name}"
+        else:
+            if not result.get("approved"):
+                block_msg = str(
+                    result.get("message")
+                    or f"BLOCKED: plugin approval required for {tool_name}"
+                )
+    return (block_msg, details.modified_args)
+
+
 def get_pre_verify_continue_message(
     *,
     session_id: str = "",

model_tools.py

@@ -1370,22 +1370,22 @@ def _return_bridge_result(result: Any) -> Any:
         if function_name in _AGENT_LOOP_TOOLS:
             return tool_error(f"{function_name} must be handled by the agent loop")
 
-        # Check plugin hooks for a block/approve directive (unless caller
+        # Check plugin hooks for a block/approve/modify 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. 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
+        # execution. _dispatch_pre_tool_call_hooks() internally calls
+        # invoke_hook("pre_tool_call", ...) once and returns both the block
+        # message (for `block`/`approve` directives) and any modified args
+        # (for `modify` directives). 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 resolve_pre_tool_block
-                block_message = resolve_pre_tool_block(
+                from hermes_cli.plugins import _dispatch_pre_tool_call_hooks
+                block_message, modified_args = _dispatch_pre_tool_call_hooks(
                     function_name,
                     function_args,
                     task_id=task_id or "",
@@ -1395,6 +1395,8 @@ def _return_bridge_result(result: Any) -> Any:
                     api_request_id=api_request_id or "",
                     middleware_trace=list(_tool_middleware_trace),
                 )
+                if modified_args is not None:
+                    function_args = modified_args
             except Exception as _hook_err:
                 logger.debug("pre_tool_call hook error: %s", _hook_err)
 

tests/agent/test_shell_hooks.py

@@ -251,6 +251,34 @@ def test_payload_schema_delivered(self, tmp_path):
 
 
 
+    def test_modify_canonical_parsing(self, tmp_path):
+        """Shell hook returning canonical modify is parsed correctly."""
+        script = _write_script(
+            tmp_path, "mod_canon.sh",
+            "#!/usr/bin/env bash\n"
+            'printf \'{"action": "modify", "args": {"path": "/safe"}}\\n\'',
+        )
+        spec = shell_hooks.ShellHookSpec(
+            event="pre_tool_call", command=str(script),
+        )
+        cb = shell_hooks._make_callback(spec)
+        result = cb(tool_name="write_file", args={"path": "/unsafe"})
+        assert result == {"action": "modify", "args": {"path": "/safe"}}
+
+    def test_modify_claude_code_parsing(self, tmp_path):
+        """Shell hook returning Claude-Code modify is normalised."""
+        script = _write_script(
+            tmp_path, "mod_cc.sh",
+            "#!/usr/bin/env bash\n"
+            'printf \'{"decision": "modify", "tool_input": {"content": "safe"}}\\n\'',
+        )
+        spec = shell_hooks.ShellHookSpec(
+            event="pre_tool_call", command=str(script),
+        )
+        cb = shell_hooks._make_callback(spec)
+        result = cb(tool_name="write_file", args={"content": "danger"})
+        assert result == {"action": "modify", "args": {"content": "safe"}}
+
 
 # ── config parsing ────────────────────────────────────────────────────────