RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #68882 from afourniernv/feat/hermes-relay-tool-metrics 

feat(observability): aggregate bounded tool metrics
This commit is contained in:
Jeffrey Quesnelle 2026-08-04 15:04:29 -04:00 committed by GitHub
parent b3e45a3d46 daf67f2e59
commit f40fbcf409
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 2712 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
agent/tool_executor.py
View File

@ -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(

30
docs/observability/relay-shared-metrics.md
View File

@ -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.

332
hermes_cli/observability/relay_shared_metrics.py
View File

@ -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():

159
hermes_cli/observability/schemas/hermes.shared_metrics.v2.schema.json
View File

@ -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
}
}
}
}
}

282
hermes_cli/observability/shared_metrics_contract.py
View File

@ -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(

18
hermes_cli/observability/shared_metrics_subscriber.py
View File

@ -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

34
hermes_cli/plugins.py
View File

@ -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.

124
model_tools.py
View File

@ -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
# =============================================================================

212
scripts/smoke_nemo_relay_shared_metrics.py
View File

@ -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")

75
tests/hermes_cli/test_plugins.py
View File

@ -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

186
tests/hermes_cli/test_relay_shared_metrics.py
View File

@ -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")

933
tests/hermes_cli/test_relay_shared_metrics_runtime.py
View File

File diff suppressed because it is too large Load Diff

97
tests/run_agent/test_run_agent.py
View File

@ -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."""

105
tests/test_model_tools.py
View File

@ -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

65
tests/tools/test_approval.py
View File

@ -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.

24
tests/tools/test_request_tool_approval.py
View File

@ -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)

42
tests/tools/test_tool_search.py
View File

@ -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.

29
tools/approval.py
View File

@ -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