diff --git a/agent/tool_executor.py b/agent/tool_executor.py index 5c13d061d1e74..e1a3f3013df60 100644 --- a/agent/tool_executor.py +++ b/agent/tool_executor.py @@ -339,6 +339,7 @@ class _ManagedToolResult: args: dict[str, Any] middleware_trace: list[dict[str, Any]] blocked: bool + dispatched: bool class _ConcurrentToolAuthorizationGate: @@ -383,12 +384,13 @@ class _ConcurrentToolAuthorizationGate: def _managed_values( outcome: _ManagedToolResult, -) -> tuple[Any, dict[str, Any], list[dict[str, Any]], bool]: +) -> tuple[Any, dict[str, Any], list[dict[str, Any]], bool, bool]: return ( outcome.result, outcome.args, outcome.middleware_trace, outcome.blocked, + outcome.dispatched, ) @@ -571,6 +573,7 @@ def _run_agent_tool_execution_middleware( args=state["args"], middleware_trace=state["middleware_trace"], blocked=bool(state["blocked"]), + dispatched=bool(state["dispatched"]), ) @@ -688,12 +691,27 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe if agent._interrupt_requested: print(f"{agent.log_prefix}⚡ Interrupt: skipping {num_tools} tool call(s)") for tc in tool_calls: + cancelled_result = ( + f"[Tool execution cancelled — {tc.function.name} was skipped " + "due to user interrupt]" + ) messages.append(make_tool_result_message( tc.function.name, - f"[Tool execution cancelled — {tc.function.name} was skipped due to user interrupt]", + cancelled_result, tc.id, effect_disposition="none", )) + _emit_terminal_post_tool_call( + agent, + function_name=tc.function.name, + function_args={}, + result=cancelled_result, + effective_task_id=effective_task_id, + tool_call_id=getattr(tc, "id", "") or "", + status="cancelled", + error_type="user_interrupt", + error_message="Tool execution skipped due to user interrupt", + ) _flush_session_db_after_tool_progress( agent, messages, @@ -840,6 +858,7 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe # submit site below (GHSA-qg5c-hvr5-hjgr, #13617). start = time.time() blocked = False + dispatched = False start_advanced = False def _advance_start(callback=None) -> None: @@ -883,6 +902,7 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe function_args = managed.args middleware_trace = managed.middleware_trace blocked = managed.blocked + dispatched = managed.dispatched except KeyboardInterrupt: try: agent.interrupt("keyboard interrupt") @@ -913,6 +933,17 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe result = f"Error executing tool '{function_name}': {tool_error}" logger.error("_invoke_tool raised for %s: %s", function_name, tool_error, exc_info=True) duration = time.time() - start + if not blocked and not dispatched: + _emit_terminal_post_tool_call( + agent, + function_name=function_name, + function_args=function_args, + result=result, + effective_task_id=effective_task_id, + tool_call_id=getattr(tool_call, "id", "") or "", + duration_ms=int(duration * 1000), + middleware_trace=list(middleware_trace), + ) is_error, _ = _detect_tool_failure(function_name, result) if is_error: logger.info("tool %s failed (%.2fs): %s", function_name, duration, result[:200]) @@ -1162,6 +1193,7 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe result=function_result, effective_task_id=effective_task_id, tool_call_id=getattr(tc, "id", "") or "", + duration_ms=int((timeout_s or 0.0) * 1000), status="timeout", error_type="tool_timeout", error_message=function_result, @@ -1204,6 +1236,19 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe name = function_name args = function_args progress_function_name = function_name + if _parse_error is not None: + _emit_terminal_post_tool_call( + agent, + function_name=function_name, + function_args=function_args, + result=function_result, + effective_task_id=effective_task_id, + tool_call_id=getattr(tc, "id", "") or "", + status="error", + error_type="invalid_tool_arguments", + error_message="Tool arguments must be a valid JSON object", + middleware_trace=list(middleware_trace), + ) if blocked: effect_disposition = "none" @@ -1394,12 +1439,27 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe agent._vprint(f"{agent.log_prefix}⚡ Interrupt: skipping {len(remaining_calls)} tool call(s)", force=True) for skipped_tc in remaining_calls: skipped_name = skipped_tc.function.name + cancelled_result = ( + f"[Tool execution cancelled — {skipped_name} was skipped " + "due to user interrupt]" + ) messages.append(make_tool_result_message( skipped_name, - f"[Tool execution cancelled — {skipped_name} was skipped due to user interrupt]", + cancelled_result, skipped_tc.id, effect_disposition="none", )) + _emit_terminal_post_tool_call( + agent, + function_name=skipped_name, + function_args={}, + result=cancelled_result, + effective_task_id=effective_task_id, + tool_call_id=getattr(skipped_tc, "id", "") or "", + status="cancelled", + error_type="user_interrupt", + error_message="Tool execution skipped due to user interrupt", + ) if not _flush_session_db_after_tool_progress( agent, messages, @@ -1414,6 +1474,17 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe tool_call.function.arguments ) if malformed_args_result is not None: + _emit_terminal_post_tool_call( + agent, + function_name=function_name, + function_args=function_args, + result=malformed_args_result, + effective_task_id=effective_task_id, + tool_call_id=getattr(tool_call, "id", "") or "", + status="error", + error_type="invalid_tool_arguments", + error_message="Tool arguments must be a valid JSON object", + ) messages.append( make_tool_result_message( function_name, @@ -1468,6 +1539,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe middleware_trace: list[dict[str, Any]] = [] _execution_blocked = False + _execution_dispatched = False tool_start_time = time.time() @@ -1479,7 +1551,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe merge=next_args.get("merge", False), store=agent._todo_store, ) - function_result, function_args, middleware_trace, _execution_blocked = _managed_values(_run_agent_tool_execution_middleware( + function_result, function_args, middleware_trace, _execution_blocked, _execution_dispatched = _managed_values(_run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, @@ -1510,7 +1582,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe db=session_db, current_session_id=agent.session_id, ) - function_result, function_args, middleware_trace, _execution_blocked = _managed_values(_run_agent_tool_execution_middleware( + function_result, function_args, middleware_trace, _execution_blocked, _execution_dispatched = _managed_values(_run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, @@ -1549,7 +1621,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe ), ) return result - function_result, function_args, middleware_trace, _execution_blocked = _managed_values(_run_agent_tool_execution_middleware( + function_result, function_args, middleware_trace, _execution_blocked, _execution_dispatched = _managed_values(_run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, @@ -1571,7 +1643,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe multi_select=next_args.get("multi_select", False), callback=agent.clarify_callback, ) - function_result, function_args, middleware_trace, _execution_blocked = _managed_values(_run_agent_tool_execution_middleware( + function_result, function_args, middleware_trace, _execution_blocked, _execution_dispatched = _managed_values(_run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, @@ -1592,7 +1664,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe count=next_args.get("count"), callback=getattr(agent, "read_terminal_callback", None), ) - function_result, function_args, middleware_trace, _execution_blocked = _managed_values(_run_agent_tool_execution_middleware( + function_result, function_args, middleware_trace, _execution_blocked, _execution_dispatched = _managed_values(_run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, @@ -1626,7 +1698,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe try: def _execute(next_args: dict) -> Any: return agent._dispatch_delegate_task(next_args) - function_result, function_args, middleware_trace, _execution_blocked = _managed_values(_run_agent_tool_execution_middleware( + function_result, function_args, middleware_trace, _execution_blocked, _execution_dispatched = _managed_values(_run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, @@ -1659,7 +1731,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe try: def _execute(next_args: dict) -> Any: return agent.context_compressor.handle_tool_call(function_name, next_args, messages=messages) - function_result, function_args, middleware_trace, _execution_blocked = _managed_values(_run_agent_tool_execution_middleware( + function_result, function_args, middleware_trace, _execution_blocked, _execution_dispatched = _managed_values(_run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, @@ -1695,7 +1767,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe try: def _execute(next_args: dict) -> Any: return agent._memory_manager.handle_tool_call(function_name, next_args) - function_result, function_args, middleware_trace, _execution_blocked = _managed_values(_run_agent_tool_execution_middleware( + function_result, function_args, middleware_trace, _execution_blocked, _execution_dispatched = _managed_values(_run_agent_tool_execution_middleware( agent, function_name=function_name, function_args=function_args, @@ -1755,6 +1827,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe function_args, middleware_trace, _execution_blocked, + _execution_dispatched, ) = _managed_values( _run_agent_tool_execution_middleware( agent, @@ -1833,6 +1906,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe function_args, middleware_trace, _execution_blocked, + _execution_dispatched, ) = _managed_values( _run_agent_tool_execution_middleware( agent, @@ -1895,7 +1969,10 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe from agent.agent_runtime_helpers import agent_runtime_owns_post_tool_hook _executor_must_emit_post_hook = ( not _execution_blocked - and agent_runtime_owns_post_tool_hook(agent, function_name) + and ( + not _execution_dispatched + or agent_runtime_owns_post_tool_hook(agent, function_name) + ) ) if _executor_must_emit_post_hook: _emit_terminal_post_tool_call( diff --git a/docs/observability/relay-shared-metrics.md b/docs/observability/relay-shared-metrics.md index c32430c21572b..9454598d6b349 100644 --- a/docs/observability/relay-shared-metrics.md +++ b/docs/observability/relay-shared-metrics.md @@ -55,11 +55,12 @@ dependency does not change the collection or privacy policy. ## Current Slices -The current vertical slices record logical model calls and top-level task runs: +The current vertical slices record logical model calls, top-level task runs, +and tool and approval outcomes: ```text -Hermes turn, API, and tool hooks - -> Relay session, task, and LLM lifecycle +Hermes turn, API, tool, and approval hooks + -> Relay session, task, LLM, tool, and mark lifecycle -> Hermes shared-metrics subscriber -> SQLite counters -> immutable JSON delta package @@ -90,6 +91,21 @@ boundary closes the task for normal returns, early returns, exceptions, and cancellations. Active task ownership follows the task ID if Hermes rotates its conversation session during context compression. +Each tool invocation is represented by a Relay tool lifecycle named +`hermes.tool_call`. The terminal counter contains only bounded tool category, +outcome, approval outcome, latency, and explicit retry-count buckets. Hermes +derives the category from the toolset already declared in its runtime registry; +custom and unrecognized toolsets collapse to `other` rather than exporting +tool or plugin names. Hermes does not infer retries from repeated tool names or +adjacent calls; when the +hook does not provide an explicit retry relationship, the retry bucket is +`unknown`. Approval decisions are emitted as `hermes.tool_approval` marks and +recorded as attributed to a tool call or explicitly `unattributed`. Tool names, +call IDs, arguments, results, commands, descriptions, and error text are not +included in shared-metrics events or packages. A started tool that is still +open when its task terminates is closed as failed, timed out, or cancelled and +remains in the task's tool-count bucket. + Local state is written under: ```text @@ -131,6 +147,8 @@ The script uses the installed `nemo-relay` dependency by default. Pass `--relay-python ../nemo-relay/python` only when testing a locally built Relay binding. -The smoke verifies the model request reached the local server, model and task -counters were stored with the expected model and provider, one package was -exported, and prompt and response canaries are absent from the package. +The smoke has the local model request a real `read_file` tool call before its +final response. It verifies model, provider, task, and bounded tool counters in +SQLite, validates the exported package against the closed schema, and checks +that prompt, response, tool-call ID, and tool-result canaries are absent from +the package. diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index bd603dd0eba66..d391344a6fda7 100644 --- a/hermes_cli/observability/relay_shared_metrics.py +++ b/hermes_cli/observability/relay_shared_metrics.py @@ -6,6 +6,7 @@ import atexit import contextvars import logging import threading +from collections import deque from dataclasses import dataclass, field from time import monotonic_ns from typing import Any, Callable @@ -21,9 +22,15 @@ from .shared_metrics_contract import ( SCHEMA_VERSION, SUBSCRIBER_NAME, TASK_SCOPE, + TOOL_APPROVAL_MARK, + TOOL_CALL_SCOPE, model_call_fields, task_start_fields, task_terminal_fields, + task_terminal_state, + tool_approval_outcome, + tool_category, + tool_terminal_fields, ) from .shared_metrics_subscriber import SharedMetricsSubscriber @@ -36,7 +43,9 @@ HANDLED_HOOKS = frozenset({ "on_session_reset", "pre_llm_call", "pre_api_request", + "pre_tool_call", "post_tool_call", + "post_approval_response", "post_api_request", "api_request_error", "subagent_stop", @@ -62,15 +71,27 @@ class _ModelCall: retry_ordinal: int | None = None +@dataclass +class _ToolCall: + handle: Any + task_id: str + category: str + started_ns: int + approval_outcome: str = "not_required" + + @dataclass class _TaskRun: + task_id: str handle: Any context: contextvars.Context started_ns: int start_fields: dict[str, str] model_call_ids: set[str] = field(default_factory=set) - tool_call_ids: set[str] = field(default_factory=set) + tool_call_ids: set[tuple[str, str, str]] = field(default_factory=set) turn_ids: set[str] = field(default_factory=set) + retired_turn_ids: frozenset[str] = field(default_factory=frozenset) + completed_tool_call_ids: set[tuple[str, str, str]] = field(default_factory=set) unidentified_tool_calls: int = 0 retry_count: int = 0 @@ -83,6 +104,12 @@ class _MetricsSession: closing: bool = False model_calls: dict[tuple[str, str], _ModelCall] = field(default_factory=dict) tasks: dict[str, _TaskRun] = field(default_factory=dict) + tool_calls: dict[tuple[str, str, str, str], _ToolCall] = field( + default_factory=dict + ) + retired_turn_ids: deque[str] = field( + default_factory=lambda: deque(maxlen=256), + ) class _Runtime: @@ -169,7 +196,10 @@ class _Runtime: if session is None: return None with session.lock: - if session.closing or session.relay_session.context is None: + if ( + session.closing + or session.relay_session.context is None + ): return None task_context = session.relay_session.context.copy() start_fields = task_start_fields(event) @@ -195,10 +225,12 @@ class _Runtime: handle = task_context.run(push_task) task = _TaskRun( + task_id=task_id, handle=handle, context=task_context, started_ns=monotonic_ns(), start_fields=start_fields, + retired_turn_ids=frozenset(session.retired_turn_ids), ) session.tasks[task_id] = task with self._task_sessions_lock: @@ -306,8 +338,8 @@ class _Runtime: return model_call.fields = model_call_fields(event) - def record_tool_call(self, event: dict[str, Any]) -> None: - """Count one unique tool invocation under its owning task.""" + def start_tool_call(self, event: dict[str, Any]) -> None: + """Open one privacy-safe Relay tool lifecycle under its task.""" task_id = str(event.get("task_id") or "") session = self._task_session(event, allow_task_id_fallback=True) task = session.tasks.get(task_id) if session is not None else None @@ -317,14 +349,121 @@ class _Runtime: if session is None or task is None: return tool_call_id = str(event.get("tool_call_id") or "") + if not tool_call_id: + return + identity = self._tool_call_identity(event) with session.lock: if session.closing: return + if not self._event_matches_task_turn(task, event): + return + self._remember_turn(session, task, event) + key = (task_id, *identity) + if identity in task.completed_tool_call_ids or key in session.tool_calls: + return + task.tool_call_ids.add(identity) + session.tool_calls[key] = self._open_tool_call(task, event) + + def record_approval(self, event: dict[str, Any]) -> None: + """Record one bounded approval result without approval text or commands.""" + session, task = self._approval_task(event) + if session is None or task is None: + return + outcome = tool_approval_outcome(event) + tool_call_id = str(event.get("tool_call_id") or "") + attribution = "unattributed" + with session.lock: + if session.closing: + return + if not self._event_matches_task_turn(task, event): + return + if tool_call_id: + identity = self._tool_call_identity(event) + tool_call = session.tool_calls.get((task.task_id, *identity)) + if tool_call is None: + matching_keys = [ + key + for key in session.tool_calls + if key[0] == task.task_id + and self._tool_call_identities_are_compatible( + key[1:], + identity, + ) + ] + tool_call = ( + session.tool_calls[matching_keys[0]] + if len(matching_keys) == 1 + else None + ) + if tool_call is not None: + tool_call.approval_outcome = outcome + attribution = "tool_call" + self._run_in_task( + task, + self.relay.scope.event, + TOOL_APPROVAL_MARK, + handle=task.handle, + data={"attribution": attribution, "outcome": outcome}, + metadata=self._event_metadata(), + ) + + def record_tool_call(self, event: dict[str, Any]) -> None: + """Close and count one unique privacy-safe tool lifecycle.""" + task_id = str(event.get("task_id") or "") + session = self._task_session(event, allow_task_id_fallback=True) + task = session.tasks.get(task_id) if session is not None else None + if session is None or task is None: + return + tool_call_id = str(event.get("tool_call_id") or "") + with session.lock: + if session.closing: + return + if not self._event_matches_task_turn(task, event): + return self._remember_turn(session, task, event) if tool_call_id: - task.tool_call_ids.add(tool_call_id) + observed_identity = self._tool_call_identity(event) + if observed_identity in task.completed_tool_call_ids: + return + identity = observed_identity + tool_call = session.tool_calls.pop((task_id, *identity), None) + if tool_call is None: + if any( + self._tool_call_identities_are_compatible( + completed_identity, + observed_identity, + ) + for completed_identity in task.completed_tool_call_ids + ): + return + matching_keys = [ + key + for key in session.tool_calls + if key[0] == task_id + and self._tool_call_identities_are_compatible( + key[1:], + observed_identity, + ) + ] + if len(matching_keys) > 1: + # Partial context cannot safely choose between + # concurrent calls that reused the provider-local ID. + return + if matching_keys: + key = matching_keys[0] + identity = key[1:] + tool_call = session.tool_calls.pop(key) + task.completed_tool_call_ids.update({ + identity, + observed_identity, + }) + task.tool_call_ids.add(identity) else: task.unidentified_tool_calls += 1 + tool_call = None + if tool_call is None: + tool_call = self._open_tool_call(task, event) + self._finish_tool_call(task, tool_call, event) def end_model_call(self, event: dict[str, Any]) -> None: session = self._task_session(event, allow_task_id_fallback=True) @@ -546,6 +685,162 @@ class _Runtime: with self._task_sessions_lock: self._turn_sessions[(session.session_id, turn_id)] = session + @staticmethod + def _tool_call_identity(event: dict[str, Any]) -> tuple[str, str, str]: + """Identify one provider-local tool call without exporting its IDs.""" + return ( + str(event.get("api_request_id") or ""), + str(event.get("turn_id") or ""), + str(event.get("tool_call_id") or ""), + ) + + @staticmethod + def _tool_call_identities_are_compatible( + candidate: tuple[str, str, str], + observed: tuple[str, str, str], + ) -> bool: + """Match partial hook context without crossing known call boundaries.""" + if not observed[2] or candidate[2] != observed[2]: + return False + return all( + not candidate_value + or not observed_value + or candidate_value == observed_value + for candidate_value, observed_value in zip( + candidate[:2], + observed[:2], + strict=True, + ) + ) + + @staticmethod + def _event_matches_task_turn( + task: _TaskRun, + event: dict[str, Any], + ) -> bool: + """Reject delayed hooks from a prior run that reused the task ID.""" + turn_id = str(event.get("turn_id") or "") + if not turn_id: + return True + if turn_id in task.retired_turn_ids: + return False + return not task.turn_ids or turn_id in task.turn_ids + + def _approval_task( + self, + event: dict[str, Any], + ) -> tuple[_MetricsSession | None, _TaskRun | None]: + """Resolve approval correlation without guessing across ambiguous turns.""" + active = relay_runtime.active_turn() + if active is not None: + correlated = { + **event, + "session_id": active.lease.session_id, + "task_id": active.task_id, + } + session = self._task_session(correlated) + task = session.tasks.get(active.task_id) if session is not None else None + if task is not None: + return session, task + + session = self._task_session(event) + task_id = str(event.get("task_id") or "") + task = session.tasks.get(task_id) if session is not None else None + if task is not None: + return session, task + + turn_id = str(event.get("turn_id") or "") + if not turn_id: + return None, None + with self._task_sessions_lock: + candidates = [ + candidate + for ( + candidate_session_id, + candidate_turn_id, + ), candidate in self._turn_sessions.items() + if candidate_turn_id == turn_id + and self._sessions.get(candidate_session_id) is candidate + ] + unique_sessions = {id(candidate): candidate for candidate in candidates} + if len(unique_sessions) != 1: + return None, None + session = next(iter(unique_sessions.values())) + matching_tasks = [ + candidate + for candidate in session.tasks.values() + if turn_id in candidate.turn_ids + ] + if len(matching_tasks) != 1: + return None, None + return session, matching_tasks[0] + + def _open_tool_call( + self, + task: _TaskRun, + event: dict[str, Any], + ) -> _ToolCall: + handle = self._run_in_task( + task, + self.relay.tools.call, + TOOL_CALL_SCOPE, + {}, + handle=task.handle, + metadata=self._event_metadata(), + ) + return _ToolCall( + handle=handle, + task_id=task.task_id, + category=tool_category(event), + started_ns=monotonic_ns(), + ) + + def _finish_tool_call( + self, + task: _TaskRun, + tool_call: _ToolCall, + event: dict[str, Any], + ) -> None: + fields = tool_terminal_fields( + event, + category=tool_call.category, + approval_outcome=tool_call.approval_outcome, + fallback_duration_ms=max( + 0, + (monotonic_ns() - tool_call.started_ns) // 1_000_000, + ), + ) + try: + self._run_in_task( + task, + self.relay.tools.call_end, + tool_call.handle, + fields, + metadata=self._event_metadata(), + ) + except Exception: + logger.warning( + "Hermes shared-metrics tool call close failed", + exc_info=True, + ) + + def _end_pending_tool_calls( + self, + session: _MetricsSession, + task: _TaskRun, + event: dict[str, Any], + ) -> None: + pending_keys = [key for key in session.tool_calls if key[0] == task.task_id] + task_outcome, _, _ = task_terminal_state(event) + status = { + "cancelled": "cancelled", + "timed_out": "timeout", + }.get(task_outcome, "error") + for key in pending_keys: + tool_call = session.tool_calls.pop(key, None) + if tool_call is not None: + self._finish_tool_call(task, tool_call, {**event, "status": status}) + def _finish_model_call( self, session: _MetricsSession, @@ -628,6 +923,7 @@ class _Runtime: task = session.tasks.get(task_id) if task is None: return False + self._end_pending_tool_calls(session, task, event) self._end_pending_model_calls(session, {**event, "task_id": task_id}) fields = task_terminal_fields( {**task.start_fields, **event}, @@ -648,6 +944,7 @@ class _Runtime: logger.warning("Hermes shared-metrics task close failed", exc_info=True) finally: session.tasks.pop(task_id, None) + session.retired_turn_ids.extend(task.turn_ids) with self._task_sessions_lock: task_key = (session.session_id, task_id) if self._task_sessions.get(task_key) is session: @@ -697,8 +994,7 @@ def enabled() -> bool: telemetry.get("shared_metrics") if isinstance(telemetry, dict) else None ) value = ( - isinstance(shared_metrics, dict) - and shared_metrics.get("enabled") is True + isinstance(shared_metrics, dict) and shared_metrics.get("enabled") is True ) if value: return True @@ -729,8 +1025,12 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None: runtime.start_task(kwargs) elif hook_name == "pre_api_request": runtime.start_model_call(kwargs) + elif hook_name == "pre_tool_call": + runtime.start_tool_call(_with_runtime_toolset(kwargs)) elif hook_name == "post_tool_call": - runtime.record_tool_call(kwargs) + runtime.record_tool_call(_with_runtime_toolset(kwargs)) + elif hook_name == "post_approval_response": + runtime.record_approval(kwargs) elif hook_name == "post_api_request": runtime.end_model_call(kwargs) elif hook_name == "api_request_error": @@ -749,6 +1049,22 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None: ) +def _with_runtime_toolset(event: dict[str, Any]) -> dict[str, Any]: + """Attach the toolset already declared by Hermes's runtime registry.""" + if event.get("toolset"): + return event + tool_name = str(event.get("tool_name") or "") + if not tool_name: + return event + try: + from model_tools import get_toolset_for_tool + + toolset = get_toolset_for_tool(tool_name) + except Exception: + toolset = None + return {**event, "toolset": toolset or "other"} + + def prepare_session_start() -> None: """Register the subscriber before any producer opens the session scope.""" if enabled(): diff --git a/hermes_cli/observability/schemas/hermes.shared_metrics.v2.schema.json b/hermes_cli/observability/schemas/hermes.shared_metrics.v2.schema.json index 112b9f3ce8698..da8e7c78d0c0f 100644 --- a/hermes_cli/observability/schemas/hermes.shared_metrics.v2.schema.json +++ b/hermes_cli/observability/schemas/hermes.shared_metrics.v2.schema.json @@ -67,6 +67,12 @@ }, { "$ref": "#/$defs/task_finished_counter" + }, + { + "$ref": "#/$defs/tool_call_counter" + }, + { + "$ref": "#/$defs/tool_approval_counter" } ] } @@ -380,6 +386,159 @@ "6_to_10", "gte_11" ] + }, + "tool_latency_bucket": { + "enum": [ + "100ms_to_250ms", + "10s_to_30s", + "1s_to_2s", + "250ms_to_500ms", + "2s_to_5s", + "500ms_to_1s", + "5s_to_10s", + "gte_30s", + "lt_100ms", + "unknown" + ] + }, + "tool_retry_bucket": { + "enum": [ + "0", + "1", + "2", + "3_to_5", + "6_to_10", + "gte_11", + "unknown" + ] + }, + "tool_call_counter": { + "type": "object", + "additionalProperties": false, + "required": [ + "name", + "type", + "dimensions", + "value" + ], + "properties": { + "name": { + "const": "hermes.tool_call.count" + }, + "type": { + "const": "counter" + }, + "dimensions": { + "type": "object", + "additionalProperties": false, + "required": [ + "approval_outcome", + "latency_bucket", + "outcome", + "retry_count_bucket", + "tool_category" + ], + "properties": { + "approval_outcome": { + "enum": [ + "approved", + "denied", + "not_required", + "timed_out", + "unknown" + ] + }, + "latency_bucket": { + "$ref": "#/$defs/tool_latency_bucket" + }, + "outcome": { + "enum": [ + "blocked", + "cancelled", + "failed", + "success", + "timed_out", + "unknown" + ] + }, + "retry_count_bucket": { + "$ref": "#/$defs/tool_retry_bucket" + }, + "tool_category": { + "enum": [ + "browser", + "code_execution", + "communication", + "computer_use", + "delegation", + "file", + "home_automation", + "mcp", + "media", + "memory", + "other", + "planning", + "project", + "scheduler", + "skill", + "terminal", + "unknown", + "web" + ] + } + } + }, + "value": { + "type": "integer", + "minimum": 1 + } + } + }, + "tool_approval_counter": { + "type": "object", + "additionalProperties": false, + "required": [ + "name", + "type", + "dimensions", + "value" + ], + "properties": { + "name": { + "const": "hermes.tool_approval.count" + }, + "type": { + "const": "counter" + }, + "dimensions": { + "type": "object", + "additionalProperties": false, + "required": [ + "attribution", + "outcome" + ], + "properties": { + "attribution": { + "enum": [ + "tool_call", + "unattributed" + ] + }, + "outcome": { + "enum": [ + "approved", + "denied", + "timed_out", + "unknown" + ] + } + } + }, + "value": { + "type": "integer", + "minimum": 1 + } + } } } } diff --git a/hermes_cli/observability/shared_metrics_contract.py b/hermes_cli/observability/shared_metrics_contract.py index bb48952cb89d0..870152634f53c 100644 --- a/hermes_cli/observability/shared_metrics_contract.py +++ b/hermes_cli/observability/shared_metrics_contract.py @@ -2,6 +2,7 @@ from __future__ import annotations +from math import isfinite from typing import Any from agent.relay_runtime import ( @@ -16,11 +17,15 @@ SCHEMA_VERSION = "hermes.metrics.event.v2" MODEL_CALL_SCOPE = "hermes.model_call" MODEL_CALL_PROFILE_MODEL = "unknown" TASK_SCOPE = "hermes.task_run" +TOOL_CALL_SCOPE = "hermes.tool_call" +TOOL_APPROVAL_MARK = "hermes.tool_approval" SUBSCRIBER_NAME = "hermes.nemo_relay.shared_metrics" LEGACY_MODEL_CALL_METRIC = "hermes.model_call.count" MODEL_ROUTE_METRIC = "hermes.model_route.count" TASK_STARTED_METRIC = "hermes.task_run.started" TASK_FINISHED_METRIC = "hermes.task_run.finished" +TOOL_CALL_METRIC = "hermes.tool_call.count" +TOOL_APPROVAL_METRIC = "hermes.tool_approval.count" MODEL_IDENTIFIER_MAX_LENGTH = 256 PROVIDER_IDENTIFIER_MAX_LENGTH = 64 _METRIC_IDENTIFIER_CHARACTERS = frozenset( @@ -95,6 +100,58 @@ COUNT_BUCKETS: frozenset[str] = frozenset({ "6_to_10", "gte_11", }) +TOOL_CATEGORIES: frozenset[str] = frozenset({ + "browser", + "code_execution", + "communication", + "computer_use", + "delegation", + "file", + "home_automation", + "mcp", + "media", + "memory", + "other", + "planning", + "project", + "scheduler", + "skill", + "terminal", + "unknown", + "web", +}) +TOOL_OUTCOMES: frozenset[str] = frozenset({ + "blocked", + "cancelled", + "failed", + "success", + "timed_out", + "unknown", +}) +TOOL_APPROVAL_OUTCOMES: frozenset[str] = frozenset({ + "approved", + "denied", + "not_required", + "timed_out", + "unknown", +}) +TOOL_APPROVAL_ATTRIBUTIONS: frozenset[str] = frozenset({ + "tool_call", + "unattributed", +}) +TOOL_LATENCY_BUCKETS: frozenset[str] = frozenset({ + "100ms_to_250ms", + "10s_to_30s", + "1s_to_2s", + "250ms_to_500ms", + "2s_to_5s", + "500ms_to_1s", + "5s_to_10s", + "gte_30s", + "lt_100ms", + "unknown", +}) +TOOL_RETRY_BUCKETS: frozenset[str] = COUNT_BUCKETS | frozenset({"unknown"}) _LEGACY_PROVIDER_FAMILIES = frozenset({ "aggregator", @@ -153,11 +210,24 @@ _COUNTER_DIMENSION_VALUES: dict[str, dict[str, frozenset[str]]] = { "termination": TASK_TERMINATIONS, "tool_call_count_bucket": COUNT_BUCKETS, }, + TOOL_CALL_METRIC: { + "approval_outcome": TOOL_APPROVAL_OUTCOMES, + "latency_bucket": TOOL_LATENCY_BUCKETS, + "outcome": TOOL_OUTCOMES, + "retry_count_bucket": TOOL_RETRY_BUCKETS, + "tool_category": TOOL_CATEGORIES, + }, + TOOL_APPROVAL_METRIC: { + "attribution": TOOL_APPROVAL_ATTRIBUTIONS, + "outcome": TOOL_APPROVAL_OUTCOMES - {"not_required"}, + }, } COUNTER_METRICS: frozenset[str] = frozenset({ MODEL_ROUTE_METRIC, TASK_FINISHED_METRIC, TASK_STARTED_METRIC, + TOOL_APPROVAL_METRIC, + TOOL_CALL_METRIC, }) @@ -184,25 +254,27 @@ def counter_dimensions_are_valid( if contract is None or set(dimensions) != set(contract): return False return all( - isinstance(dimensions[field], str) - and dimensions[field] in allowed_values + isinstance(dimensions[field], str) and dimensions[field] in allowed_values for field, allowed_values in contract.items() ) +def _event_metadata_is_valid(event: Any) -> bool: + metadata = getattr(event, "metadata", None) + if not isinstance(metadata, dict) or metadata.get(SCHEMA_KEY) != SCHEMA_VERSION: + return False + relay_metadata = set(metadata) - {SCHEMA_KEY, RUNTIME_INSTANCE_KEY} + return not relay_metadata - {"otel.status_code"} and metadata.get( + "otel.status_code", "OK" + ) in {"OK", "ERROR"} + + def model_call_dimensions(event: Any) -> dict[str, str] | None: """Return package dimensions for one valid logical model-call end event.""" auxiliary = _auxiliary_model_call_dimensions(event) if auxiliary is not None: return auxiliary - - metadata = getattr(event, "metadata", None) - if not isinstance(metadata, dict) or metadata.get(SCHEMA_KEY) != SCHEMA_VERSION: - return None - relay_metadata = set(metadata) - {SCHEMA_KEY, RUNTIME_INSTANCE_KEY} - if relay_metadata - {"otel.status_code"} or metadata.get( - "otel.status_code", "OK" - ) not in {"OK", "ERROR"}: + if not _event_metadata_is_valid(event): return None if ( str(getattr(event, "kind", "") or "") != "scope" @@ -277,13 +349,7 @@ def _auxiliary_model_call_dimensions(event: Any) -> dict[str, str] | None: def task_counter(event: Any) -> tuple[str, dict[str, str]] | None: """Return one validated task counter from a task scope event.""" - metadata = getattr(event, "metadata", None) - if not isinstance(metadata, dict) or metadata.get(SCHEMA_KEY) != SCHEMA_VERSION: - return None - relay_metadata = set(metadata) - {SCHEMA_KEY, RUNTIME_INSTANCE_KEY} - if relay_metadata - {"otel.status_code"} or metadata.get( - "otel.status_code", "OK" - ) not in {"OK", "ERROR"}: + if not _event_metadata_is_valid(event): return None if ( str(getattr(event, "kind", "") or "") != "scope" @@ -331,6 +397,56 @@ def task_counter(event: Any) -> tuple[str, dict[str, str]] | None: return TASK_FINISHED_METRIC, dimensions +def tool_call_dimensions(event: Any) -> dict[str, str] | None: + """Return package dimensions for one allowlisted tool lifecycle end event.""" + if not _event_metadata_is_valid(event): + return None + if ( + str(getattr(event, "kind", "") or "") != "scope" + or str(getattr(event, "category", "") or "") != "tool" + or str(getattr(event, "name", "") or "") != TOOL_CALL_SCOPE + or str(getattr(event, "scope_category", "") or "") != "end" + or getattr(event, "category_profile", None) != {} + ): + return None + data = getattr(event, "data", None) + expected_fields = { + "approval_outcome", + "latency_bucket", + "outcome", + "retry_count_bucket", + "tool_category", + } + if not isinstance(data, dict) or set(data) != expected_fields: + return None + dimensions = {field: data.get(field) for field in sorted(expected_fields)} + if not counter_dimensions_are_valid(TOOL_CALL_METRIC, dimensions): + return None + return dimensions + + +def tool_approval_counter(event: Any) -> tuple[str, dict[str, str]] | None: + """Return one validated approval counter from a safe Relay mark event.""" + if not _event_metadata_is_valid(event): + return None + if ( + str(getattr(event, "kind", "") or "") != "mark" + or str(getattr(event, "name", "") or "") != TOOL_APPROVAL_MARK + or getattr(event, "category", None) is not None + or getattr(event, "scope_category", None) is not None + or getattr(event, "category_profile", None) is not None + ): + return None + data = getattr(event, "data", None) + expected_fields = {"attribution", "outcome"} + if not isinstance(data, dict) or set(data) != expected_fields: + return None + dimensions = {field: data.get(field) for field in sorted(expected_fields)} + if not counter_dimensions_are_valid(TOOL_APPROVAL_METRIC, dimensions): + return None + return TOOL_APPROVAL_METRIC, dimensions + + def execution_surface(kwargs: dict[str, Any]) -> str: """Normalize the safe session surface carried by the parent Relay scope.""" value = ( @@ -459,6 +575,138 @@ def count_bucket(count: int) -> str: return "gte_11" +def tool_category(kwargs: dict[str, Any]) -> str: + """Map Hermes registry toolset metadata to a low-cardinality category.""" + toolset = str(kwargs.get("toolset") or "").strip().lower() + if not toolset: + return "unknown" + if toolset in TOOL_CATEGORIES: + return toolset + if toolset.startswith("mcp"): + return "mcp" + if toolset.startswith("browser"): + return "browser" + if toolset.startswith(("image", "tts", "video", "vision")): + return "media" + if toolset.startswith("homeassistant"): + return "home_automation" + if toolset in {"clarify", "kanban", "todo"}: + return "planning" + if toolset == "session_search": + return "memory" + if toolset == "cronjob": + return "scheduler" + if toolset == "skills": + return "skill" + if toolset == "x_search": + return "web" + if toolset.startswith( + ("discord", "email", "feishu", "hermes-yuanbao", "slack", "sms") + ): + return "communication" + return "other" + + +def tool_outcome(kwargs: dict[str, Any]) -> str: + """Normalize the terminal Hermes tool status without inspecting its result.""" + status = str(kwargs.get("status") or "").strip().lower() + return { + "blocked": "blocked", + "cancelled": "cancelled", + "error": "failed", + "failed": "failed", + "ok": "success", + "success": "success", + "timed_out": "timed_out", + "timeout": "timed_out", + }.get(status, "unknown") + + +def tool_approval_outcome(kwargs: dict[str, Any]) -> str: + """Normalize a terminal approval choice to a bounded outcome.""" + choice = str(kwargs.get("choice") or "").strip().lower() + if choice in {"always", "approve", "approved", "once", "session", "smart_approve"}: + return "approved" + if choice in {"deny", "denied", "smart_deny"}: + return "denied" + if choice in {"timed_out", "timeout"}: + return "timed_out" + return "unknown" + + +def tool_terminal_fields( + kwargs: dict[str, Any], + *, + category: str | None = None, + approval_outcome: str = "not_required", + fallback_duration_ms: int | None = None, +) -> dict[str, str]: + """Build one bounded tool-call terminal payload.""" + return { + "approval_outcome": ( + approval_outcome + if approval_outcome in TOOL_APPROVAL_OUTCOMES + else "unknown" + ), + "latency_bucket": tool_latency_bucket( + kwargs.get("duration_ms"), + fallback_duration_ms=fallback_duration_ms, + ), + "outcome": tool_outcome(kwargs), + "retry_count_bucket": tool_retry_bucket(kwargs.get("retry_count")), + "tool_category": ( + category if category in TOOL_CATEGORIES else tool_category(kwargs) + ), + } + + +def tool_latency_bucket( + value: Any, + *, + fallback_duration_ms: int | None = None, +) -> str: + """Bucket a tool duration reported in milliseconds.""" + duration_ms = _non_negative_number(value) + if duration_ms is None: + duration_ms = _non_negative_number(fallback_duration_ms) + if duration_ms is None: + return "unknown" + if duration_ms < 100: + return "lt_100ms" + if duration_ms < 250: + return "100ms_to_250ms" + if duration_ms < 500: + return "250ms_to_500ms" + if duration_ms < 1_000: + return "500ms_to_1s" + if duration_ms < 2_000: + return "1s_to_2s" + if duration_ms < 5_000: + return "2s_to_5s" + if duration_ms < 10_000: + return "5s_to_10s" + if duration_ms < 30_000: + return "10s_to_30s" + return "gte_30s" + + +def tool_retry_bucket(value: Any) -> str: + """Bucket only explicit tool retries; missing relationships stay unknown.""" + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + return "unknown" + return count_bucket(value) + + +def _non_negative_number(value: Any) -> float | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + try: + number = float(value) + except (OverflowError, TypeError, ValueError): + return None + return number if isfinite(number) and number >= 0 else None + + def model_call_fields(kwargs: dict[str, Any]) -> dict[str, str]: """Return the terminal model identity and provider route known to Hermes.""" model = _metric_identifier( diff --git a/hermes_cli/observability/shared_metrics_subscriber.py b/hermes_cli/observability/shared_metrics_subscriber.py index 259a6f6b4a2d5..086f88d4c23b6 100644 --- a/hermes_cli/observability/shared_metrics_subscriber.py +++ b/hermes_cli/observability/shared_metrics_subscriber.py @@ -9,7 +9,14 @@ from typing import Any from agent.relay_runtime import RUNTIME_INSTANCE_KEY from .shared_metrics import SharedMetricsStore -from .shared_metrics_contract import MODEL_ROUTE_METRIC, model_call_dimensions, task_counter +from .shared_metrics_contract import ( + MODEL_ROUTE_METRIC, + TOOL_CALL_METRIC, + model_call_dimensions, + task_counter, + tool_approval_counter, + tool_call_dimensions, +) logger = logging.getLogger(__name__) @@ -46,10 +53,13 @@ class SharedMetricsSubscriber: dimensions = model_call_dimensions(event) metric_name = MODEL_ROUTE_METRIC if dimensions is None: - task_metric = task_counter(event) - if task_metric is None: + dimensions = tool_call_dimensions(event) + metric_name = TOOL_CALL_METRIC + if dimensions is None: + metric = task_counter(event) or tool_approval_counter(event) + if metric is None: return - metric_name, dimensions = task_metric + metric_name, dimensions = metric with self._lock: if not self._active: return diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 8dd02a1bdf5ec..23ed3c26e3c3e 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -2161,7 +2161,9 @@ def _get_pre_tool_call_directive_details( message=fmt.format(tool_name=tool_name), ) - hook_results = invoke_hook( + from hermes_cli.lifecycle import invoke_hook as invoke_lifecycle_hook + + hook_results = invoke_lifecycle_hook( "pre_tool_call", tool_name=tool_name, args=args if isinstance(args, dict) else {}, @@ -2277,12 +2279,32 @@ def resolve_pre_tool_block( return details.message if 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, + from tools.approval import ( + request_tool_approval, + reset_current_observability_context, + set_current_observability_context, ) + + approval_tokens = None + try: + approval_tokens = set_current_observability_context( + turn_id=turn_id, + tool_call_id=tool_call_id, + ) + except Exception: + pass + try: + result = request_tool_approval( + tool_name, + details.message or "", + rule_key=details.rule_key or tool_name, + ) + finally: + if approval_tokens is not None: + try: + reset_current_observability_context(approval_tokens) + except Exception: + pass except Exception: # Fail-closed: if the gate itself errors, block rather than # silently execute an action a plugin flagged for approval. diff --git a/model_tools.py b/model_tools.py index 40a8f83b3ec88..0f1290258985e 100644 --- a/model_tools.py +++ b/model_tools.py @@ -1032,13 +1032,24 @@ def _coerce_boolean(value: str): return value -def _tool_result_observer_fields(result: Any) -> tuple[str, Optional[str], Optional[str]]: +def _tool_result_observer_fields( + tool_name: str, + 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 + try: + from agent.display import _detect_tool_failure + + failed, suffix = _detect_tool_failure(tool_name, result) + if failed: + return "error", "tool_error", suffix.strip().strip("[]") or None + except Exception: + pass return "ok", None, None @@ -1072,7 +1083,10 @@ def _emit_post_tool_call_hook( if not has_hook("post_tool_call"): return if status is None: - status, error_type, error_message = _tool_result_observer_fields(result) + status, error_type, error_message = _tool_result_observer_fields( + function_name, + result, + ) invoke_hook( "post_tool_call", tool_name=function_name, @@ -1145,6 +1159,23 @@ def handle_function_call( # inline. tool_call is unwrapped to the underlying tool so that every # downstream hook (pre/post, edit approval, guardrails) sees the real # tool name, not the bridge. + _dispatch_start = time.monotonic() + + def _return_bridge_result(result: Any) -> Any: + _emit_post_tool_call_hook( + function_name=function_name, + function_args=function_args, + result=result, + task_id=task_id, + session_id=session_id, + tool_call_id=tool_call_id, + turn_id=turn_id, + api_request_id=api_request_id, + duration_ms=int((time.monotonic() - _dispatch_start) * 1000), + middleware_trace=list(_tool_middleware_trace), + ) + return result + _ts_mod = None try: from tools import tool_search as _ts_mod # noqa: F401 @@ -1174,15 +1205,25 @@ def handle_function_call( except Exception: current_defs = [] if function_name == _ts_mod.TOOL_SEARCH_NAME: - return _ts_mod.dispatch_tool_search(function_args or {}, - current_tool_defs=current_defs) + return _return_bridge_result( + _ts_mod.dispatch_tool_search( + function_args or {}, + current_tool_defs=current_defs, + ) + ) if function_name == _ts_mod.TOOL_DESCRIBE_NAME: - return _ts_mod.dispatch_tool_describe(function_args or {}, - current_tool_defs=current_defs) + return _return_bridge_result( + _ts_mod.dispatch_tool_describe( + function_args or {}, + current_tool_defs=current_defs, + ) + ) if function_name == _ts_mod.TOOL_CALL_NAME: underlying_name, underlying_args, err = _ts_mod.resolve_underlying_call(function_args or {}) if err or not underlying_name: - return tool_error(err or "tool_call could not be resolved") + return _return_bridge_result( + tool_error(err or "tool_call could not be resolved") + ) # Defense in depth: the underlying tool MUST be in the session's # scoped deferrable catalog. resolve_underlying_call() only checks # that the name is deferrable in the global registry; this gate @@ -1191,16 +1232,18 @@ def handle_function_call( # the bridge even if the catalog scoping above regressed. _scoped_deferrable = _ts_mod.scoped_deferrable_names(current_defs) if underlying_name not in _scoped_deferrable: - return tool_error( - f"'{underlying_name}' is not available in this session. " - "Use tool_search to find tools you can call." + return _return_bridge_result( + tool_error( + f"'{underlying_name}' is not available in this session. " + "Use tool_search to find tools you can call." + ) ) # Probe-validate against the deferred tool's schema (ironclaw#5149): # a blind call missing required arguments returns the parameter # schema instead of dispatching into an opaque downstream failure. _probe_err = _ts_mod.validate_deferred_call_args(underlying_name, underlying_args) if _probe_err is not None: - return _probe_err + return _return_bridge_result(_probe_err) # Recurse with the underlying tool. All hooks fire against the # real tool name. The bridge is invisible to hooks by design. return handle_function_call( @@ -1209,6 +1252,8 @@ def handle_function_call( task_id=task_id, tool_call_id=tool_call_id, session_id=session_id, + turn_id=turn_id, + api_request_id=api_request_id, user_task=user_task, enabled_tools=enabled_tools, skip_pre_tool_call_hook=skip_pre_tool_call_hook, @@ -1297,11 +1342,38 @@ def handle_function_call( edit_block_message = maybe_require_edit_approval(function_name, function_args) if edit_block_message is not None: + _emit_post_tool_call_hook( + function_name=function_name, + function_args=function_args, + result=edit_block_message, + task_id=task_id, + session_id=session_id, + tool_call_id=tool_call_id, + turn_id=turn_id, + api_request_id=api_request_id, + status="blocked", + error_type="edit_approval_denied", + middleware_trace=list(_tool_middleware_trace), + ) return edit_block_message except Exception as _edit_approval_err: logger.debug("ACP edit approval guard error: %s", _edit_approval_err) if function_name in {"write_file", "patch"}: - return tool_error("Edit approval denied: approval guard failed") + result = tool_error("Edit approval denied: approval guard failed") + _emit_post_tool_call_hook( + function_name=function_name, + function_args=function_args, + result=result, + task_id=task_id, + session_id=session_id, + tool_call_id=tool_call_id, + turn_id=turn_id, + api_request_id=api_request_id, + status="blocked", + error_type="edit_approval_error", + middleware_trace=list(_tool_middleware_trace), + ) + return result # Notify the read-loop tracker when a non-read/search tool runs, # so the *consecutive* counter resets (reads after other work are fine). @@ -1400,7 +1472,10 @@ def handle_function_call( try: from hermes_cli.lifecycle import has_hook, invoke_hook if has_hook("transform_tool_result"): - status, error_type, error_message = _tool_result_observer_fields(result) + status, error_type, error_message = _tool_result_observer_fields( + function_name, + result, + ) hook_results = invoke_hook( "transform_tool_result", tool_name=function_name, @@ -1428,7 +1503,28 @@ def handle_function_call( except Exception as e: error_msg = f"Error executing {function_name}: {str(e)}" logger.exception(error_msg) - return tool_error(_sanitize_tool_error(error_msg)) + result = tool_error(_sanitize_tool_error(error_msg)) + duration_ms = ( + int((time.monotonic() - _dispatch_start) * 1000) + if _dispatch_start is not None + else 0 + ) + _emit_post_tool_call_hook( + function_name=function_name, + function_args=function_args, + result=result, + task_id=task_id, + session_id=session_id, + tool_call_id=tool_call_id, + turn_id=turn_id, + api_request_id=api_request_id, + duration_ms=duration_ms, + status="error", + error_type=type(e).__name__, + error_message=str(e), + middleware_trace=list(_tool_middleware_trace), + ) + return result # ============================================================================= diff --git a/scripts/smoke_nemo_relay_shared_metrics.py b/scripts/smoke_nemo_relay_shared_metrics.py index a8bf4adb0121a..64c69bdc3c068 100644 --- a/scripts/smoke_nemo_relay_shared_metrics.py +++ b/scripts/smoke_nemo_relay_shared_metrics.py @@ -20,6 +20,9 @@ from typing import Any PROMPT_CANARY = "relay-smoke-sensitive-prompt" MODEL_CANARY = "gpt-relay-smoke-sensitive-model" RESPONSE_CANARY = "relay-smoke-sensitive-response" +TOOL_CALL_CANARY = "relay-smoke-sensitive-tool-call" +TOOL_RESULT_CANARY = "relay-smoke-sensitive-tool-result" +TOOL_FILE = "relay-smoke-input.txt" def _resolve_hermes_executable(hermes_repo: Path) -> Path: @@ -68,30 +71,51 @@ class _ModelHandler(BaseHTTPRequestHandler): length = int(self.headers.get("Content-Length", "0")) request = json.loads(self.rfile.read(length) or b"{}") type(self).requests.append(request) + request_tool = not any( + message.get("role") == "tool" + for message in request.get("messages") or [] + if isinstance(message, dict) + ) if request.get("stream"): - self._write_stream() + self._write_stream(request_tool=request_tool) else: - self._write_json({ - "id": "chatcmpl-relay-smoke", - "object": "chat.completion", - "created": int(time.time()), - "model": MODEL_CANARY, - "choices": [ - { - "index": 0, - "message": { - "role": "assistant", - "content": RESPONSE_CANARY, - }, - "finish_reason": "stop", - } - ], - "usage": { - "prompt_tokens": 10, - "completion_tokens": 1, - "total_tokens": 11, - }, - }) + self._write_json(self._completion(request_tool=request_tool)) + + def _completion(self, *, request_tool: bool) -> dict[str, Any]: + message: dict[str, Any] = { + "role": "assistant", + "content": "" if request_tool else RESPONSE_CANARY, + } + finish_reason = "tool_calls" if request_tool else "stop" + if request_tool: + message["tool_calls"] = [ + { + "id": TOOL_CALL_CANARY, + "type": "function", + "function": { + "name": "read_file", + "arguments": json.dumps({"path": TOOL_FILE}), + }, + } + ] + return { + "id": "chatcmpl-relay-smoke", + "object": "chat.completion", + "created": int(time.time()), + "model": MODEL_CANARY, + "choices": [ + { + "index": 0, + "message": message, + "finish_reason": finish_reason, + } + ], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 1, + "total_tokens": 11, + }, + } def log_message(self, format: str, *args: Any) -> None: return @@ -106,9 +130,9 @@ class _ModelHandler(BaseHTTPRequestHandler): self.wfile.write(body) self.close_connection = True - def _write_stream(self) -> None: + def _write_stream(self, *, request_tool: bool) -> None: now = int(time.time()) - chunks = [ + chunks: list[dict[str, Any]] = [ { "id": "chatcmpl-relay-smoke", "object": "chat.completion.chunk", @@ -119,18 +143,66 @@ class _ModelHandler(BaseHTTPRequestHandler): "index": 0, "delta": { "role": "assistant", - "content": RESPONSE_CANARY, + "content": "", }, "finish_reason": None, } ], - }, + } + ] + if request_tool: + chunks.append({ + "id": "chatcmpl-relay-smoke", + "object": "chat.completion.chunk", + "created": now, + "model": MODEL_CANARY, + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": 0, + "id": TOOL_CALL_CANARY, + "type": "function", + "function": { + "name": "read_file", + "arguments": json.dumps({"path": TOOL_FILE}), + }, + } + ] + }, + "finish_reason": None, + } + ], + }) + else: + chunks.append({ + "id": "chatcmpl-relay-smoke", + "object": "chat.completion.chunk", + "created": now, + "model": MODEL_CANARY, + "choices": [ + { + "index": 0, + "delta": {"content": RESPONSE_CANARY}, + "finish_reason": None, + } + ], + }) + chunks.extend([ { "id": "chatcmpl-relay-smoke", "object": "chat.completion.chunk", "created": now, "model": MODEL_CANARY, - "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "choices": [ + { + "index": 0, + "delta": {}, + "finish_reason": "tool_calls" if request_tool else "stop", + } + ], }, { "id": "chatcmpl-relay-smoke", @@ -144,7 +216,7 @@ class _ModelHandler(BaseHTTPRequestHandler): "total_tokens": 11, }, }, - ] + ]) self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") @@ -220,25 +292,29 @@ def _validate_store(database_path: Path) -> list[dict[str, Any]]: } for name, dimensions, value, packaged_value in rows ] - by_name = {counter["name"]: counter for counter in counters} + by_name: dict[str, list[dict[str, Any]]] = {} + for counter in counters: + by_name.setdefault(counter["name"], []).append(counter) if set(by_name) != { "hermes.model_route.count", "hermes.task_run.finished", "hermes.task_run.started", + "hermes.tool_call.count", }: raise AssertionError( f"Unexpected SQLite counters:\n{json.dumps(counters, indent=2)}" ) + [model] = by_name["hermes.model_route.count"] expected_model = { "name": "hermes.model_route.count", "dimensions": { "model": MODEL_CANARY, "provider": "custom", }, - "value": 1, - "packaged_value": 1, + "value": 2, + "packaged_value": 2, } - if by_name["hermes.model_route.count"] != expected_model: + if model != expected_model: raise AssertionError( f"Unexpected model counter: {by_name['hermes.model_route.count']}" ) @@ -251,21 +327,21 @@ def _validate_store(database_path: Path) -> list[dict[str, Any]]: "value": 1, "packaged_value": 1, } - if by_name["hermes.task_run.started"] != expected_start: + if by_name["hermes.task_run.started"] != [expected_start]: raise AssertionError( f"Unexpected task start: {by_name['hermes.task_run.started']}" ) - terminal = by_name["hermes.task_run.finished"] + [terminal] = by_name["hermes.task_run.finished"] expected_terminal_dimensions = { "duration_bucket": terminal["dimensions"].get("duration_bucket"), "end_reason": "completed", "entrypoint": "interactive", "execution_surface": "cli", - "model_call_count_bucket": "1", + "model_call_count_bucket": "2", "outcome": "success", "retry_count_bucket": "0", "termination": "none", - "tool_call_count_bucket": "0", + "tool_call_count_bucket": "1", } if ( terminal["dimensions"] != expected_terminal_dimensions @@ -273,6 +349,21 @@ def _validate_store(database_path: Path) -> list[dict[str, Any]]: or terminal["packaged_value"] != 1 ): raise AssertionError(f"Unexpected task terminal counter: {terminal}") + [tool] = by_name["hermes.tool_call.count"] + expected_tool_dimensions = { + "approval_outcome": "not_required", + "latency_bucket": tool["dimensions"].get("latency_bucket"), + "outcome": "success", + "retry_count_bucket": "unknown", + "tool_category": "file", + } + if ( + tool["dimensions"] != expected_tool_dimensions + or tool["dimensions"]["latency_bucket"] == "unknown" + or tool["value"] != 1 + or tool["packaged_value"] != 1 + ): + raise AssertionError(f"Unexpected tool counter: {tool}") return counters @@ -292,41 +383,62 @@ def _validate_package(outbox: Path, schema_path: Path) -> tuple[Path, dict[str, jsonschema.validate(package, schema) serialized = json.dumps(package) - for prohibited in (PROMPT_CANARY, RESPONSE_CANARY): + for prohibited in ( + PROMPT_CANARY, + RESPONSE_CANARY, + TOOL_CALL_CANARY, + TOOL_RESULT_CANARY, + ): if prohibited in serialized: raise AssertionError( f"Exported package leaked prohibited value: {prohibited!r}" ) - metrics = {metric["name"]: metric for metric in package.get("metrics", [])} + metrics: dict[str, list[dict[str, Any]]] = {} + for metric in package.get("metrics", []): + metrics.setdefault(metric["name"], []).append(metric) if set(metrics) != { "hermes.model_route.count", "hermes.task_run.finished", "hermes.task_run.started", + "hermes.tool_call.count", }: raise AssertionError( f"Unexpected package metrics:\n{json.dumps(package.get('metrics'), indent=2)}" ) - model_dimensions = metrics["hermes.model_route.count"]["dimensions"] - if model_dimensions != { + [model] = metrics["hermes.model_route.count"] + if model["dimensions"] != { "model": MODEL_CANARY, "provider": "custom", - }: + } or model["value"] != 2: raise AssertionError( f"Unexpected model metric: {metrics['hermes.model_route.count']}" ) - terminal = metrics["hermes.task_run.finished"] + [terminal] = metrics["hermes.task_run.finished"] if terminal["dimensions"] != { "duration_bucket": terminal["dimensions"].get("duration_bucket"), "end_reason": "completed", "entrypoint": "interactive", "execution_surface": "cli", - "model_call_count_bucket": "1", + "model_call_count_bucket": "2", "outcome": "success", "retry_count_bucket": "0", "termination": "none", - "tool_call_count_bucket": "0", + "tool_call_count_bucket": "1", }: raise AssertionError(f"Unexpected task terminal metric: {terminal}") + [tool] = metrics["hermes.tool_call.count"] + if ( + tool["dimensions"] + != { + "approval_outcome": "not_required", + "latency_bucket": tool["dimensions"].get("latency_bucket"), + "outcome": "success", + "retry_count_bucket": "unknown", + "tool_category": "file", + } + or tool["dimensions"]["latency_bucket"] == "unknown" + ): + raise AssertionError(f"Unexpected tool metric: {tool}") return package_path, package @@ -353,6 +465,7 @@ def main() -> int: home = root / "hermes-home" workdir = root / "workspace" workdir.mkdir() + (workdir / TOOL_FILE).write_text(TOOL_RESULT_CANARY, encoding="utf-8") home.mkdir() (home / ".no-bundled-skills").touch() @@ -382,7 +495,7 @@ def main() -> int: "--quiet", "--ignore-rules", "--toolsets", - "search", + "file", "--max-turns", "2", ], @@ -404,13 +517,18 @@ def main() -> int: f"Hermes exited with {result.returncode}\n" f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}" ) - if not _ModelHandler.requests: - raise AssertionError("Hermes did not call the local model endpoint") + if len(_ModelHandler.requests) != 2: + raise AssertionError( + f"Expected two model requests, got {len(_ModelHandler.requests)}" + ) request = _ModelHandler.requests[0] if request.get("model") != MODEL_CANARY: raise AssertionError(f"Unexpected model request: {request.get('model')!r}") if PROMPT_CANARY not in json.dumps(request.get("messages", [])): raise AssertionError("Hermes model request did not contain the prompt canary") + follow_up = json.dumps(_ModelHandler.requests[1].get("messages", [])) + if TOOL_CALL_CANARY not in follow_up or TOOL_RESULT_CANARY not in follow_up: + raise AssertionError("Hermes did not return the tool result to the model") if RESPONSE_CANARY not in result.stdout: raise AssertionError("Hermes did not print the mock model response") diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py index 4d604ef8995c8..d4ea437b82255 100644 --- a/tests/hermes_cli/test_plugins.py +++ b/tests/hermes_cli/test_plugins.py @@ -383,6 +383,54 @@ class TestPreToolCallBlocking: class TestPreToolCallDirective: """Tests for the extended (block | approve) directive helper.""" + def test_first_party_observer_receives_pre_tool_call(self, monkeypatch): + from hermes_cli import observability + from hermes_cli.plugins import get_pre_tool_call_directive + + observed = [] + monkeypatch.setattr( + observability, + "observe_lifecycle", + lambda hook_name, **kwargs: observed.append((hook_name, kwargs)), + ) + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [], + ) + + assert get_pre_tool_call_directive( + "write_file", + {"path": "README.md"}, + task_id="task-1", + session_id="session-1", + tool_call_id="call-1", + ) == (None, None) + assert observed == [ + ( + "pre_tool_call", + { + "tool_name": "write_file", + "args": {"path": "README.md"}, + "task_id": "task-1", + "session_id": "session-1", + "tool_call_id": "call-1", + "turn_id": "", + "api_request_id": "", + "middleware_trace": [], + }, + ) + ] + + def test_approve_directive_returned(self, monkeypatch): + from hermes_cli.plugins import get_pre_tool_call_directive + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [ + {"action": "approve", "message": "needs human ok"} + ], + ) + assert get_pre_tool_call_directive("write_file", {}) == ( + "approve", "needs human ok") def test_approve_without_message_is_valid(self, monkeypatch): """approve may omit a message (block may not).""" @@ -399,6 +447,33 @@ class TestResolvePreToolBlock: directive (incl. the approve→gate escalation) to a block message.""" + def test_approve_gate_receives_tool_observability_context(self, monkeypatch): + from hermes_cli.plugins import resolve_pre_tool_block + from tools import approval + + seen = {} + monkeypatch.setattr( + "hermes_cli.plugins.invoke_hook", + lambda hook_name, **kwargs: [ + {"action": "approve", "message": "why"} + ], + ) + + def _approve(*args, **kwargs): + seen["turn_id"] = approval._approval_turn_id.get() + seen["tool_call_id"] = approval._approval_tool_call_id.get() + return {"approved": True, "message": None} + + monkeypatch.setattr("tools.approval.request_tool_approval", _approve) + + assert resolve_pre_tool_block( + "write_file", + {}, + turn_id="turn-1", + tool_call_id="call-1", + ) is None + assert seen == {"turn_id": "turn-1", "tool_call_id": "call-1"} + def test_approve_passes_plugin_rule_key_to_gate(self, monkeypatch): from hermes_cli.plugins import resolve_pre_tool_block diff --git a/tests/hermes_cli/test_relay_shared_metrics.py b/tests/hermes_cli/test_relay_shared_metrics.py index 71abad092d48f..18190d9cb19ba 100644 --- a/tests/hermes_cli/test_relay_shared_metrics.py +++ b/tests/hermes_cli/test_relay_shared_metrics.py @@ -36,6 +36,12 @@ from hermes_cli.observability.shared_metrics_contract import ( TASK_ENTRYPOINTS, TASK_OUTCOMES, TASK_TERMINATIONS, + TOOL_APPROVAL_ATTRIBUTIONS, + TOOL_APPROVAL_OUTCOMES, + TOOL_CATEGORIES, + TOOL_LATENCY_BUCKETS, + TOOL_OUTCOMES, + TOOL_RETRY_BUCKETS, count_bucket, duration_bucket, execution_surface, @@ -45,6 +51,14 @@ from hermes_cli.observability.shared_metrics_contract import ( task_start_fields, task_terminal_fields, task_terminal_state, + tool_approval_counter, + tool_approval_outcome, + tool_call_dimensions, + tool_category, + tool_latency_bucket, + tool_outcome, + tool_retry_bucket, + tool_terminal_fields, ) @@ -78,6 +92,11 @@ def _task_dimension_schema(kind: str) -> dict[str, object]: return schema["$defs"][kind]["properties"]["dimensions"] +def _tool_dimension_schema(kind: str) -> dict[str, object]: + schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) + return schema["$defs"][kind]["properties"]["dimensions"] + + def _dimensions() -> dict[str, str]: return { "model": "anthropic/claude-sonnet-4.6", @@ -329,6 +348,129 @@ def test_v1_package_schema_retains_the_legacy_model_contract(): } +def test_package_schema_matches_the_tool_contract(): + schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) + tool = _tool_dimension_schema("tool_call_counter")["properties"] + approval = _tool_dimension_schema("tool_approval_counter")["properties"] + + assert set(tool["tool_category"]["enum"]) == TOOL_CATEGORIES + assert set(tool["outcome"]["enum"]) == TOOL_OUTCOMES + assert set(tool["approval_outcome"]["enum"]) == TOOL_APPROVAL_OUTCOMES + assert tool["latency_bucket"] == {"$ref": "#/$defs/tool_latency_bucket"} + assert tool["retry_count_bucket"] == {"$ref": "#/$defs/tool_retry_bucket"} + assert set(schema["$defs"]["tool_latency_bucket"]["enum"]) == ( + TOOL_LATENCY_BUCKETS + ) + assert set(schema["$defs"]["tool_retry_bucket"]["enum"]) == TOOL_RETRY_BUCKETS + assert set(approval["attribution"]["enum"]) == TOOL_APPROVAL_ATTRIBUTIONS + assert set(approval["outcome"]["enum"]) == ( + TOOL_APPROVAL_OUTCOMES - {"not_required"} + ) + + +@pytest.mark.parametrize( + ("toolset", "expected"), + [ + ("", "unknown"), + ("file", "file"), + ("terminal", "terminal"), + ("code_execution", "code_execution"), + ("delegation", "delegation"), + ("skills", "skill"), + ("browser-cdp", "browser"), + ("image_gen", "media"), + ("homeassistant", "home_automation"), + ("kanban", "planning"), + ("project", "project"), + ("discord", "communication"), + ("feishu_doc", "communication"), + ("mcp-github", "mcp"), + ("private_plugin", "other"), + ], +) +def test_tool_category_uses_bounded_runtime_toolsets(toolset, expected): + assert tool_category({"toolset": toolset}) == expected + + +def test_tool_category_does_not_classify_raw_tool_names(): + assert tool_category({"tool_name": "read_file"}) == "unknown" + + +@pytest.mark.parametrize( + ("status", "expected"), + [ + ("ok", "success"), + ("error", "failed"), + ("blocked", "blocked"), + ("cancelled", "cancelled"), + ("timeout", "timed_out"), + ("private", "unknown"), + (None, "unknown"), + ], +) +def test_tool_outcome_is_bounded(status, expected): + assert tool_outcome({"status": status}) == expected + + +@pytest.mark.parametrize( + ("choice", "expected"), + [ + ("once", "approved"), + ("session", "approved"), + ("always", "approved"), + ("smart_approve", "approved"), + ("deny", "denied"), + ("smart_deny", "denied"), + ("timeout", "timed_out"), + (None, "unknown"), + ], +) +def test_tool_approval_outcome_is_bounded(choice, expected): + assert tool_approval_outcome({"choice": choice}) == expected + + +@pytest.mark.parametrize( + ("duration_ms", "expected"), + [ + (0, "lt_100ms"), + (100, "100ms_to_250ms"), + (250, "250ms_to_500ms"), + (500, "500ms_to_1s"), + (1_000, "1s_to_2s"), + (2_000, "2s_to_5s"), + (5_000, "5s_to_10s"), + (10_000, "10s_to_30s"), + (30_000, "gte_30s"), + (-1, "unknown"), + (True, "unknown"), + ("100", "unknown"), + ], +) +def test_tool_latency_bucket_is_bounded(duration_ms, expected): + assert tool_latency_bucket(duration_ms) == expected + + +@pytest.mark.parametrize( + ("retry_count", "expected"), + [ + (0, "0"), + (1, "1"), + (2, "2"), + (3, "3_to_5"), + (6, "6_to_10"), + (11, "gte_11"), + (None, "unknown"), + (-1, "unknown"), + (True, "unknown"), + ], +) +def test_tool_retry_bucket_requires_an_explicit_non_negative_count( + retry_count, + expected, +): + assert tool_retry_bucket(retry_count) == expected + + def test_model_call_fields_report_terminal_model_and_provider_without_a_catalog(): assert model_call_fields({ "model": "fallback/model", @@ -425,6 +567,50 @@ def test_model_call_fields_collapse_malformed_identifiers(field, value): assert model_call_fields(event)[field] == "unknown" +def test_tool_subscriber_contract_accepts_only_bounded_events(): + terminal = SimpleNamespace( + kind="scope", + category="tool", + category_profile={}, + name="hermes.tool_call", + scope_category="end", + metadata={"hermes.metrics.schema_version": "hermes.metrics.event.v2"}, + data={ + "approval_outcome": "approved", + "latency_bucket": "250ms_to_500ms", + "outcome": "success", + "retry_count_bucket": "0", + "tool_category": "terminal", + }, + ) + assert tool_call_dimensions(terminal) == terminal.data + + terminal.data["result"] = "must-not-pass" + assert tool_call_dimensions(terminal) is None + terminal.data.pop("result") + terminal.data["tool_category"] = "private-tool-name" + assert tool_call_dimensions(terminal) is None + terminal.data["tool_category"] = "terminal" + terminal.category_profile["tool_name"] = "must-not-pass" + assert tool_call_dimensions(terminal) is None + + approval = SimpleNamespace( + kind="mark", + category=None, + category_profile=None, + name="hermes.tool_approval", + scope_category=None, + metadata={"hermes.metrics.schema_version": "hermes.metrics.event.v2"}, + data={"attribution": "unattributed", "outcome": "denied"}, + ) + assert tool_approval_counter(approval) == ( + "hermes.tool_approval.count", + approval.data, + ) + approval.data["command"] = "must-not-pass" + assert tool_approval_counter(approval) is None + + def test_store_does_not_record_the_retired_model_metric(tmp_path): store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox") diff --git a/tests/hermes_cli/test_relay_shared_metrics_runtime.py b/tests/hermes_cli/test_relay_shared_metrics_runtime.py index ec5f4e3c7f436..3dabef54493ec 100644 --- a/tests/hermes_cli/test_relay_shared_metrics_runtime.py +++ b/tests/hermes_cli/test_relay_shared_metrics_runtime.py @@ -30,11 +30,14 @@ class _Relay: self.events: list[tuple[Any, ...]] = [] self._callbacks: dict[str, Any] = {} self._starts: dict[Any, dict[str, Any]] = {} + self._tool_starts: dict[Any, dict[str, Any]] = {} self._scope_starts: dict[Any, dict[str, Any]] = {} self._scope = contextvars.ContextVar("relay_scope", default=None) self._scope_stack = contextvars.ContextVar("relay_scope_stack", default=None) self._scope_serial = 0 - self.ScopeType = SimpleNamespace(Agent="agent", Function="function") + self.ScopeType = SimpleNamespace( + Agent="agent", Function="function", Tool="tool" + ) self.LLMRequest = _Request self.scope = SimpleNamespace( push=self._scope_push, @@ -42,6 +45,7 @@ class _Relay: event=self._scope_event, ) self.llm = SimpleNamespace(call=self._llm_call, call_end=self._llm_call_end) + self.tools = SimpleNamespace(call=self._tool_call, call_end=self._tool_call_end) self.subscribers = SimpleNamespace( register=self._register, deregister=self._deregister, @@ -102,6 +106,17 @@ class _Relay: def _scope_event(self, name: str, **kwargs: Any) -> None: self.events.append(("scope.event", name, kwargs)) + event = SimpleNamespace( + kind="mark", + category=None, + name=name, + scope_category=None, + category_profile=None, + metadata=kwargs.get("metadata"), + data=kwargs.get("data"), + ) + for callback in list(self._callbacks.values()): + callback(event) def _get_scope_stack(self) -> Any: stack = self._scope_stack.get() @@ -145,6 +160,41 @@ class _Relay: for callback in list(self._callbacks.values()): callback(event) + def _tool_call( + self, + name: str, + args: dict[str, Any], + **kwargs: Any, + ) -> Any: + handle = ("tool", name, len(self._tool_starts)) + self._tool_starts[handle] = kwargs + self.events.append(("tool.call", name, args, kwargs)) + return handle + + def _tool_call_end( + self, + handle: Any, + result: dict[str, Any], + **kwargs: Any, + ) -> None: + start = self._tool_starts.pop(handle) + self.events.append(("tool.call_end", handle, result, kwargs)) + event = SimpleNamespace( + kind="scope", + category="tool", + name=handle[1], + scope_category="end", + category_profile={}, + metadata={ + **start["metadata"], + **kwargs["metadata"], + "otel.status_code": "OK", + }, + data=result, + ) + for callback in list(self._callbacks.values()): + callback(event) + def _register(self, name: str, callback: Any) -> None: self._callbacks[name] = callback self.events.append(("subscribers.register", name)) @@ -196,6 +246,7 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa base = { "session_id": "sensitive-session", "task_id": "task-1", + "turn_id": "turn-1", "api_request_id": "request-1", "platform": "cli", "provider": "custom", @@ -211,14 +262,33 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa **base, request={"body": {"messages": ["sensitive-prompt"]}}, ) + lifecycle.invoke_hook( + "pre_tool_call", + **base, + tool_call_id="sensitive-tool-call", + tool_name="terminal", + toolset="terminal", + args={"command": "sensitive-command"}, + ) + lifecycle.invoke_hook( + "post_approval_response", + turn_id=base["turn_id"], + tool_call_id="sensitive-tool-call", + choice="once", + command="sensitive-command", + description="sensitive-approval-description", + ) lifecycle.invoke_hook( "post_tool_call", **base, tool_call_id="sensitive-tool-call", tool_name="terminal", + toolset="terminal", args={"command": "sensitive-command"}, result={"output": "sensitive-tool-result"}, status="ok", + duration_ms=275, + retry_count=0, ) lifecycle.invoke_hook( "api_request_error", @@ -258,6 +328,10 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa starts = [event for event in direct_runtime.events if event[0] == "llm.call"] ends = [event for event in direct_runtime.events if event[0] == "llm.call_end"] + tool_starts = [event for event in direct_runtime.events if event[0] == "tool.call"] + tool_ends = [ + event for event in direct_runtime.events if event[0] == "tool.call_end" + ] scope_starts = [ event for event in direct_runtime.events if event[0] == "scope.push" ] @@ -272,6 +346,17 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa } assert len(starts) == 1 assert len(ends) == 1 + assert len(tool_starts) == 1 + assert len(tool_ends) == 1 + assert tool_starts[0][1] == "hermes.tool_call" + assert tool_starts[0][2] == {} + assert tool_ends[0][2] == { + "approval_outcome": "approved", + "latency_bucket": "250ms_to_500ms", + "outcome": "success", + "retry_count_bucket": "0", + "tool_category": "terminal", + } assert starts[0][2] == {} assert starts[0][3]["model_name"] == "unknown" assert ends[0][2] == { @@ -285,6 +370,7 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa assert "sensitive-command" not in serialized_events assert "sensitive-tool-result" not in serialized_events assert "sensitive-tool-call" not in serialized_events + assert "sensitive-approval-description" not in serialized_events assert "gpt-sensitive-model-id" not in serialized_events assert plugins.get_plugin_manager().list_plugins() == [] @@ -297,12 +383,35 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa "hermes.model_route.count", "hermes.task_run.finished", "hermes.task_run.started", + "hermes.tool_approval.count", + "hermes.tool_call.count", } assert metrics["hermes.model_route.count"]["dimensions"] == { "model": "claude-sonnet", "provider": "anthropic", } assert metrics["hermes.model_route.count"]["value"] == 1 + assert metrics["hermes.tool_call.count"] == { + "name": "hermes.tool_call.count", + "type": "counter", + "dimensions": { + "approval_outcome": "approved", + "latency_bucket": "250ms_to_500ms", + "outcome": "success", + "retry_count_bucket": "0", + "tool_category": "terminal", + }, + "value": 1, + } + assert metrics["hermes.tool_approval.count"] == { + "name": "hermes.tool_approval.count", + "type": "counter", + "dimensions": { + "attribution": "tool_call", + "outcome": "approved", + }, + "value": 1, + } assert metrics["hermes.task_run.started"] == { "name": "hermes.task_run.started", "type": "counter", @@ -370,6 +479,21 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot( error={"message": prompt_canary}, ) lifecycle.invoke_hook("pre_api_request", **success, retry_count=1) + lifecycle.invoke_hook( + "pre_tool_call", + **success, + tool_call_id="sensitive-tool-call", + tool_name="terminal", + args={"command": prompt_canary}, + ) + lifecycle.invoke_hook( + "post_approval_response", + turn_id=success["turn_id"], + tool_call_id="sensitive-tool-call", + choice="session", + command=prompt_canary, + description="sensitive-approval-description", + ) lifecycle.invoke_hook( "post_tool_call", **success, @@ -378,6 +502,8 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot( args={"command": prompt_canary}, result={"output": tool_canary}, status="ok", + duration_ms=125, + retry_count=0, ) lifecycle.invoke_hook( "post_api_request", @@ -399,6 +525,23 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot( lifecycle.invoke_hook("on_session_start", **failed) lifecycle.invoke_hook("pre_llm_call", **failed, messages=[prompt_canary]) lifecycle.invoke_hook("pre_api_request", **failed, retry_count=0) + lifecycle.invoke_hook( + "pre_tool_call", + **failed, + tool_call_id="sensitive-failed-tool-call", + tool_name="read_file", + args={"path": prompt_canary}, + ) + lifecycle.invoke_hook( + "post_tool_call", + **failed, + tool_call_id="sensitive-failed-tool-call", + tool_name="read_file", + args={"path": prompt_canary}, + result={"error": tool_canary}, + status="error", + duration_ms=750, + ) lifecycle.invoke_hook( "api_request_error", **failed, @@ -420,6 +563,23 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot( lifecycle.invoke_hook("on_session_start", **cancelled) lifecycle.invoke_hook("pre_llm_call", **cancelled, messages=[prompt_canary]) lifecycle.invoke_hook("pre_api_request", **cancelled, retry_count=0) + lifecycle.invoke_hook( + "pre_tool_call", + **cancelled, + tool_call_id="sensitive-cancelled-tool-call", + tool_name="browser_navigate", + args={"url": prompt_canary}, + ) + lifecycle.invoke_hook( + "post_tool_call", + **cancelled, + tool_call_id="sensitive-cancelled-tool-call", + tool_name="browser_navigate", + args={"url": prompt_canary}, + result={"error": tool_canary}, + status="cancelled", + duration_ms=31_000, + ) lifecycle.invoke_hook( "on_session_end", **cancelled, @@ -454,6 +614,43 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot( "provider": "custom", } assert model_counter["value"] == 3 + assert { + counter["dimensions"]["outcome"] + for counter in by_metric["hermes.tool_call.count"] + } == {"success", "failed", "cancelled"} + tool_by_outcome = { + counter["dimensions"]["outcome"]: counter["dimensions"] + for counter in by_metric["hermes.tool_call.count"] + } + assert tool_by_outcome["success"] == { + "approval_outcome": "approved", + "latency_bucket": "100ms_to_250ms", + "outcome": "success", + "retry_count_bucket": "0", + "tool_category": "terminal", + } + assert tool_by_outcome["failed"] == { + "approval_outcome": "not_required", + "latency_bucket": "500ms_to_1s", + "outcome": "failed", + "retry_count_bucket": "unknown", + "tool_category": "file", + } + assert tool_by_outcome["cancelled"] == { + "approval_outcome": "not_required", + "latency_bucket": "gte_30s", + "outcome": "cancelled", + "retry_count_bucket": "unknown", + "tool_category": "browser", + } + assert len(by_metric["hermes.tool_approval.count"]) == 1 + approval_counter = by_metric["hermes.tool_approval.count"][0] + assert approval_counter["dimensions"] == { + "attribution": "tool_call", + "outcome": "approved", + } + assert approval_counter["value"] == 1 + assert approval_counter["packaged_value"] == 1 terminal_by_outcome = { counter["dimensions"]["outcome"]: counter for counter in by_metric["hermes.task_run.finished"] @@ -502,10 +699,183 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot( "sensitive-task", "sensitive-request", "sensitive-tool-call", + "sensitive-failed-tool-call", + "sensitive-cancelled-tool-call", + "sensitive-approval-description", ): assert canary not in serialized_analytics +def test_real_binding_correlates_plugin_approval_denial_to_tool_metric( + real_binding_runtime, + tmp_path, + monkeypatch, +): + from hermes_cli.observability.shared_metrics import SharedMetricsStore + from tools import approval + + assert real_binding_runtime._native is not None + base = { + "session_id": "sensitive-session", + "task_id": "sensitive-task", + "turn_id": "sensitive-turn", + "api_request_id": "sensitive-request", + "platform": "cli", + } + + def plugin_hook(hook_name: str, **kwargs: Any) -> list[dict[str, str]]: + if hook_name == "pre_tool_call": + return [{"action": "approve", "message": "sensitive-rule"}] + return [] + + monkeypatch.setattr(plugins, "invoke_hook", plugin_hook) + monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", False) + monkeypatch.setattr(approval, "is_current_session_yolo_enabled", lambda: False) + monkeypatch.setattr(approval, "is_approved", lambda *args: False) + monkeypatch.setattr(approval, "get_current_session_key", lambda: "session-key") + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True) + monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) + monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *args, **kwargs: "deny") + + lifecycle.invoke_hook("on_session_start", **base) + lifecycle.invoke_hook("pre_llm_call", **base, messages=["sensitive-prompt"]) + block_message = plugins.resolve_pre_tool_block( + "write_file", + {"path": "sensitive-path"}, + task_id=base["task_id"], + session_id=base["session_id"], + turn_id=base["turn_id"], + api_request_id=base["api_request_id"], + tool_call_id="sensitive-tool-call", + ) + assert block_message is not None + assert "User denied" in block_message + + lifecycle.invoke_hook( + "post_tool_call", + **base, + tool_call_id="sensitive-tool-call", + tool_name="write_file", + args={"path": "sensitive-path"}, + result={"error": block_message}, + status="blocked", + duration_ms=12, + ) + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=False, + failed=True, + interrupted=False, + turn_exit_reason="approval_denied", + ) + lifecycle.finalize_session(session_id=base["session_id"]) + + root = tmp_path / "hermes-home" / "telemetry" / "shared_metrics" + store = SharedMetricsStore(root / "metrics.sqlite3", root / "outbox") + snapshot = store.counter_snapshot() + tool_metrics = [ + counter + for counter in snapshot + if counter["metric_name"] == "hermes.tool_call.count" + ] + assert len(tool_metrics) == 1 + assert tool_metrics[0]["dimensions"] == { + "approval_outcome": "denied", + "latency_bucket": "lt_100ms", + "outcome": "blocked", + "retry_count_bucket": "unknown", + "tool_category": "file", + } + approval_metrics = [ + counter + for counter in snapshot + if counter["metric_name"] == "hermes.tool_approval.count" + ] + assert len(approval_metrics) == 1 + assert approval_metrics[0]["dimensions"] == { + "attribution": "tool_call", + "outcome": "denied", + } + assert "sensitive" not in json.dumps(snapshot) + + +def test_real_binding_aggregates_tool_and_approval_timeouts( + real_binding_runtime, + tmp_path, +): + from hermes_cli.observability.shared_metrics import SharedMetricsStore + + assert real_binding_runtime._native is not None + base = { + "session_id": "timeout-sensitive-session", + "task_id": "timeout-sensitive-task", + "turn_id": "timeout-sensitive-turn", + "platform": "cli", + } + + lifecycle.invoke_hook("on_session_start", **base) + lifecycle.invoke_hook("pre_llm_call", **base, messages=["timeout-sensitive-prompt"]) + lifecycle.invoke_hook( + "pre_tool_call", + **base, + tool_call_id="timeout-sensitive-tool-call", + tool_name="terminal", + args={"command": "timeout-sensitive-command"}, + ) + lifecycle.invoke_hook( + "post_approval_response", + **base, + tool_call_id="timeout-sensitive-tool-call", + choice="timeout", + command="timeout-sensitive-command", + ) + lifecycle.invoke_hook( + "post_tool_call", + **base, + tool_call_id="timeout-sensitive-tool-call", + tool_name="terminal", + result={"error": "timeout-sensitive-result"}, + status="timeout", + duration_ms=30_000, + ) + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=False, + failed=True, + interrupted=False, + turn_exit_reason="provider_timeout", + ) + lifecycle.finalize_session(session_id=base["session_id"]) + + root = tmp_path / "hermes-home" / "telemetry" / "shared_metrics" + snapshot = SharedMetricsStore( + root / "metrics.sqlite3", + root / "outbox", + ).counter_snapshot() + [tool_metric] = [ + counter + for counter in snapshot + if counter["metric_name"] == "hermes.tool_call.count" + ] + assert tool_metric["dimensions"] == { + "approval_outcome": "timed_out", + "latency_bucket": "gte_30s", + "outcome": "timed_out", + "retry_count_bucket": "unknown", + "tool_category": "terminal", + } + [approval_metric] = [ + counter + for counter in snapshot + if counter["metric_name"] == "hermes.tool_approval.count" + ] + assert approval_metric["dimensions"] == { + "attribution": "tool_call", + "outcome": "timed_out", + } + assert "timeout-sensitive" not in json.dumps(snapshot) @@ -810,24 +1180,29 @@ def test_disabling_shared_metrics_stops_collection_and_shutdown_export( assert not relay_shared_metrics.enabled() counters_before_stale_event = runtime.subscriber.store.counter_snapshot() - runtime.subscriber(SimpleNamespace( - kind="scope", - category="function", - category_profile=None, - name="hermes.task_run", - scope_category="start", - metadata={ - "hermes.metrics.schema_version": "hermes.metrics.event.v1", - relay_runtime.RUNTIME_INSTANCE_KEY: runtime.host.runtime_id, - }, - data={"entrypoint": "interactive", "execution_surface": "cli"}, - )) + runtime.subscriber( + SimpleNamespace( + kind="scope", + category="function", + category_profile=None, + name="hermes.task_run", + scope_category="start", + metadata={ + "hermes.metrics.schema_version": "hermes.metrics.event.v1", + relay_runtime.RUNTIME_INSTANCE_KEY: runtime.host.runtime_id, + }, + data={"entrypoint": "interactive", "execution_surface": "cli"}, + ) + ) assert runtime.subscriber.store.counter_snapshot() == counters_before_stale_event - assert runtime.start_task({ - "session_id": "session", - "task_id": "stale-runtime-task", - "platform": "cli", - }) is None + assert ( + runtime.start_task({ + "session_id": "session", + "task_id": "stale-runtime-task", + "platform": "cli", + }) + is None + ) relay_shared_metrics.finish_task_run( session_id="session", task_id="task", @@ -1092,6 +1467,7 @@ def test_subagent_agent_boundary_closes_its_own_scope( ) AIAgent.run_conversation(child_agent, "private", task_id="child-task") elif terminal == "exception": + def fail(*_args, **_kwargs): raise RuntimeError("child failed") @@ -1099,6 +1475,7 @@ def test_subagent_agent_boundary_closes_its_own_scope( with pytest.raises(RuntimeError, match="child failed"): AIAgent.run_conversation(child_agent, "private", task_id="child-task") elif terminal == "cancelled": + def cancel(*_args, **_kwargs): raise KeyboardInterrupt @@ -1106,6 +1483,7 @@ def test_subagent_agent_boundary_closes_its_own_scope( with pytest.raises(KeyboardInterrupt): AIAgent.run_conversation(child_agent, "private", task_id="child-task") else: + def time_out(*_args, **_kwargs): raise TimeoutError("child timed out") @@ -1278,6 +1656,525 @@ def test_same_request_id_is_isolated_between_tasks(direct_runtime): assert all(fields["retry_count_bucket"] == "0" for fields in task_ends) +def test_reused_tool_call_id_is_counted_for_each_provider_request(direct_runtime): + base = { + "session_id": "s1", + "task_id": "t1", + "turn_id": "turn-1", + "platform": "cli", + } + lifecycle.invoke_hook("pre_llm_call", **base) + + for api_request_id in ("request-1", "request-2"): + call = { + **base, + "api_request_id": api_request_id, + "tool_call_id": "provider-reused-id", + "tool_name": "terminal", + } + lifecycle.invoke_hook("pre_tool_call", **call, args={"command": "private"}) + lifecycle.invoke_hook( + "post_tool_call", + **call, + result={"output": "private"}, + status="ok", + ) + + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=True, + failed=False, + interrupted=False, + turn_exit_reason="text_response(stop)", + ) + lifecycle.finalize_session(session_id="s1") + + tool_ends = [ + event for event in direct_runtime.events if event[0] == "tool.call_end" + ] + assert len(tool_ends) == 2 + [task_end] = [ + event + for event in direct_runtime.events + if event[0] == "scope.pop" and event[1][1] == "hermes.task_run" + ] + assert task_end[2]["output"]["tool_call_count_bucket"] == "2" + + +def test_partial_terminal_context_reuses_the_pending_tool_span(direct_runtime): + base = { + "session_id": "s1", + "task_id": "t1", + "turn_id": "turn-1", + "api_request_id": "request-1", + "platform": "cli", + "tool_call_id": "tool-1", + "tool_name": "terminal", + } + lifecycle.invoke_hook("pre_llm_call", **base) + lifecycle.invoke_hook("pre_tool_call", **base) + lifecycle.invoke_hook( + "post_tool_call", + **{key: value for key, value in base.items() if key != "api_request_id"}, + result={"output": "private"}, + status="ok", + ) + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=True, + failed=False, + interrupted=False, + turn_exit_reason="text_response(stop)", + ) + lifecycle.finalize_session(session_id="s1") + + [tool_end] = [ + event for event in direct_runtime.events if event[0] == "tool.call_end" + ] + assert tool_end[2]["outcome"] == "success" + [task_end] = [ + event + for event in direct_runtime.events + if event[0] == "scope.pop" and event[1][1] == "hermes.task_run" + ] + assert task_end[2]["output"]["tool_call_count_bucket"] == "1" + + +def test_partial_terminal_variants_do_not_double_count_a_completed_call( + direct_runtime, +): + base = { + "session_id": "s1", + "task_id": "t1", + "turn_id": "turn-1", + "api_request_id": "request-1", + "platform": "cli", + "tool_call_id": "tool-1", + "tool_name": "terminal", + } + lifecycle.invoke_hook("pre_llm_call", **base) + lifecycle.invoke_hook("pre_tool_call", **base) + for omitted_field in ("api_request_id", "turn_id"): + lifecycle.invoke_hook( + "post_tool_call", + **{key: value for key, value in base.items() if key != omitted_field}, + result={"output": "private"}, + status="ok", + ) + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=True, + failed=False, + interrupted=False, + turn_exit_reason="text_response(stop)", + ) + lifecycle.finalize_session(session_id="s1") + + tool_ends = [ + event for event in direct_runtime.events if event[0] == "tool.call_end" + ] + assert len(tool_ends) == 1 + [task_end] = [ + event + for event in direct_runtime.events + if event[0] == "scope.pop" and event[1][1] == "hermes.task_run" + ] + assert task_end[2]["output"]["tool_call_count_bucket"] == "1" + + +def test_ambiguous_partial_terminal_does_not_create_a_phantom_tool_span( + direct_runtime, +): + base = { + "session_id": "s1", + "task_id": "t1", + "turn_id": "turn-1", + "platform": "cli", + "tool_call_id": "provider-reused-id", + "tool_name": "terminal", + } + lifecycle.invoke_hook("pre_llm_call", **base) + for api_request_id in ("request-1", "request-2"): + lifecycle.invoke_hook( + "pre_tool_call", + **base, + api_request_id=api_request_id, + ) + + lifecycle.invoke_hook( + "post_tool_call", + **base, + result={"output": "ambiguous-private-result"}, + status="ok", + ) + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=False, + failed=True, + interrupted=False, + turn_exit_reason="system_aborted", + ) + lifecycle.finalize_session(session_id="s1") + + tool_ends = [ + event for event in direct_runtime.events if event[0] == "tool.call_end" + ] + assert len(tool_ends) == 2 + assert all(event[2]["outcome"] == "failed" for event in tool_ends) + [task_end] = [ + event + for event in direct_runtime.events + if event[0] == "scope.pop" and event[1][1] == "hermes.task_run" + ] + assert task_end[2]["output"]["tool_call_count_bucket"] == "2" + + +def test_reused_task_id_starts_a_new_run_for_each_turn(direct_runtime): + for turn_id in ("turn-1", "turn-2"): + base = { + "session_id": "reused-session", + "task_id": "reused-session", + "turn_id": turn_id, + "platform": "api", + } + lifecycle.invoke_hook("pre_llm_call", **base) + lifecycle.invoke_hook( + "post_tool_call", + **base, + api_request_id=f"request-{turn_id}", + tool_call_id=f"tool-{turn_id}", + tool_name="read_file", + result={"output": "private"}, + status="ok", + ) + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=True, + failed=False, + interrupted=False, + turn_exit_reason="text_response(stop)", + ) + + lifecycle.finalize_session(session_id="reused-session") + + task_starts = [ + event + for event in direct_runtime.events + if event[0] == "scope.push" and event[1] == "hermes.task_run" + ] + task_ends = [ + event + for event in direct_runtime.events + if event[0] == "scope.pop" and event[1][1] == "hermes.task_run" + ] + tool_ends = [ + event for event in direct_runtime.events if event[0] == "tool.call_end" + ] + assert len(task_starts) == 2 + assert len(task_ends) == 2 + assert len(tool_ends) == 2 + assert all( + event[2]["output"]["tool_call_count_bucket"] == "1" for event in task_ends + ) + + +def test_late_tool_result_does_not_attach_to_reused_task_id(direct_runtime): + first = { + "session_id": "reused-session", + "task_id": "reused-session", + "turn_id": "turn-1", + "platform": "api", + } + lifecycle.invoke_hook("pre_llm_call", **first) + lifecycle.invoke_hook( + "pre_tool_call", + **first, + api_request_id="request-1", + tool_call_id="tool-1", + tool_name="terminal", + ) + lifecycle.invoke_hook( + "on_session_end", + **first, + completed=False, + failed=True, + interrupted=False, + turn_exit_reason="timed_out", + ) + + second = {**first, "turn_id": "turn-2"} + runtime = relay_shared_metrics._get_runtime() + assert runtime is not None + assert runtime.start_task({ + "session_id": second["session_id"], + "task_id": second["task_id"], + "platform": second["platform"], + }) + lifecycle.invoke_hook( + "post_tool_call", + **first, + api_request_id="request-1", + tool_call_id="tool-1", + tool_name="terminal", + result={"output": "late-private-result"}, + status="ok", + ) + lifecycle.invoke_hook("pre_llm_call", **second) + lifecycle.invoke_hook( + "post_tool_call", + **second, + api_request_id="request-2", + tool_call_id="tool-2", + tool_name="read_file", + result={"output": "current-private-result"}, + status="ok", + ) + lifecycle.invoke_hook( + "on_session_end", + **second, + completed=True, + failed=False, + interrupted=False, + turn_exit_reason="text_response(stop)", + ) + lifecycle.finalize_session(session_id="reused-session") + + tool_ends = [ + event for event in direct_runtime.events if event[0] == "tool.call_end" + ] + assert len(tool_ends) == 2 + assert [event[2]["outcome"] for event in tool_ends] == [ + "timed_out", + "success", + ] + assert [event[2]["tool_category"] for event in tool_ends] == [ + "terminal", + "file", + ] + task_ends = [ + event + for event in direct_runtime.events + if event[0] == "scope.pop" and event[1][1] == "hermes.task_run" + ] + assert [ + event[2]["output"]["tool_call_count_bucket"] for event in task_ends + ] == ["1", "1"] + + +def test_pending_tool_is_closed_and_counted_when_task_is_interrupted(direct_runtime): + base = { + "session_id": "s1", + "task_id": "t1", + "turn_id": "turn-1", + "platform": "cli", + } + + lifecycle.invoke_hook("on_session_start", **base) + lifecycle.invoke_hook( + "pre_tool_call", + **base, + tool_call_id="tool-1", + tool_name="terminal", + args={"command": "must-not-pass"}, + ) + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=False, + failed=False, + interrupted=True, + turn_exit_reason="interrupted_by_user", + ) + lifecycle.invoke_hook( + "post_tool_call", + **base, + tool_call_id="tool-1", + tool_name="terminal", + result={"output": "late-result-must-not-pass"}, + status="ok", + ) + lifecycle.finalize_session(session_id="s1") + + [tool_end] = [ + event for event in direct_runtime.events if event[0] == "tool.call_end" + ] + assert tool_end[2] == { + "approval_outcome": "not_required", + "latency_bucket": tool_end[2]["latency_bucket"], + "outcome": "cancelled", + "retry_count_bucket": "unknown", + "tool_category": "terminal", + } + [task_end] = [ + event + for event in direct_runtime.events + if event[0] == "scope.pop" and event[1][1] == "hermes.task_run" + ] + assert task_end[2]["output"]["tool_call_count_bucket"] == "1" + task_starts = [ + event + for event in direct_runtime.events + if event[0] == "scope.push" and event[1] == "hermes.task_run" + ] + assert len(task_starts) == 1 + + +def test_pending_tool_uses_the_outer_task_timeout_outcome(direct_runtime): + base = { + "session_id": "s1", + "task_id": "t1", + "turn_id": "turn-1", + "platform": "api", + } + lifecycle.invoke_hook("pre_llm_call", **base) + lifecycle.invoke_hook( + "pre_tool_call", + **base, + tool_call_id="tool-1", + tool_name="web_search", + ) + + relay_shared_metrics.finish_task_run( + session_id="s1", + task_id="t1", + platform="api", + error=TimeoutError("private timeout detail"), + ) + lifecycle.finalize_session(session_id="s1") + + [tool_end] = [ + event for event in direct_runtime.events if event[0] == "tool.call_end" + ] + assert tool_end[2]["outcome"] == "timed_out" + [task_end] = [ + event + for event in direct_runtime.events + if event[0] == "scope.pop" and event[1][1] == "hermes.task_run" + ] + assert task_end[2]["output"]["outcome"] == "timed_out" + assert task_end[2]["output"]["tool_call_count_bucket"] == "1" + assert "private timeout detail" not in repr(direct_runtime.events) + + +def test_approval_without_tool_context_is_counted_as_unattributed(direct_runtime): + base = { + "session_id": "s1", + "task_id": "t1", + "turn_id": "turn-1", + "platform": "cli", + } + + lifecycle.invoke_hook("on_session_start", **base) + lifecycle.invoke_hook("pre_llm_call", **base) + lifecycle.invoke_hook( + "post_approval_response", + turn_id="turn-1", + choice="deny", + command="must-not-pass", + ) + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=False, + failed=True, + interrupted=False, + turn_exit_reason="approval_denied", + ) + lifecycle.finalize_session(session_id="s1") + + [approval] = [ + event + for event in direct_runtime.events + if event[0] == "scope.event" and event[1] == "hermes.tool_approval" + ] + assert approval[2]["data"] == { + "attribution": "unattributed", + "outcome": "denied", + } + + +def test_approval_with_unmatched_tool_id_is_counted_as_unattributed(direct_runtime): + base = { + "session_id": "s1", + "task_id": "t1", + "turn_id": "turn-1", + "platform": "cli", + } + + lifecycle.invoke_hook("pre_llm_call", **base) + lifecycle.invoke_hook( + "post_approval_response", + **base, + tool_call_id="spoofed-tool-call", + choice="deny", + command="must-not-pass", + ) + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=False, + failed=True, + interrupted=False, + turn_exit_reason="approval_denied", + ) + lifecycle.finalize_session(session_id="s1") + + [approval] = [ + event + for event in direct_runtime.events + if event[0] == "scope.event" and event[1] == "hermes.tool_approval" + ] + assert approval[2]["data"] == { + "attribution": "unattributed", + "outcome": "denied", + } + + +def test_tool_category_comes_from_runtime_registry_metadata( + direct_runtime, + monkeypatch, +): + import model_tools + + monkeypatch.setattr( + model_tools, + "get_toolset_for_tool", + lambda name: "terminal" if name == "runtime_only_tool" else None, + ) + base = { + "session_id": "s1", + "task_id": "t1", + "turn_id": "turn-1", + "platform": "cli", + } + lifecycle.invoke_hook("pre_llm_call", **base) + lifecycle.invoke_hook( + "post_tool_call", + **base, + tool_call_id="tool-1", + tool_name="runtime_only_tool", + result={"output": "private"}, + status="ok", + ) + lifecycle.invoke_hook( + "on_session_end", + **base, + completed=True, + failed=False, + interrupted=False, + turn_exit_reason="text_response(stop)", + ) + lifecycle.finalize_session(session_id="s1") + + [tool_end] = [ + event for event in direct_runtime.events if event[0] == "tool.call_end" + ] + assert tool_end[2]["tool_category"] == "terminal" def test_task_retry_count_survives_provider_fallback_ordinal_reset(direct_runtime): base = { "session_id": "s1", diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index e26728214dc10..721dafbeae4f4 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -1580,11 +1580,18 @@ class TestExecuteToolCalls: assert post_calls[0]["error_type"] == "keyboard_interrupt" assert json.loads(post_calls[0]["result"])["status"] == "cancelled" - def test_interrupt_skips_remaining(self, agent): + def test_interrupt_skips_remaining(self, agent, monkeypatch): tc1 = _mock_tool_call(name="web_search", arguments="{}", call_id="c1") tc2 = _mock_tool_call(name="web_search", arguments="{}", call_id="c2") mock_msg = _mock_assistant_msg(content="", tool_calls=[tc1, tc2]) messages = [] + hook_calls = [] + + monkeypatch.setattr("hermes_cli.lifecycle.has_hook", lambda name: True) + monkeypatch.setattr( + "hermes_cli.lifecycle.invoke_hook", + lambda hook_name, **kwargs: hook_calls.append((hook_name, kwargs)) or [], + ) with patch("run_agent._set_interrupt"): agent.interrupt() @@ -1596,13 +1603,22 @@ class TestExecuteToolCalls: "cancelled" in messages[0]["content"].lower() or "interrupted" in messages[0]["content"].lower() ) + post_calls = [kwargs for name, kwargs in hook_calls if name == "post_tool_call"] + assert [call["tool_call_id"] for call in post_calls] == ["c1", "c2"] + assert all(call["status"] == "cancelled" for call in post_calls) - def test_invalid_json_args_are_rejected_without_dispatch(self, agent): + def test_invalid_json_args_are_rejected_without_dispatch(self, agent, monkeypatch): tc = _mock_tool_call( name="web_search", arguments="not valid json", call_id="c1" ) mock_msg = _mock_assistant_msg(content="", tool_calls=[tc]) messages = [] + hook_calls = [] + monkeypatch.setattr("hermes_cli.lifecycle.has_hook", lambda name: True) + monkeypatch.setattr( + "hermes_cli.lifecycle.invoke_hook", + lambda hook_name, **kwargs: hook_calls.append((hook_name, kwargs)) or [], + ) with patch("run_agent.handle_function_call", return_value="ok") as mock_hfc: agent._execute_tool_calls(mock_msg, messages, "task-1") mock_hfc.assert_not_called() @@ -1611,6 +1627,34 @@ class TestExecuteToolCalls: assert messages[0]["tool_call_id"] == "c1" assert "valid json object" in messages[0]["content"].lower() assert "tool was not executed" in messages[0]["content"].lower() + [post_call] = [ + kwargs for name, kwargs in hook_calls if name == "post_tool_call" + ] + assert post_call["tool_call_id"] == "c1" + assert post_call["status"] == "error" + assert post_call["error_type"] == "invalid_tool_arguments" + + def test_concurrent_invalid_json_args_emit_terminal_hook(self, agent, monkeypatch): + tc = _mock_tool_call( + name="web_search", arguments="not valid json", call_id="c1" + ) + mock_msg = _mock_assistant_msg(content="", tool_calls=[tc]) + messages = [] + hook_calls = [] + monkeypatch.setattr("hermes_cli.lifecycle.has_hook", lambda name: True) + monkeypatch.setattr( + "hermes_cli.lifecycle.invoke_hook", + lambda hook_name, **kwargs: hook_calls.append((hook_name, kwargs)) or [], + ) + + agent._execute_tool_calls_concurrent(mock_msg, messages, "task-1") + + [post_call] = [ + kwargs for name, kwargs in hook_calls if name == "post_tool_call" + ] + assert post_call["tool_call_id"] == "c1" + assert post_call["status"] == "error" + assert post_call["error_type"] == "invalid_tool_arguments" def test_none_args_rejected_without_dispatch(self, agent): """None arguments must not crash the dispatch path. Current contract: @@ -1884,7 +1928,6 @@ class TestConcurrentToolExecution: - def test_invoke_tool_dispatches_to_handle_function_call(self, agent): """_invoke_tool should route regular tools through handle_function_call.""" with patch("run_agent.handle_function_call", return_value="result") as mock_hfc: @@ -2036,6 +2079,54 @@ class TestConcurrentToolExecution: + @pytest.mark.parametrize("concurrent", [False, True]) + def test_tool_execution_middleware_replacement_emits_one_terminal_hook( + self, + agent, + monkeypatch, + concurrent, + ): + """A middleware replacement owns the result but not lifecycle closure.""" + tool_call = _mock_tool_call( + name="terminal", + arguments='{"command":"must-not-run"}', + call_id="terminal-1", + ) + mock_msg = _mock_assistant_msg(content="", tool_calls=[tool_call]) + messages = [] + hook_calls = [] + + def execution_middleware(**kwargs): + return '{"intercepted":true}' + + manager = SimpleNamespace(_middleware={ + "tool_request": [], + "tool_execution": [execution_middleware], + }) + monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager) + monkeypatch.setattr( + "hermes_cli.lifecycle.invoke_hook", + lambda hook_name, **kwargs: hook_calls.append((hook_name, kwargs)) or [], + ) + monkeypatch.setattr("hermes_cli.lifecycle.has_hook", lambda name: True) + + with patch( + "run_agent.handle_function_call", + side_effect=AssertionError("middleware replacement must not dispatch"), + ): + if concurrent: + agent._execute_tool_calls_concurrent(mock_msg, messages, "task-1") + else: + agent._execute_tool_calls_sequential(mock_msg, messages, "task-1") + + post_calls = [ + payload for name, payload in hook_calls if name == "post_tool_call" + ] + assert len(post_calls) == 1 + assert post_calls[0]["tool_name"] == "terminal" + assert post_calls[0]["tool_call_id"] == "terminal-1" + assert post_calls[0]["status"] == "ok" + assert post_calls[0]["result"] == '{"intercepted":true}' def test_agent_runtime_post_hook_ownership_predicate_covers_agent_tools(self, agent): """Sequential and concurrent agent-level paths share post-hook ownership.""" diff --git a/tests/test_model_tools.py b/tests/test_model_tools.py index 5110f42162993..a967f615759a5 100644 --- a/tests/test_model_tools.py +++ b/tests/test_model_tools.py @@ -60,6 +60,42 @@ 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_terminal_nonzero_exit_is_reported_as_error(self): + result = json.dumps({"output": "", "exit_code": 1, "error": None}) + with ( + patch("model_tools.registry.dispatch", return_value=result), + patch("hermes_cli.plugins.has_hook", return_value=True), + patch("hermes_cli.plugins.invoke_hook") as mock_invoke_hook, + ): + assert handle_function_call("terminal", {"command": "false"}) == result + + kwargs_by_hook = { + hook.args[0]: hook.kwargs for hook in mock_invoke_hook.call_args_list + } + for hook_name in ("post_tool_call", "transform_tool_result"): + assert kwargs_by_hook[hook_name]["status"] == "error" + assert kwargs_by_hook[hook_name]["error_type"] == "tool_error" + assert kwargs_by_hook[hook_name]["error_message"] == "exit 1" + + 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 def test_tool_request_and_execution_middleware_wrap_registry_dispatch(self, monkeypatch): seen = {} @@ -115,6 +151,75 @@ class TestHandleFunctionCall: assert pre_call[1]["middleware_trace"] == expected_trace assert post_call[1]["middleware_trace"] == expected_trace + def test_registry_exception_emits_terminal_tool_hook(self, monkeypatch): + from hermes_cli import lifecycle + + hook_calls = [] + monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_args, **_kwargs: []) + monkeypatch.setattr(lifecycle, "has_hook", lambda name: name == "post_tool_call") + monkeypatch.setattr( + lifecycle, + "invoke_hook", + lambda name, **kwargs: hook_calls.append((name, kwargs)) or [], + ) + monkeypatch.setattr( + "model_tools.registry.dispatch", + lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("boom")), + ) + + result = json.loads( + handle_function_call( + "web_search", + {"q": "test"}, + task_id="task-1", + session_id="session-1", + tool_call_id="tool-1", + ) + ) + + assert "error" in result + [post_call] = [call for call in hook_calls if call[0] == "post_tool_call"] + assert post_call[1]["status"] == "error" + assert post_call[1]["error_type"] == "RuntimeError" + assert post_call[1]["duration_ms"] >= 0 + + def test_acp_edit_denial_emits_blocked_terminal_tool_hook(self, monkeypatch): + from hermes_cli import lifecycle + + hook_calls = [] + monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_args, **_kwargs: []) + monkeypatch.setattr(lifecycle, "has_hook", lambda name: name == "post_tool_call") + monkeypatch.setattr( + lifecycle, + "invoke_hook", + lambda name, **kwargs: hook_calls.append((name, kwargs)) or [], + ) + monkeypatch.setattr( + "acp_adapter.edit_approval.maybe_require_edit_approval", + lambda *_args, **_kwargs: json.dumps({"error": "Edit approval denied"}), + ) + monkeypatch.setattr( + "model_tools.registry.dispatch", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("denied edit must not dispatch") + ), + ) + + result = json.loads( + handle_function_call( + "write_file", + {"path": "private.txt", "content": "private"}, + task_id="task-1", + session_id="session-1", + tool_call_id="tool-1", + ) + ) + + assert result == {"error": "Edit approval denied"} + [post_call] = [call for call in hook_calls if call[0] == "post_tool_call"] + assert post_call[1]["status"] == "blocked" + assert post_call[1]["error_type"] == "edit_approval_denied" + # ========================================================================= # Agent loop tools diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index 9c4b0cbc681ba..8892a089fc419 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -1245,6 +1245,71 @@ class TestApprovalTimeoutIsNotConsent: assert "NOT consented" in r["message"] assert "rephrase" in r["message"].lower() + def test_timeout_emits_post_hook_with_timeout_outcome(self, monkeypatch): + """Plugins must be able to distinguish timeout from explicit deny. + + This is what an audit / notification plugin needs to alert + operators on 'agent asked, user never replied' incidents like #24912. + """ + from tools import approval as mod + self._force_short_timeout(monkeypatch, seconds=1) + mod.register_gateway_notify(self.SESSION_KEY, lambda data: None) + + hook_calls = [] + original_fire = mod._fire_approval_hook + + def _capture(event_name, **kwargs): + hook_calls.append((event_name, kwargs)) + return original_fire(event_name, **kwargs) + + monkeypatch.setattr(mod, "_fire_approval_hook", _capture) + + mod.check_all_command_guards("rm -rf .git", "local") + + # post_approval_response must be in the hook log with choice=timeout + posts = [c for c in hook_calls if c[0] == "post_approval_response"] + assert posts, "post_approval_response hook did not fire" + last_post = posts[-1][1] + assert last_post.get("choice") == "timeout", ( + f"hook choice should be 'timeout' on no-response, got {last_post.get('choice')!r}" + ) + + def test_notify_failure_emits_post_hook_and_cleans_up(self, monkeypatch): + """A failed notification still terminates the approval lifecycle.""" + from tools import approval as mod + + hook_calls = [] + + def _capture(event_name, **kwargs): + hook_calls.append((event_name, kwargs)) + + monkeypatch.setattr(mod, "_fire_approval_hook", _capture) + + def _fail_notify(_data): + raise RuntimeError("private gateway failure") + + decision = mod._await_gateway_decision( + self.SESSION_KEY, + _fail_notify, + { + "command": "redacted-command", + "description": "redacted-description", + "pattern_key": "dangerous", + "pattern_keys": ["dangerous"], + }, + ) + + assert decision == { + "resolved": False, + "choice": None, + "notify_failed": True, + } + assert self.SESSION_KEY not in mod._gateway_queues + assert [name for name, _ in hook_calls] == [ + "pre_approval_request", + "post_approval_response", + ] + assert hook_calls[-1][1]["choice"] == "notify_failed" class TestTirithImportErrorFailOpenPolicy: """Regression guard for #20733. diff --git a/tests/tools/test_request_tool_approval.py b/tests/tools/test_request_tool_approval.py index da6aaa27b97ed..5f2c8ef401630 100644 --- a/tests/tools/test_request_tool_approval.py +++ b/tests/tools/test_request_tool_approval.py @@ -50,13 +50,35 @@ class TestRequestToolApproval: assert res["approved"] is True def test_cli_deny_blocks(self, monkeypatch): + from hermes_cli import lifecycle + monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True) monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: False) monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *a, **k: "deny") - res = request_tool_approval("terminal", "curl PUT to external API") + events = [] + monkeypatch.setattr( + lifecycle, + "invoke_hook", + lambda hook_name, **kwargs: events.append((hook_name, kwargs)) or [], + ) + tokens = approval.set_current_observability_context( + turn_id="turn-1", + tool_call_id="call-1", + ) + try: + res = request_tool_approval("terminal", "curl PUT to external API") + finally: + approval.reset_current_observability_context(tokens) assert res["approved"] is False assert "denied" in res["message"].lower() assert res["pattern_key"].startswith("plugin_rule:") + assert [name for name, _ in events] == [ + "pre_approval_request", + "post_approval_response", + ] + assert all(event["turn_id"] == "turn-1" for _, event in events) + assert all(event["tool_call_id"] == "call-1" for _, event in events) + assert events[-1][1]["choice"] == "deny" def test_cli_session_persists_session_only(self, monkeypatch): monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True) diff --git a/tests/tools/test_tool_search.py b/tests/tools/test_tool_search.py index 575fcffb21afd..d4d66f283c13a 100644 --- a/tests/tools/test_tool_search.py +++ b/tests/tools/test_tool_search.py @@ -297,6 +297,48 @@ class TestHandleFunctionCallIntegration: # dispatch path completed without error. assert "matches" in parsed or "error" in parsed + def test_tool_search_emits_one_terminal_hook(self, monkeypatch): + """Inline bridge results still complete the tool lifecycle.""" + import model_tools + from hermes_cli import lifecycle + from tools import tool_search + + events = [] + monkeypatch.setattr( + lifecycle, + "has_hook", + lambda name: name == "post_tool_call", + ) + monkeypatch.setattr( + lifecycle, + "invoke_hook", + lambda name, **kwargs: events.append((name, kwargs)), + ) + monkeypatch.setattr( + tool_search, + "dispatch_tool_search", + lambda *args, **kwargs: json.dumps({"matches": []}), + ) + + result = model_tools.handle_function_call( + function_name="tool_search", + function_args={"query": "private-query"}, + session_id="private-session", + task_id="private-task", + turn_id="private-turn", + api_request_id="private-request", + tool_call_id="private-call", + ) + + assert json.loads(result) == {"matches": []} + assert len(events) == 1 + hook_name, payload = events[0] + assert hook_name == "post_tool_call" + assert payload["status"] == "ok" + assert payload["turn_id"] == "private-turn" + assert payload["api_request_id"] == "private-request" + assert payload["tool_call_id"] == "private-call" + class TestRegression_OpenClawCron84141: """Regression guard for the OpenClaw cron-tool-loss class of bug. diff --git a/tools/approval.py b/tools/approval.py index 151e534065539..f5e7f0beb1546 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -3170,8 +3170,27 @@ def _run_approval_gate( ), } + _fire_approval_hook( + "pre_approval_request", + command=display_target, + description=description, + pattern_key=pattern_key, + pattern_keys=[pattern_key], + session_key=session_key, + surface="cli", + ) choice = prompt_dangerous_approval(display_target, description, approval_callback=approval_callback) + _fire_approval_hook( + "post_approval_response", + command=display_target, + description=description, + pattern_key=pattern_key, + pattern_keys=[pattern_key], + session_key=session_key, + surface="cli", + choice=choice, + ) if choice == "timeout": return { @@ -3464,6 +3483,16 @@ def _await_gateway_decision(session_key: str, notify_cb, approval_data: dict, except Exception as exc: logger.warning("Gateway approval notify failed: %s", exc) _drop_entry() + _fire_approval_hook( + "post_approval_response", + command=command, + description=description, + pattern_key=primary_key, + pattern_keys=list(all_keys), + session_key=session_key, + surface=surface, + choice="notify_failed", + ) return {"resolved": False, "choice": None, "notify_failed": True} # Block until the user responds or the canonical approval timeout elapses