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

feat(observability): aggregate bounded tool metrics

This commit is contained in:
Alex Fournier 2026-07-29 09:04:34 -07:00
parent a0476b3605
commit 8b0c3da8c0
17 changed files with 2351 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
agent/tool_executor.py
View File

@ -299,6 +299,7 @@ class _ManagedToolResult:
args: dict[str, Any]
middleware_trace: list[dict[str, Any]]
blocked: bool
dispatched: bool
class _ConcurrentToolAuthorizationGate:
@ -343,12 +344,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,
)
@ -531,6 +533,7 @@ def _run_agent_tool_execution_middleware(
args=state["args"],
middleware_trace=state["middleware_trace"],
blocked=bool(state["blocked"]),
dispatched=bool(state["dispatched"]),
)
@ -648,12 +651,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,
@ -800,6 +818,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:
@ -843,6 +862,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")
@ -873,6 +893,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])
@ -1122,6 +1153,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,
@ -1164,6 +1196,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"
@ -1336,12 +1381,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,
@ -1356,6 +1416,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,
@ -1411,6 +1482,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()
@ -1422,7 +1494,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,
@ -1453,7 +1525,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,
@ -1492,7 +1564,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,
@ -1514,7 +1586,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,
@ -1535,7 +1607,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,
@ -1569,7 +1641,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,
@ -1602,7 +1674,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,
@ -1638,7 +1710,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,
@ -1698,6 +1770,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,
@ -1768,6 +1841,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,
@ -1823,7 +1897,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(

27
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
@ -87,6 +88,18 @@ 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
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
@ -125,6 +138,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.

259
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,85 @@ 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 = "tool_call" if tool_call_id else "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:
matches = [
candidate
for key, candidate in session.tool_calls.items()
if key[0] == task.task_id and key[-1] == tool_call_id
]
tool_call = matches[0] if len(matches) == 1 else None
if tool_call is not None:
tool_call.approval_outcome = outcome
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)
identity = self._tool_call_identity(event)
if identity in task.completed_tool_call_ids:
return
task.completed_tool_call_ids.add(identity)
task.tool_call_ids.add(identity)
tool_call = session.tool_calls.pop(
(task_id, *identity),
None,
)
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 +649,143 @@ 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 _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 +868,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 +889,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:
@ -695,8 +937,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
@ -725,8 +966,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(kwargs)
elif hook_name == "post_tool_call":
runtime.record_tool_call(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":

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

@ -64,6 +64,12 @@
},
{
"$ref": "#/$defs/task_finished_counter"
},
{
"$ref": "#/$defs/tool_call_counter"
},
{
"$ref": "#/$defs/tool_approval_counter"
}
]
}
@ -74,6 +80,31 @@
"type": "string",
"pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
},
"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"
]
},
"model_call_counter": {
"type": "object",
"additionalProperties": false,
@ -118,6 +149,134 @@
}
}
},
"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
}
}
},
"task_started_counter": {
"type": "object",
"additionalProperties": false,

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 RUNTIME_INSTANCE_KEY
@ -11,10 +12,14 @@ SCHEMA_VERSION = "hermes.metrics.event.v1"
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"
MODEL_CALL_METRIC = "hermes.model_call.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(
@ -89,6 +94,72 @@ 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"})
_TOOL_NAMES_BY_CATEGORY: dict[str, frozenset[str]] = {
"code_execution": frozenset({"execute_code"}),
"communication": frozenset({"discord", "email", "meet"}),
"computer_use": frozenset({"computer_use"}),
"delegation": frozenset({"delegate_task"}),
"file": frozenset({"patch", "read_file", "search_files", "write_file"}),
"memory": frozenset({"memory", "session_search"}),
"planning": frozenset({"clarify", "todo"}),
"scheduler": frozenset({"cronjob"}),
"skill": frozenset({"skill_manage", "skill_view", "skills_list"}),
"terminal": frozenset({"close_terminal", "process", "read_terminal", "terminal"}),
"web": frozenset({"web_extract", "web_search", "x_search"}),
}
_COUNTER_DIMENSION_VALUES: dict[str, dict[str, frozenset[str]]] = {
TASK_STARTED_METRIC: {
@ -106,6 +177,17 @@ _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(
{*_COUNTER_DIMENSION_VALUES, MODEL_CALL_METRIC}
@ -135,21 +217,24 @@ 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 model_call_dimensions(event: Any) -> dict[str, str] | None:
"""Return package dimensions for one valid primary model-call end event."""
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 None
return False
relay_metadata = set(metadata) - {SCHEMA_KEY, RUNTIME_INSTANCE_KEY}
if relay_metadata - {"otel.status_code"} or metadata.get(
return not relay_metadata - {"otel.status_code"} and metadata.get(
"otel.status_code", "OK"
) not in {"OK", "ERROR"}:
) in {"OK", "ERROR"}
def model_call_dimensions(event: Any) -> dict[str, str] | None:
"""Return package dimensions for one valid primary model-call end event."""
if not _event_metadata_is_valid(event):
return None
if (
str(getattr(event, "kind", "") or "") != "scope"
@ -179,13 +264,7 @@ def 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"
@ -233,6 +312,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 = (
@ -361,6 +490,131 @@ def count_bucket(count: int) -> str:
return "gte_11"
def tool_category(kwargs: dict[str, Any]) -> str:
"""Map a raw Hermes tool name to a stable low-cardinality category."""
name = str(kwargs.get("tool_name") or "").strip().lower()
if not name:
return "unknown"
if name.startswith(("mcp.", "mcp_", "mcp__")):
return "mcp"
for category, names in _TOOL_NAMES_BY_CATEGORY.items():
if name in names:
return category
if name.startswith("browser_"):
return "browser"
if name.startswith(("vision_", "image_", "video_", "text_to_speech")):
return "media"
if name.startswith("ha_"):
return "home_automation"
if name.startswith("kanban_"):
return "planning"
if name.startswith("project_"):
return "project"
if name.startswith(("discord_", "email_", "feishu_", "slack_", "sms_", "yb_")):
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_CALL_METRIC, model_call_dimensions, task_counter
from .shared_metrics_contract import (
MODEL_CALL_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_CALL_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

@ -1020,13 +1020,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
@ -1060,7 +1071,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,
@ -1133,6 +1147,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
@ -1162,15 +1193,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
@ -1179,16 +1220,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(
@ -1197,6 +1240,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,
@ -1285,11 +1330,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).
@ -1388,7 +1460,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,
@ -1416,7 +1491,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_call.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_call.count"]
expected_model = {
"name": "hermes.model_call.count",
"dimensions": {
"model": MODEL_CANARY,
"provider": "custom",
},
"value": 1,
"packaged_value": 1,
"value": 2,
"packaged_value": 2,
}
if by_name["hermes.model_call.count"] != expected_model:
if model != expected_model:
raise AssertionError(
f"Unexpected model counter: {by_name['hermes.model_call.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_call.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_call.count"]["dimensions"]
if model_dimensions != {
[model] = metrics["hermes.model_call.count"]
if model["dimensions"] != {
"model": MODEL_CANARY,
"provider": "custom",
}:
} or model["value"] != 2:
raise AssertionError(
f"Unexpected model metric: {metrics['hermes.model_call.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")

65
tests/hermes_cli/test_plugins.py
View File

@ -888,6 +888,44 @@ 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(
@ -981,6 +1019,33 @@ class TestResolvePreToolBlock:
)
assert resolve_pre_tool_block("write_file", {}) is None
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

182
tests/hermes_cli/test_relay_shared_metrics.py
View File

@ -31,6 +31,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,
@ -40,6 +46,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,
)
@ -72,6 +86,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",
@ -197,6 +216,125 @@ def test_package_schema_matches_the_task_contract():
assert set(terminal["termination"]["enum"]) == TASK_TERMINATIONS
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(
("name", "expected"),
[
("", "unknown"),
("read_file", "file"),
("terminal", "terminal"),
("execute_code", "code_execution"),
("delegate_task", "delegation"),
("skill_manage", "skill"),
("browser_navigate", "browser"),
("image_generate", "media"),
("ha_call_service", "home_automation"),
("kanban_create", "planning"),
("project_switch", "project"),
("discord", "communication"),
("feishu_doc_read", "communication"),
("mcp__github__get_issue", "mcp"),
("private_plugin_tool", "other"),
],
)
def test_tool_category_is_bounded(name, expected):
assert tool_category({"tool_name": name}) == expected
@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",
@ -449,6 +587,50 @@ def test_task_subscriber_contract_accepts_only_bounded_scope_events():
assert task_counter(end) is None
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.v1"},
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.v1"},
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_rejects_an_unsupported_schema_version(tmp_path):
database_path = tmp_path / "metrics.sqlite3"
with sqlite3.connect(database_path) as connection:

748
tests/hermes_cli/test_relay_shared_metrics_runtime.py
View File

@ -30,10 +30,13 @@ 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_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,
@ -41,6 +44,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,
@ -89,6 +93,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:
current = self._scope.get()
@ -130,6 +145,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))
@ -181,6 +231,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",
@ -196,6 +247,21 @@ 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",
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,
@ -204,6 +270,8 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa
args={"command": "sensitive-command"},
result={"output": "sensitive-tool-result"},
status="ok",
duration_ms=275,
retry_count=0,
)
lifecycle.invoke_hook(
"api_request_error",
@ -243,6 +311,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"
]
@ -257,6 +329,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] == {
@ -270,6 +353,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() == []
@ -282,12 +366,35 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa
"hermes.model_call.count",
"hermes.task_run.finished",
"hermes.task_run.started",
"hermes.tool_approval.count",
"hermes.tool_call.count",
}
assert metrics["hermes.model_call.count"]["dimensions"] == {
"model": "claude-sonnet",
"provider": "anthropic",
}
assert metrics["hermes.model_call.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",
@ -355,6 +462,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,
@ -363,6 +485,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",
@ -384,6 +508,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,
@ -405,6 +546,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,
@ -439,6 +597,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"]
@ -487,10 +682,185 @@ 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)
def test_direct_runtime_is_disabled_by_default(tmp_path, monkeypatch):
fake = _Relay()
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes-home"))
@ -1061,9 +1431,7 @@ def test_shared_metrics_policy_and_store_are_profile_scoped(tmp_path, monkeypatc
monkeypatch.setattr(
"hermes_cli.config.read_raw_config",
lambda: {
"telemetry": {
"shared_metrics": {"enabled": get_hermes_home() == profile_a}
}
"telemetry": {"shared_metrics": {"enabled": get_hermes_home() == profile_a}}
},
)
relay_shared_metrics._reset_for_tests()
@ -1135,9 +1503,7 @@ def test_shared_metrics_subscribers_isolate_two_enabled_profiles(tmp_path, monke
platform="cli",
result={"completed": True},
)
relay_shared_metrics._get_runtime().close_session(
{"session_id": "shared"}
)
relay_shared_metrics._get_runtime().close_session({"session_id": "shared"})
finally:
reset_hermes_home_override(token)
@ -1269,24 +1635,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",
@ -1709,12 +2080,10 @@ def test_core_runtime_parents_subagent_session_without_exposing_ids(
def test_subagent_stop_hook_does_not_own_child_session_lifetime(direct_runtime):
runtime = relay_runtime.get_runtime()
assert runtime is not None
child = runtime.register_subagent(
{
"parent_session_id": "parent",
"child_session_id": "child",
}
)
child = runtime.register_subagent({
"parent_session_id": "parent",
"child_session_id": "child",
})
assert child is not None
lifecycle.invoke_hook(
@ -1779,6 +2148,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")
@ -1786,6 +2156,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
@ -1793,6 +2164,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")
@ -2171,6 +2543,11 @@ def test_task_terminal_counts_logical_calls_retries_and_unique_tools(direct_runt
"model": "nvidia/nemotron-3-super-120b-a12b",
"provider": "nvidia",
}
tool_ends = [
event for event in direct_runtime.events if event[0] == "tool.call_end"
]
assert len(tool_ends) == 2
assert all(event[2]["tool_category"] == "terminal" for event in tool_ends)
[task_end] = [
event
for event in direct_runtime.events
@ -2189,6 +2566,317 @@ def test_task_terminal_counts_logical_calls_retries_and_unique_tools(direct_runt
}
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_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_task_terminal_counts_explicit_retry_with_new_request_id(direct_runtime):
base = {
"session_id": "s1",

113
tests/run_agent/test_run_agent.py
View File

@ -2466,11 +2466,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()
@ -2482,13 +2489,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()
@ -2497,6 +2513,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:
@ -2982,10 +3026,17 @@ class TestConcurrentToolExecution:
flushed.append([m.copy() for m in flush_messages if m.get("role") == "tool"])
agent._flush_messages_to_session_db = MagicMock(side_effect=record_flush)
terminal_hook = MagicMock()
start = _time.monotonic()
try:
with patch("run_agent.handle_function_call", side_effect=fake_handle):
with (
patch("run_agent.handle_function_call", side_effect=fake_handle),
patch(
"model_tools._emit_post_tool_call_hook",
terminal_hook,
),
):
agent._execute_tool_calls_concurrent(mock_msg, messages, "task-1")
finally:
blocker.set()
@ -3000,6 +3051,13 @@ class TestConcurrentToolExecution:
assert [batch[-1]["tool_call_id"] for batch in flushed] == ["c1", "c2"]
assert "fast-result" in flushed[0][-1]["content"]
assert "timed out after" in flushed[1][-1]["content"]
[timeout_hook] = [
call
for call in terminal_hook.call_args_list
if call.kwargs["tool_call_id"] == "c2"
]
assert timeout_hook.kwargs["status"] == "timeout"
assert timeout_hook.kwargs["duration_ms"] == 100
def test_concurrent_timeout_prefers_late_real_result_over_timeout_message(self, agent, monkeypatch):
"""A worker that finishes in the window between the deadline snapshot
@ -3475,6 +3533,55 @@ class TestConcurrentToolExecution:
assert post_call[1]["args"] == {"todos": [], "request_rewritten": True, "merge": True}
assert post_call[1]["middleware_trace"] == [{"source": "request-test"}]
@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_concurrent_agent_level_tool_preserves_request_middleware_trace(self, agent, monkeypatch):
tool_call = _mock_tool_call(name="todo", arguments='{"todos":[]}', call_id="todo-1")
mock_msg = _mock_assistant_msg(content="", tool_calls=[tool_call])

86
tests/test_model_tools.py
View File

@ -127,6 +127,23 @@ 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
@ -201,6 +218,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

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

@ -456,6 +456,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.

19
tools/approval.py
View File

@ -2990,8 +2990,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 == "deny":
return {