perf(observability): gate tool-hook emit on has_hook; slim per-tool footprint

The salvaged observer contract gated the API-request hot path on has_hook()
but left the per-tool emit ungated: every tool call ran result-field
derivation + payload dict build + invoke_hook dispatch even with zero
plugins registered.

- _emit_post_tool_call_hook now short-circuits on has_hook("post_tool_call")
  and derives status/error fields lazily (after the gate, only when a
  listener will consume them). status defaults to None -> derived; explicit
  blocked/cancelled callers still pass status through.
- transform_tool_result emit (pre-existing hook) likewise gated on
  has_hook(); skips _tool_result_observer_fields when no listener.
- Removed the now-redundant _tool_result_observer_fields pre-computation at
  the three ok-path call sites (model_tools, agent_runtime_helpers,
  tool_executor) — the helper derives them, so the no-listener path costs
  one dict lookup and the call sites shrink.
- Tests: stub has_hook=True where payload correctness is asserted; add a
  no-listener regression proving post_tool_call/transform_tool_result emit
  is skipped when nothing is registered.

agent/agent_runtime_helpers.py

@@ -1667,8 +1667,7 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
 
     def _finish_agent_tool(result: Any) -> Any:
         try:
-            from model_tools import _emit_post_tool_call_hook, _tool_result_observer_fields
+            from model_tools import _emit_post_tool_call_hook
-            status, error_type, error_message = _tool_result_observer_fields(result)
             _emit_post_tool_call_hook(
                 function_name=function_name,
                 function_args=function_args,
@@ -1679,9 +1678,6 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
                 turn_id=getattr(agent, "_current_turn_id", "") or "",
                 api_request_id=getattr(agent, "_current_api_request_id", "") or "",
                 duration_ms=int((time.monotonic() - tool_start_time) * 1000),
-                status=status,
-                error_type=error_type,
-                error_message=error_message,
             )
         except Exception:
             pass

agent/tool_executor.py

@@ -72,9 +72,7 @@ def _emit_terminal_post_tool_call(
     error_message: str | None = None,
 ) -> None:
     try:
-        from model_tools import _emit_post_tool_call_hook, _tool_result_observer_fields
+        from model_tools import _emit_post_tool_call_hook
-        if status is None:
-            status, error_type, error_message = _tool_result_observer_fields(result)
         _emit_post_tool_call_hook(
             function_name=function_name,
             function_args=function_args,

model_tools.py

@@ -799,6 +799,16 @@ def _coerce_boolean(value: str):
     return value
 
 
+def _tool_result_observer_fields(result: Any) -> tuple[str, Optional[str], Optional[str]]:
+    try:
+        parsed_result = json.loads(result) if isinstance(result, str) else result
+        if isinstance(parsed_result, dict) and parsed_result.get("error"):
+            return "error", "tool_error", str(parsed_result.get("error"))
+    except Exception:
+        pass
+    return "ok", None, None
+
+
 def _emit_post_tool_call_hook(
     *,
     function_name: str,
@@ -810,12 +820,25 @@ def _emit_post_tool_call_hook(
     turn_id: Optional[str] = None,
     api_request_id: Optional[str] = None,
     duration_ms: int = 0,
-    status: str = "ok",
+    status: Optional[str] = None,
     error_type: Optional[str] = None,
     error_message: Optional[str] = None,
 ) -> None:
+    """Emit the ``post_tool_call`` observer hook.
+
+    No-ops cheaply when no plugin has registered for ``post_tool_call`` —
+    the ``has_hook`` gate skips both the result-field derivation and the
+    payload dispatch so the no-listener path costs one dict lookup.  When
+    ``status`` is not supplied, the ok/error fields are derived from the
+    result *after* the gate (parsing the result is only worth it when a
+    listener will actually consume it).
+    """
     try:
-        from hermes_cli.plugins import invoke_hook
+        from hermes_cli.plugins import has_hook, invoke_hook
+        if not has_hook("post_tool_call"):
+            return
+        if status is None:
+            status, error_type, error_message = _tool_result_observer_fields(result)
         invoke_hook(
             "post_tool_call",
             tool_name=function_name,
@@ -835,16 +858,6 @@ def _emit_post_tool_call_hook(
         logger.debug("post_tool_call hook error: %s", _hook_err)
 
 
-def _tool_result_observer_fields(result: Any) -> tuple[str, Optional[str], Optional[str]]:
-    try:
-        parsed_result = json.loads(result) if isinstance(result, str) else result
-        if isinstance(parsed_result, dict) and parsed_result.get("error"):
-            return "error", "tool_error", str(parsed_result.get("error"))
-    except Exception:
-        pass
-    return "ok", None, None
-
-
 def handle_function_call(
     function_name: str,
     function_args: Dict[str, Any],
@@ -1077,7 +1090,6 @@ def handle_function_call(
                 except Exception:
                     pass
         duration_ms = int((time.monotonic() - _dispatch_start) * 1000)
-        status, error_type, error_message = _tool_result_observer_fields(result)
 
         _emit_post_tool_call_hook(
             function_name=function_name,
@@ -1089,9 +1101,6 @@ def handle_function_call(
             turn_id=turn_id,
             api_request_id=api_request_id,
             duration_ms=duration_ms,
-            status=status,
-            error_type=error_type,
-            error_message=error_message,
         )
 
         # Generic tool-result canonicalization seam: plugins receive the
@@ -1100,8 +1109,12 @@ def handle_function_call(
         # post_tool_call (which stays observational) and before the result
         # is appended back into conversation context. Fail-open; the first
         # valid string return wins; non-string returns are ignored.
+        # Gated on has_hook so the no-listener path skips both the result
+        # field derivation and the payload dispatch.
         try:
-            from hermes_cli.plugins import invoke_hook
+            from hermes_cli.plugins import has_hook, invoke_hook
+            if has_hook("transform_tool_result"):
+                status, error_type, error_message = _tool_result_observer_fields(result)
                 hook_results = invoke_hook(
                     "transform_tool_result",
                     tool_name=function_name,

tests/run_agent/test_run_agent.py

@@ -2067,6 +2067,7 @@ class TestExecuteToolCalls:
             return []
 
         monkeypatch.setattr("hermes_cli.plugins.invoke_hook", _capture_hook)
+        monkeypatch.setattr("hermes_cli.plugins.has_hook", lambda name: True)
 
         with (
             patch("run_agent.handle_function_call", side_effect=KeyboardInterrupt),
@@ -2524,6 +2525,7 @@ class TestConcurrentToolExecution:
             "hermes_cli.plugins.invoke_hook",
             lambda hook_name, **kwargs: hook_calls.append((hook_name, kwargs)) or [],
         )
+        monkeypatch.setattr("hermes_cli.plugins.has_hook", lambda name: True)
 
         with patch("tools.todo_tool.todo_tool", return_value='{"ok":true}') as mock_todo:
             result = agent._invoke_tool("todo", {"todos": []}, "task-1", tool_call_id="todo-1")
@@ -2606,6 +2608,7 @@ class TestConcurrentToolExecution:
             "hermes_cli.plugins.invoke_hook",
             lambda hook_name, **kwargs: hook_calls.append((hook_name, kwargs)) or [],
         )
+        monkeypatch.setattr("hermes_cli.plugins.has_hook", lambda name: True)
 
         with patch("run_agent.handle_function_call", side_effect=AssertionError("should not run")):
             agent._execute_tool_calls_sequential(mock_msg, messages, "task-1")
@@ -2632,6 +2635,7 @@ class TestConcurrentToolExecution:
             "hermes_cli.plugins.invoke_hook",
             lambda hook_name, **kwargs: hook_calls.append((hook_name, kwargs)) or [],
         )
+        monkeypatch.setattr("hermes_cli.plugins.has_hook", lambda name: True)
 
         with patch("tools.todo_tool.todo_tool", return_value='{"ok":true}') as mock_todo:
             agent._execute_tool_calls_sequential(mock_msg, messages, "task-1")

tests/test_model_tools.py

@@ -42,6 +42,7 @@ class TestHandleFunctionCall:
     def test_tool_hooks_receive_session_and_tool_call_ids(self):
         with (
             patch("model_tools.registry.dispatch", return_value='{"ok":true}'),
+            patch("hermes_cli.plugins.has_hook", return_value=True),
             patch("hermes_cli.plugins.invoke_hook") as mock_invoke_hook,
         ):
             result = handle_function_call(
@@ -104,6 +105,7 @@ class TestHandleFunctionCall:
         """
         with (
             patch("model_tools.registry.dispatch", return_value='{"ok":true}'),
+            patch("hermes_cli.plugins.has_hook", return_value=True),
             patch("hermes_cli.plugins.invoke_hook") as mock_invoke_hook,
         ):
             handle_function_call("web_search", {"q": "test"}, task_id="t1")
@@ -123,6 +125,26 @@ class TestHandleFunctionCall:
         # pre_tool_call does NOT get duration_ms (nothing has run yet).
         assert "duration_ms" not in kwargs_by_hook["pre_tool_call"]
 
+    def test_no_listener_skips_post_and_transform_emit(self):
+        """When no plugin is registered for post_tool_call /
+        transform_tool_result, the emit path must short-circuit on
+        ``has_hook`` and never build/dispatch a payload — so the
+        no-listener hot path stays cheap.  ``pre_tool_call`` is always
+        polled (block-check), so it may still fire; the observer/transform
+        emits must not.
+        """
+        with (
+            patch("model_tools.registry.dispatch", return_value='{"ok":true}'),
+            patch("hermes_cli.plugins.has_hook", return_value=False),
+            patch("hermes_cli.plugins.invoke_hook") as mock_invoke_hook,
+        ):
+            result = handle_function_call("web_search", {"q": "test"}, task_id="t1")
+
+        assert result == '{"ok":true}'
+        fired = {c.args[0] for c in mock_invoke_hook.call_args_list}
+        assert "post_tool_call" not in fired
+        assert "transform_tool_result" not in fired
+
 
 # =========================================================================
 # Agent loop tools
@@ -165,6 +187,7 @@ class TestPreToolCallBlocking:
             raise AssertionError("dispatch should not run when blocked")
 
         monkeypatch.setattr("hermes_cli.plugins.invoke_hook", fake_invoke_hook)
+        monkeypatch.setattr("hermes_cli.plugins.has_hook", lambda name: True)
         monkeypatch.setattr("model_tools.registry.dispatch", fake_dispatch)
 
         result = json.loads(handle_function_call("read_file", {"path": "test.txt"}, task_id="t1"))
@@ -228,6 +251,7 @@ class TestPreToolCallBlocking:
             return []
 
         monkeypatch.setattr("hermes_cli.plugins.invoke_hook", fake_invoke_hook)
+        monkeypatch.setattr("hermes_cli.plugins.has_hook", lambda name: True)
         monkeypatch.setattr("model_tools.registry.dispatch",
                             lambda *a, **kw: json.dumps({"ok": True}))