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

fix(observability): harden Relay metrics isolation and aggregation 

Signed-off-by: Alex Fournier <afournier@nvidia.com>
This commit is contained in:
Alex Fournier 2026-07-19 08:55:07 -04:00
parent 64faff6768
commit 056e7df0e0
20 changed files with 2481 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
agent/turn_context.py
View File

@ -703,6 +703,7 @@ def build_turn_context(
is_first_turn=(not bool(conversation_history)),
model=agent.model,
platform=getattr(agent, "platform", None) or "",
parent_session_id=getattr(agent, "_parent_session_id", None) or "",
sender_id=getattr(agent, "_user_id", None) or "",
)
_ctx_parts: list[str] = []

2
agent/turn_finalizer.py
View File

@ -571,7 +571,9 @@ def finalize_turn(
task_id=effective_task_id,
turn_id=turn_id,
completed=completed,
failed=failed,
interrupted=interrupted,
turn_exit_reason=_turn_exit_reason,
model=agent.model,
platform=getattr(agent, "platform", None) or "",
)

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

@ -26,15 +26,15 @@ marks do not require Hermes plugin registration. Shared-metrics marks must
still contain only fields approved by the versioned allowlist; the hard
dependency does not change the collection or privacy policy.
## Current Slice
## Current Slices
The first vertical slice records one logical model-call counter:
The current vertical slices record logical model calls and top-level task runs:
```text
Hermes API hooks
-> Relay session and LLM lifecycle
Hermes turn, API, and tool hooks
-> Relay session, task, and LLM lifecycle
-> Hermes shared-metrics subscriber
-> SQLite counter
-> SQLite counters
-> immutable JSON delta package
```
@ -44,6 +44,18 @@ and outcome values. Prompts, responses, exact model IDs, endpoints, errors,
session IDs, task IDs, and request IDs are not included in the metrics event or
package.
Each task run is a Relay `Function` scope named `hermes.task_run`, parented to
the owning Hermes session. The start counter contains only bounded execution
surface and entrypoint values. The terminal counter contains bounded outcome,
end reason, termination status, duration, logical model-call count, terminal
tool-call count, and provider-retry count buckets. Retries are additional
provider attempts for the same Hermes API request ID; they do not inflate the
logical model-call count. Tool calls are deduplicated by their Hermes tool-call
ID after a terminal tool result is observed. The outer `AIAgent` execution
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.
Local state is written under:
```text
@ -67,6 +79,6 @@ 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, one counter was
stored, one package was exported, and prompt, response, and exact-model
canaries are absent from the package.
The smoke verifies the model request reached the local server, model and task
counters were stored, one package was exported, and prompt, response, and
exact-model canaries are absent from the package.

12
hermes_cli/hooks.py
View File

@ -149,7 +149,17 @@ _DEFAULT_PAYLOADS = {
"changed_paths": ["src/app.tsx"],
},
"on_session_start": {"session_id": "test-session"},
"on_session_end": {"session_id": "test-session"},
"on_session_end": {
"session_id": "test-session",
"task_id": "test-task",
"turn_id": "test-turn",
"completed": True,
"failed": False,
"interrupted": False,
"turn_exit_reason": "text_response(stop)",
"model": "gpt-4",
"platform": "cli",
},
"on_session_finalize": {"session_id": "test-session"},
"on_session_reset": {"session_id": "test-session"},
"pre_api_request": {

29
hermes_cli/middleware.py
View File

@ -127,18 +127,31 @@ def apply_tool_request_middleware(
Middleware may return ``{"args": {...}}`` to replace the effective tool
arguments before hooks, guardrails, approvals, and execution see them.
"""
if not _has_middleware(TOOL_REQUEST_MIDDLEWARE):
return RequestMiddlewareResult(
payload=args,
original_payload=args,
changed=False,
trace=[],
)
original_args = _safe_copy(args)
current_args = _safe_copy(original_args)
trace: List[Dict[str, Any]] = []
session_id = str(context.get("session_id") or "")
if session_id:
from hermes_cli.observability import relay_runtime
relay_args = relay_runtime.apply_tool_request_intercepts(
session_id=session_id,
tool_name=tool_name,
args=current_args,
)
if relay_args != current_args:
current_args = _safe_copy(relay_args)
trace.append({"source": "nemo_relay"})
if not _has_middleware(TOOL_REQUEST_MIDDLEWARE):
return RequestMiddlewareResult(
payload=args if not trace else current_args,
original_payload=args,
changed=bool(trace),
trace=trace,
)
for result in _invoke_middleware(
TOOL_REQUEST_MIDDLEWARE,
tool_name=tool_name,

152
hermes_cli/observability/relay_runtime.py
View File

@ -1,20 +1,26 @@
"""Process-wide NeMo Relay runtime owned by Hermes core."""
"""Profile-scoped NeMo Relay runtimes owned by Hermes core."""
from __future__ import annotations
import atexit
import asyncio
import contextvars
import importlib
import inspect
import logging
import threading
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable
from hermes_constants import get_hermes_home
logger = logging.getLogger(__name__)
SESSION_SCOPE = "hermes.session"
RUNTIME_SCHEMA_KEY = "hermes.relay.schema_version"
RUNTIME_SCHEMA_VERSION = "hermes.relay.runtime.v1"
RUNTIME_INSTANCE_KEY = "hermes.relay.runtime_instance"
SESSION_START_HOOKS = frozenset({"on_session_start"})
SESSION_CLOSE_HOOKS = frozenset({"on_session_finalize", "on_session_reset"})
@ -28,7 +34,7 @@ HANDLED_HOOKS = (
)
_RUNTIME_FAILED = object()
_RUNTIME: RelayRuntime | object | None = None
_RUNTIMES: dict[str, RelayRuntime | object] = {}
_RUNTIME_LOCK = threading.RLock()
@ -47,8 +53,10 @@ class RelaySession:
class RelayRuntime:
"""Own Relay session scopes independently of any exporter or plugin."""
def __init__(self, relay: Any = None) -> None:
def __init__(self, relay: Any = None, *, profile_key: str | None = None) -> None:
self.relay = relay or _load_nemo_relay()
self.profile_key = profile_key or current_profile_key()
self.runtime_id = uuid.uuid4().hex
self._sessions_lock = threading.RLock()
self._sessions: dict[str, RelaySession] = {}
self._subagent_parents: dict[str, str] = {}
@ -83,6 +91,7 @@ class RelayRuntime:
scope_metadata = {
**(metadata or {}),
RUNTIME_SCHEMA_KEY: RUNTIME_SCHEMA_VERSION,
RUNTIME_INSTANCE_KEY: self.runtime_id,
}
if session.parent_session_id:
parent = self.ensure_session({
@ -123,10 +132,11 @@ class RelayRuntime:
self._subagent_parents[child_session_id] = parent_session_id
def unregister_subagent(self, event: dict[str, Any]) -> None:
"""Forget a delegated-session relationship after its terminal hook."""
"""Close a delegated session and forget its parent relationship."""
child_session_id = str(event.get("child_session_id") or "")
if not child_session_id:
return
self.close_session({"session_id": child_session_id})
with self._sessions_lock:
self._subagent_parents.pop(child_session_id, None)
@ -167,6 +177,32 @@ class RelayRuntime:
# re-enter the same logical session without re-entering Context.
return session.context.copy().run(invoke)
async def run_in_session_async(
self,
session: RelaySession,
callback: Callable[..., Any],
*args: Any,
allow_closing: bool = False,
**kwargs: Any,
) -> Any:
"""Create and await an operation inside the session's saved context."""
with session.lock:
if session.closing and not allow_closing:
raise RuntimeError("Hermes Relay session is closing")
if session.context is None or session.handle is None:
raise RuntimeError("Hermes Relay session context is unavailable")
context = session.context.copy()
async def invoke() -> Any:
self.relay.get_scope_stack()
result = callback(*args, **kwargs)
if inspect.isawaitable(result):
return await result
return result
task = context.run(asyncio.create_task, invoke())
return await task
def emit_mark(
self,
name: str,
@ -189,6 +225,32 @@ class RelayRuntime:
)
return True
def apply_tool_request_intercepts(
self,
*,
session_id: str,
tool_name: str,
args: dict[str, Any],
) -> dict[str, Any]:
"""Apply Relay request rewriting before Hermes authorizes a tool call."""
request_intercepts = getattr(
getattr(self.relay, "tools", None),
"request_intercepts",
None,
)
if not callable(request_intercepts):
return args
session = self.ensure_session({"session_id": session_id})
if session is None:
return args
result = self.run_in_session(
session,
request_intercepts,
tool_name,
args,
)
return result if isinstance(result, dict) else args
def close_session(self, event: dict[str, Any]) -> None:
"""Close one session scope and remove it from the core registry."""
session_id = _session_id(event)
@ -208,7 +270,10 @@ class RelayRuntime:
self.relay.scope.pop,
session.handle,
output={},
metadata={RUNTIME_SCHEMA_KEY: RUNTIME_SCHEMA_VERSION},
metadata={
RUNTIME_SCHEMA_KEY: RUNTIME_SCHEMA_VERSION,
RUNTIME_INSTANCE_KEY: self.runtime_id,
},
allow_closing=True,
)
except Exception as exc:
@ -301,6 +366,25 @@ def emit_mark(
return False
def apply_tool_request_intercepts(
*,
session_id: str,
tool_name: str,
args: dict[str, Any],
) -> dict[str, Any]:
"""Return Relay-rewritten arguments at Hermes's authorization boundary."""
if not session_id:
return args
runtime = get_runtime()
if runtime is None:
return args
return runtime.apply_tool_request_intercepts(
session_id=session_id,
tool_name=tool_name,
args=args,
)
def ensure_session(*, session_id: str, **context: Any) -> RelaySession | None:
"""Create or return the shared Relay session used by Hermes core."""
runtime = get_runtime()
@ -331,27 +415,56 @@ def run_in_session(
return runtime.run_in_session(session, callback, *args, **kwargs)
async def run_in_session_async(
session_id: str,
callback: Callable[..., Any],
*args: Any,
**kwargs: Any,
) -> Any:
"""Await a Relay operation inside a shared Hermes session context."""
runtime = get_runtime()
if runtime is None:
raise RuntimeError("Hermes Relay runtime is unavailable")
session = runtime.get_session(session_id)
if session is None:
session = runtime.ensure_session({"session_id": session_id})
if session is None:
raise RuntimeError("Hermes Relay session is unavailable")
return await runtime.run_in_session_async(session, callback, *args, **kwargs)
def get_session_handle(session_id: str) -> Any:
"""Return the shared Relay handle for direct core instrumentation."""
runtime = get_runtime(create=False)
return None if runtime is None else runtime.get_session_handle(session_id)
def get_runtime(*, create: bool = True) -> RelayRuntime | None:
"""Return the process-wide Hermes Relay host."""
global _RUNTIME
def get_runtime(
*,
create: bool = True,
profile_key: str | None = None,
) -> RelayRuntime | None:
"""Return the Relay host for the active Hermes profile."""
key = profile_key or current_profile_key()
with _RUNTIME_LOCK:
if isinstance(_RUNTIME, RelayRuntime):
return _RUNTIME
if _RUNTIME is _RUNTIME_FAILED or not create:
runtime = _RUNTIMES.get(key)
if isinstance(runtime, RelayRuntime):
return runtime
if runtime is _RUNTIME_FAILED or not create:
return None
try:
_RUNTIME = RelayRuntime()
runtime = RelayRuntime(profile_key=key)
except Exception:
logger.warning("Hermes Relay runtime initialization failed", exc_info=True)
_RUNTIME = _RUNTIME_FAILED
_RUNTIMES[key] = _RUNTIME_FAILED
return None
return _RUNTIME
_RUNTIMES[key] = runtime
return runtime
def current_profile_key() -> str:
"""Return the canonical profile identity used for runtime isolation."""
return str(get_hermes_home().expanduser().resolve())
def _load_nemo_relay() -> Any:
@ -364,9 +477,10 @@ def _session_id(event: dict[str, Any]) -> str:
def _reset_for_tests() -> None:
"""Reset process-global core Relay state for isolated tests."""
global _RUNTIME
"""Reset all profile-scoped Relay hosts for isolated tests."""
with _RUNTIME_LOCK:
if isinstance(_RUNTIME, RelayRuntime):
_RUNTIME.shutdown()
_RUNTIME = None
runtimes = list(_RUNTIMES.values())
_RUNTIMES.clear()
for runtime in runtimes:
if isinstance(runtime, RelayRuntime):
runtime.shutdown()

511
hermes_cli/observability/relay_shared_metrics.py
View File

@ -3,10 +3,11 @@
from __future__ import annotations
import atexit
import contextvars
import logging
import threading
from dataclasses import dataclass, field
from functools import lru_cache
from time import monotonic_ns
from typing import Any, Callable
from hermes_cli import __version__
@ -18,8 +19,11 @@ from .shared_metrics_contract import (
SCHEMA_KEY,
SCHEMA_VERSION,
SUBSCRIBER_NAME,
TASK_SCOPE,
model_call_fields,
model_call_outcome,
task_start_fields,
task_terminal_fields,
)
from .shared_metrics_subscriber import SharedMetricsSubscriber
@ -30,13 +34,16 @@ HANDLED_HOOKS = frozenset({
"on_session_end",
"on_session_finalize",
"on_session_reset",
"pre_llm_call",
"pre_api_request",
"post_tool_call",
"post_api_request",
"api_request_error",
"subagent_stop",
})
_RUNTIME_FAILED = object()
_RUNTIME: _Runtime | object | None = None
_RUNTIMES: dict[str, _Runtime | object] = {}
_RUNTIME_LOCK = threading.RLock()
@ -47,6 +54,19 @@ class _ModelCall:
fields: dict[str, str]
@dataclass
class _TaskRun:
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)
turn_ids: set[str] = field(default_factory=set)
unidentified_tool_calls: int = 0
retry_count: int = 0
@dataclass
class _MetricsSession:
session_id: str
@ -54,6 +74,7 @@ class _MetricsSession:
lock: threading.RLock = field(default_factory=threading.RLock, repr=False)
closing: bool = False
model_calls: dict[str, _ModelCall] = field(default_factory=dict)
tasks: dict[str, _TaskRun] = field(default_factory=dict)
class _Runtime:
@ -66,9 +87,19 @@ class _Runtime:
self.host: relay_runtime.RelayRuntime = resolved_host
self.relay = self.host.relay
self._sessions_lock = threading.RLock()
self._active = True
self._sessions: dict[str, _MetricsSession] = {}
self.subscriber = SharedMetricsSubscriber(SharedMetricsStore(), __version__)
self.relay.subscribers.register(SUBSCRIBER_NAME, self.subscriber)
self._task_creation_lock = threading.RLock()
self._task_sessions_lock = threading.RLock()
self._task_sessions: dict[tuple[str, str], _MetricsSession] = {}
self._turn_sessions: dict[str, _MetricsSession] = {}
self._subscriber_name = f"{SUBSCRIBER_NAME}.{self.host.runtime_id}"
self.subscriber = SharedMetricsSubscriber(
SharedMetricsStore(),
__version__,
runtime_id=self.host.runtime_id,
)
self.relay.subscribers.register(self._subscriber_name, self.subscriber)
self._registered = True
atexit.register(self.shutdown)
@ -76,10 +107,12 @@ class _Runtime:
session_id = str(event.get("session_id") or "")
if not session_id:
return None
relay_session = self.host.ensure_session(event)
if relay_session is None:
return None
with self._sessions_lock:
if not self._active:
return None
relay_session = self.host.ensure_session(event)
if relay_session is None:
return None
session = self._sessions.get(session_id)
if session is None:
session = _MetricsSession(
@ -106,8 +139,77 @@ class _Runtime:
**kwargs,
)
def start_task(self, event: dict[str, Any]) -> _TaskRun | None:
"""Open one Relay function scope for a Hermes task run."""
task_key = self._task_key(event)
if task_key is None:
return None
_, task_id = task_key
with self._task_creation_lock:
owner = self._task_session(event)
if owner is not None:
with owner.lock:
if owner.closing:
return None
task = owner.tasks.get(task_id)
if task is not None:
self._remember_turn(owner, task, event)
return task
session = self.ensure_session(event)
if session is None:
return None
with session.lock:
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)
def push_task() -> Any:
self.relay.get_scope_stack()
return self.relay.scope.push(
TASK_SCOPE,
self.relay.ScopeType.Function,
handle=session.relay_session.handle,
input=start_fields,
metadata=self._event_metadata(),
)
handle = task_context.run(push_task)
task = _TaskRun(
handle=handle,
context=task_context,
started_ns=monotonic_ns(),
start_fields=start_fields,
)
session.tasks[task_id] = task
with self._task_sessions_lock:
self._task_sessions[task_key] = session
self._remember_turn(session, task, event)
return task
def _run_in_task(
self,
task: _TaskRun,
callback: Callable[..., Any],
*args: Any,
**kwargs: Any,
) -> Any:
def invoke() -> Any:
self.relay.get_scope_stack()
return callback(*args, **kwargs)
return task.context.copy().run(invoke)
def start_model_call(self, event: dict[str, Any]) -> None:
session = self.ensure_session(event)
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 task is None:
task = self.start_task(event)
session = self._task_session(event) if task is not None else None
if session is None:
session = self.ensure_session(event)
if session is None:
return
request_id = str(event.get("api_request_id") or "")
@ -118,27 +220,65 @@ class _Runtime:
with session.lock:
if session.closing:
return
if task is not None:
self._remember_turn(session, task, event)
existing = session.model_calls.get(request_id)
if existing is not None:
existing.fields = fields
if task is not None:
task.retry_count += 1
return
handle = self._run_in_session(
session,
self.relay.llm.call,
MODEL_CALL_SCOPE,
self.relay.LLMRequest({}, {}),
handle=session.relay_session.handle,
metadata={SCHEMA_KEY: SCHEMA_VERSION},
model_name=model_family,
)
if task is not None:
task.model_call_ids.add(request_id)
handle = self._run_in_task(
task,
self.relay.llm.call,
MODEL_CALL_SCOPE,
self.relay.LLMRequest({}, {}),
handle=task.handle,
metadata=self._event_metadata(),
model_name=model_family,
)
else:
handle = self._run_in_session(
session,
self.relay.llm.call,
MODEL_CALL_SCOPE,
self.relay.LLMRequest({}, {}),
handle=session.relay_session.handle,
metadata=self._event_metadata(),
model_name=model_family,
)
session.model_calls[request_id] = _ModelCall(
handle=handle,
task_id=str(event.get("task_id") or ""),
fields=fields,
)
def record_tool_call(self, event: dict[str, Any]) -> None:
"""Count one unique tool invocation under its owning 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
if task is None:
task = self.start_task(event)
session = self._task_session(event) if task 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
self._remember_turn(session, task, event)
if tool_call_id:
task.tool_call_ids.add(tool_call_id)
else:
task.unidentified_tool_calls += 1
def end_model_call(self, event: dict[str, Any], outcome: str | None = None) -> None:
session = self._session(event)
session = self._task_session(event, allow_task_id_fallback=True)
if session is None:
session = self._session(event)
if session is None:
return
request_id = str(event.get("api_request_id") or "")
@ -157,7 +297,9 @@ class _Runtime:
)
def end_pending_model_calls(self, event: dict[str, Any]) -> None:
session = self._session(event)
session = self._task_session(event, allow_task_id_fallback=True)
if session is None:
session = self._session(event)
if session is None:
return
with session.lock:
@ -165,6 +307,20 @@ class _Runtime:
return
self._end_pending_model_calls(session, event)
def finish_task(self, event: dict[str, Any]) -> None:
"""Close one task scope exactly once with bounded terminal fields."""
task_id = str(event.get("task_id") or "")
session = self._task_session(
event,
allow_task_id_fallback=True,
) or self._session(event)
if session is None:
return
with session.lock:
if session.closing:
return
self._finish_task(session, task_id, event)
def close_session(self, event: dict[str, Any]) -> None:
session = self._session(event)
if session is None:
@ -174,6 +330,19 @@ class _Runtime:
if session.closing:
return
session.closing = True
for task_id in list(session.tasks):
self._finish_task(
session,
task_id,
{
**event,
"task_id": task_id,
"completed": False,
"failed": True,
"interrupted": False,
"turn_exit_reason": "system_aborted",
},
)
self._end_pending_model_calls(session, event)
try:
self.relay.subscribers.flush()
@ -192,6 +361,7 @@ class _Runtime:
def shutdown(self) -> None:
with self._sessions_lock:
self._active = False
session_ids = list(self._sessions)
for session_id in session_ids:
self._safe(self.close_session, {"session_id": session_id})
@ -199,18 +369,103 @@ class _Runtime:
return
self._safe(self.relay.subscribers.flush)
self._export()
self._safe(self.relay.subscribers.deregister, SUBSCRIBER_NAME)
self._safe(self.relay.subscribers.deregister, self._subscriber_name)
self._registered = False
try:
atexit.unregister(self.shutdown)
except Exception:
pass
def deactivate(self) -> None:
"""Stop collection without exporting locally aggregated metrics."""
with self._sessions_lock:
self._active = False
self.subscriber.deactivate()
if self._registered:
self._safe(self.relay.subscribers.deregister, self._subscriber_name)
self._registered = False
with self._sessions_lock:
sessions = list(self._sessions.values())
for session in sessions:
with session.lock:
if session.closing:
continue
session.closing = True
for task_id in list(session.tasks):
self._finish_task(
session,
task_id,
{
"session_id": session.session_id,
"task_id": task_id,
"failed": True,
"turn_exit_reason": "system_aborted",
},
)
self._end_pending_model_calls(session, {})
with self._sessions_lock:
self._sessions.clear()
with self._task_sessions_lock:
self._task_sessions.clear()
self._turn_sessions.clear()
try:
atexit.unregister(self.shutdown)
except Exception:
pass
def _session(self, event: dict[str, Any]) -> _MetricsSession | None:
session_id = str(event.get("session_id") or "")
with self._sessions_lock:
return self._sessions.get(session_id)
@staticmethod
def _task_key(event: dict[str, Any]) -> tuple[str, str] | None:
session_id = str(event.get("session_id") or "")
task_id = str(event.get("task_id") or "")
if not session_id or not task_id:
return None
return session_id, task_id
def _task_session(
self,
event: dict[str, Any],
*,
allow_task_id_fallback: bool = False,
) -> _MetricsSession | None:
task_key = self._task_key(event)
if task_key is None:
return None
turn_id = str(event.get("turn_id") or "")
with self._task_sessions_lock:
if turn_id:
owner = self._turn_sessions.get(turn_id)
if owner is not None:
return owner
owner = self._task_sessions.get(task_key)
if owner is not None or not allow_task_id_fallback:
return owner
task_id = task_key[1]
candidates: list[_MetricsSession] = []
for (_, candidate_task_id), session in self._task_sessions.items():
if candidate_task_id != task_id:
continue
if not any(candidate is session for candidate in candidates):
candidates.append(session)
return candidates[0] if len(candidates) == 1 else None
def _remember_turn(
self,
session: _MetricsSession,
task: _TaskRun,
event: dict[str, Any],
) -> None:
turn_id = str(event.get("turn_id") or "")
if not turn_id:
return
task.turn_ids.add(turn_id)
with self._task_sessions_lock:
self._turn_sessions[turn_id] = session
def _finish_model_call(
self,
session: _MetricsSession,
@ -221,13 +476,23 @@ class _Runtime:
if model_call is None:
return
try:
self._run_in_session(
session,
self.relay.llm.call_end,
model_call.handle,
{**model_call.fields, "outcome": outcome},
metadata={SCHEMA_KEY: SCHEMA_VERSION},
)
task = session.tasks.get(model_call.task_id)
if task is not None:
self._run_in_task(
task,
self.relay.llm.call_end,
model_call.handle,
{**model_call.fields, "outcome": outcome},
metadata=self._event_metadata(),
)
else:
self._run_in_session(
session,
self.relay.llm.call_end,
model_call.handle,
{**model_call.fields, "outcome": outcome},
metadata=self._event_metadata(),
)
except Exception:
logger.warning(
"Hermes shared-metrics model call close failed", exc_info=True
@ -248,9 +513,52 @@ class _Runtime:
for request_id in request_ids:
self._finish_model_call(session, request_id, outcome)
def _finish_task(
self,
session: _MetricsSession,
task_id: str,
event: dict[str, Any],
) -> None:
task = session.tasks.get(task_id)
if task is None:
return
self._end_pending_model_calls(session, {**event, "task_id": task_id})
fields = task_terminal_fields(
{**task.start_fields, **event},
duration_ms=max(0, (monotonic_ns() - task.started_ns) // 1_000_000),
model_call_count=len(task.model_call_ids),
tool_call_count=len(task.tool_call_ids) + task.unidentified_tool_calls,
retry_count=task.retry_count,
)
try:
self._run_in_task(
task,
self.relay.scope.pop,
task.handle,
output=fields,
metadata=self._event_metadata(),
)
except Exception:
logger.warning("Hermes shared-metrics task close failed", exc_info=True)
finally:
session.tasks.pop(task_id, None)
with self._task_sessions_lock:
task_key = (session.session_id, task_id)
if self._task_sessions.get(task_key) is session:
self._task_sessions.pop(task_key, None)
for turn_id in task.turn_ids:
if self._turn_sessions.get(turn_id) is session:
self._turn_sessions.pop(turn_id, None)
def _export(self) -> None:
self._safe(self.subscriber.store.create_and_export_package)
def _event_metadata(self) -> dict[str, str]:
return {
SCHEMA_KEY: SCHEMA_VERSION,
relay_runtime.RUNTIME_INSTANCE_KEY: self.host.runtime_id,
}
@staticmethod
def _safe(callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
try:
@ -260,23 +568,32 @@ class _Runtime:
return None
@lru_cache(maxsize=1)
def enabled() -> bool:
"""Return the process-lifetime Hermes shared-metrics policy."""
"""Return the shared-metrics policy for the active Hermes profile."""
profile_key = relay_runtime.current_profile_key()
try:
from hermes_cli.config import load_config_readonly
config = load_config_readonly() or {}
except Exception:
logger.debug("Unable to read Hermes shared-metrics policy", exc_info=True)
return False
if not isinstance(config, dict):
return False
telemetry = config.get("telemetry")
if not isinstance(telemetry, dict):
return False
shared_metrics = telemetry.get("shared_metrics")
return isinstance(shared_metrics, dict) and shared_metrics.get("enabled") is True
value = False
else:
telemetry = config.get("telemetry") if isinstance(config, dict) else None
shared_metrics = (
telemetry.get("shared_metrics") if isinstance(telemetry, dict) else None
)
value = (
isinstance(shared_metrics, dict)
and shared_metrics.get("enabled") is True
)
if value:
return True
with _RUNTIME_LOCK:
runtime = _RUNTIMES.pop(profile_key, None)
if isinstance(runtime, _Runtime):
runtime.deactivate()
return False
def handles_hook(hook_name: str) -> bool:
@ -293,15 +610,23 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None:
try:
if hook_name == "on_session_start":
runtime.ensure_session(kwargs)
elif hook_name == "pre_llm_call":
runtime.start_task(kwargs)
elif hook_name == "pre_api_request":
runtime.start_model_call(kwargs)
elif hook_name == "post_tool_call":
runtime.record_tool_call(kwargs)
elif hook_name == "post_api_request":
runtime.end_model_call(kwargs, "success")
elif hook_name == "api_request_error":
if kwargs.get("retryable") is False:
runtime.end_model_call(kwargs, "failed")
elif hook_name == "on_session_end":
runtime.end_pending_model_calls(kwargs)
runtime.finish_task(kwargs)
elif hook_name == "subagent_stop":
child_session_id = str(kwargs.get("child_session_id") or "")
if child_session_id:
runtime.close_session({"session_id": child_session_id})
elif hook_name in {"on_session_finalize", "on_session_reset"}:
runtime.close_session(kwargs)
except Exception:
@ -313,30 +638,110 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None:
def prepare_session_start() -> None:
"""Register the subscriber before any producer opens the session scope."""
if enabled():
_get_runtime()
_get_runtime(retry_failed=True)
def _get_runtime() -> _Runtime | None:
global _RUNTIME
def start_task_run(
*,
session_id: str,
task_id: str,
platform: str,
parent_session_id: str = "",
) -> None:
"""Start task metrics at the outer Hermes execution boundary."""
if not enabled():
return
runtime = _get_runtime(retry_failed=True)
if runtime is None:
return
runtime._safe(
runtime.start_task,
{
"session_id": session_id,
"task_id": task_id,
"platform": platform,
"parent_session_id": parent_session_id,
},
)
def finish_task_run(
*,
session_id: str,
task_id: str,
platform: str,
result: dict[str, Any] | None = None,
error: BaseException | None = None,
) -> None:
"""Finish task metrics for every return or exception path."""
if not enabled():
return
runtime = _get_runtime()
if runtime is None:
return
terminal = result if isinstance(result, dict) else {}
interrupted = terminal.get("interrupted") is True
completed = terminal.get("completed") is True
failed = terminal.get("failed") is True
reason = str(
terminal.get("turn_exit_reason") or terminal.get("failure_reason") or ""
)
if error is not None:
interrupted = isinstance(error, (KeyboardInterrupt, InterruptedError)) or (
type(error).__name__ == "CancelledError"
)
timed_out = isinstance(error, TimeoutError)
completed = False
failed = not interrupted
if interrupted:
reason = "interrupted_by_user"
elif timed_out:
reason = "timed_out"
else:
reason = "system_aborted"
elif not reason:
reason = "failed" if failed else "unknown"
runtime._safe(
runtime.finish_task,
{
"session_id": session_id,
"task_id": task_id,
"platform": platform,
"completed": completed,
"failed": failed,
"interrupted": interrupted,
"turn_exit_reason": reason,
},
)
def _get_runtime(*, retry_failed: bool = False) -> _Runtime | None:
profile_key = relay_runtime.current_profile_key()
with _RUNTIME_LOCK:
if isinstance(_RUNTIME, _Runtime):
return _RUNTIME
if _RUNTIME is _RUNTIME_FAILED:
runtime = _RUNTIMES.get(profile_key)
if isinstance(runtime, _Runtime):
return runtime
if runtime is _RUNTIME_FAILED and not retry_failed:
return None
if runtime is _RUNTIME_FAILED:
_RUNTIMES.pop(profile_key, None)
try:
_RUNTIME = _Runtime()
runtime = _Runtime()
except Exception:
logger.warning("Hermes shared metrics initialization failed", exc_info=True)
_RUNTIME = _RUNTIME_FAILED
_RUNTIMES[profile_key] = _RUNTIME_FAILED
return None
return _RUNTIME
_RUNTIMES[profile_key] = runtime
return runtime
def _reset_for_tests() -> None:
"""Reset process-global state for isolated tests."""
global _RUNTIME
"""Reset all profile-scoped shared-metrics state for isolated tests."""
with _RUNTIME_LOCK:
if isinstance(_RUNTIME, _Runtime):
_RUNTIME.shutdown()
_RUNTIME = None
enabled.cache_clear()
runtimes = list(_RUNTIMES.values())
_RUNTIMES.clear()
for runtime in runtimes:
if isinstance(runtime, _Runtime):
runtime.shutdown()

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

@ -54,7 +54,17 @@
"type": "array",
"minItems": 1,
"items": {
"$ref": "#/$defs/model_call_counter"
"oneOf": [
{
"$ref": "#/$defs/model_call_counter"
},
{
"$ref": "#/$defs/task_started_counter"
},
{
"$ref": "#/$defs/task_finished_counter"
}
]
}
}
},
@ -148,6 +158,180 @@
"minimum": 1
}
}
},
"task_started_counter": {
"type": "object",
"additionalProperties": false,
"required": [
"name",
"type",
"dimensions",
"value"
],
"properties": {
"name": {
"const": "hermes.task_run.started"
},
"type": {
"const": "counter"
},
"dimensions": {
"type": "object",
"additionalProperties": false,
"required": [
"entrypoint",
"execution_surface"
],
"properties": {
"entrypoint": {
"$ref": "#/$defs/task_entrypoint"
},
"execution_surface": {
"$ref": "#/$defs/execution_surface"
}
}
},
"value": {
"type": "integer",
"minimum": 1
}
}
},
"task_finished_counter": {
"type": "object",
"additionalProperties": false,
"required": [
"name",
"type",
"dimensions",
"value"
],
"properties": {
"name": {
"const": "hermes.task_run.finished"
},
"type": {
"const": "counter"
},
"dimensions": {
"type": "object",
"additionalProperties": false,
"required": [
"duration_bucket",
"end_reason",
"entrypoint",
"execution_surface",
"model_call_count_bucket",
"outcome",
"retry_count_bucket",
"termination",
"tool_call_count_bucket"
],
"properties": {
"duration_bucket": {
"$ref": "#/$defs/duration_bucket"
},
"end_reason": {
"enum": [
"approval_denied",
"completed",
"failed",
"guardrail_blocked",
"iteration_limit",
"system_aborted",
"timed_out",
"unknown",
"user_cancelled"
]
},
"entrypoint": {
"$ref": "#/$defs/task_entrypoint"
},
"execution_surface": {
"$ref": "#/$defs/execution_surface"
},
"model_call_count_bucket": {
"$ref": "#/$defs/count_bucket"
},
"outcome": {
"enum": [
"cancelled",
"failed",
"success",
"timed_out",
"unknown"
]
},
"retry_count_bucket": {
"$ref": "#/$defs/count_bucket"
},
"termination": {
"enum": [
"none",
"system_aborted",
"timed_out",
"unknown",
"user_cancelled"
]
},
"tool_call_count_bucket": {
"$ref": "#/$defs/count_bucket"
}
}
},
"value": {
"type": "integer",
"minimum": 1
}
}
},
"execution_surface": {
"enum": [
"api",
"batch",
"cli",
"desktop",
"gateway",
"other",
"python",
"scheduled_task",
"tui",
"unknown"
]
},
"task_entrypoint": {
"enum": [
"api",
"background",
"batch",
"delegated",
"gateway_message",
"interactive",
"other",
"python",
"scheduled_task",
"unknown"
]
},
"duration_bucket": {
"enum": [
"1s_to_5s",
"2m_to_10m",
"30s_to_2m",
"5s_to_30s",
"gte_10m",
"lt_1s"
]
},
"count_bucket": {
"enum": [
"0",
"1",
"2",
"3_to_5",
"6_to_10",
"gte_11"
]
}
}
}

22
hermes_cli/observability/shared_metrics.py
View File

@ -18,6 +18,13 @@ from utils import atomic_json_write
_PACKAGE_SCHEMA_VERSION = "hermes.shared_metrics.v1"
_MODEL_CALL_METRIC = "hermes.model_call.count"
_TASK_STARTED_METRIC = "hermes.task_run.started"
_TASK_FINISHED_METRIC = "hermes.task_run.finished"
_COUNTER_METRICS = frozenset({
_MODEL_CALL_METRIC,
_TASK_FINISHED_METRIC,
_TASK_STARTED_METRIC,
})
_STORE_SCHEMA_VERSION = "1"
_BUSY_TIMEOUT_MS = 250
@ -31,7 +38,7 @@ def _isoformat(value: datetime) -> str:
class SharedMetricsStore:
"""Persist model-call counters and export immutable delta packages."""
"""Persist allowlisted counters and export immutable delta packages."""
def __init__(
self,
@ -51,6 +58,17 @@ class SharedMetricsStore:
hermes_version: str,
) -> None:
"""Increment the terminal model-call counter for the current UTC day."""
self.record_counter(_MODEL_CALL_METRIC, dimensions, hermes_version)
def record_counter(
self,
metric_name: str,
dimensions: dict[str, str],
hermes_version: str,
) -> None:
"""Increment one allowlisted counter for the current UTC day."""
if metric_name not in _COUNTER_METRICS:
raise ValueError(f"Unsupported shared metric: {metric_name}")
dimensions_json = json.dumps(
dimensions,
sort_keys=True,
@ -78,7 +96,7 @@ class SharedMetricsStore:
""",
(
period_start,
_MODEL_CALL_METRIC,
metric_name,
hermes_version or "unknown",
dimensions_json,
),

230
hermes_cli/observability/shared_metrics_contract.py
View File

@ -6,9 +6,12 @@ import re
from functools import lru_cache
from typing import Any
from .relay_runtime import RUNTIME_INSTANCE_KEY
SCHEMA_KEY = "hermes.metrics.schema_version"
SCHEMA_VERSION = "hermes.metrics.event.v1"
MODEL_CALL_SCOPE = "hermes.model_call"
TASK_SCOPE = "hermes.task_run"
SUBSCRIBER_NAME = "hermes.nemo_relay.shared_metrics"
PRIMARY_MODEL_CALL_ROLE = "primary"
@ -33,6 +36,59 @@ PROVIDER_FAMILIES: frozenset[str] = frozenset({
})
MODEL_LOCALITIES: frozenset[str] = frozenset({"local", "remote", "unknown"})
MODEL_OUTCOMES: frozenset[str] = frozenset({"cancelled", "failed", "success"})
TASK_OUTCOMES: frozenset[str] = frozenset({
"cancelled",
"failed",
"success",
"timed_out",
"unknown",
})
TASK_END_REASONS: frozenset[str] = frozenset({
"approval_denied",
"completed",
"failed",
"guardrail_blocked",
"iteration_limit",
"system_aborted",
"timed_out",
"unknown",
"user_cancelled",
})
TASK_TERMINATIONS: frozenset[str] = frozenset({
"none",
"system_aborted",
"timed_out",
"unknown",
"user_cancelled",
})
TASK_ENTRYPOINTS: frozenset[str] = frozenset({
"api",
"background",
"batch",
"delegated",
"gateway_message",
"interactive",
"other",
"python",
"scheduled_task",
"unknown",
})
DURATION_BUCKETS: frozenset[str] = frozenset({
"1s_to_5s",
"2m_to_10m",
"30s_to_2m",
"5s_to_30s",
"gte_10m",
"lt_1s",
})
COUNT_BUCKETS: frozenset[str] = frozenset({
"0",
"1",
"2",
"3_to_5",
"6_to_10",
"gte_11",
})
# Shared metrics use an explicit family allowlist rather than raw model IDs or
# dynamically sourced catalog values. The latter would make the exported schema
@ -94,7 +150,7 @@ def model_call_dimensions(event: Any) -> dict[str, str] | None:
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}
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"}:
@ -141,6 +197,75 @@ 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"}:
return None
if (
str(getattr(event, "kind", "") or "") != "scope"
or str(getattr(event, "category", "") or "") != "function"
or str(getattr(event, "name", "") or "") != TASK_SCOPE
):
return None
if getattr(event, "category_profile", None) is not None:
return None
scope_category = str(getattr(event, "scope_category", "") or "")
data = getattr(event, "data", None)
if scope_category == "start":
expected_fields = {"entrypoint", "execution_surface"}
if not isinstance(data, dict) or set(data) != expected_fields:
return None
if (
data.get("entrypoint") not in TASK_ENTRYPOINTS
or data.get("execution_surface") not in EXECUTION_SURFACES
):
return None
return "hermes.task_run.started", {
"entrypoint": data["entrypoint"],
"execution_surface": data["execution_surface"],
}
expected_fields = {
"duration_bucket",
"end_reason",
"entrypoint",
"execution_surface",
"model_call_count_bucket",
"outcome",
"retry_count_bucket",
"termination",
"tool_call_count_bucket",
}
if (
scope_category != "end"
or not isinstance(data, dict)
or set(data) != expected_fields
):
return None
if (
data.get("duration_bucket") not in DURATION_BUCKETS
or data.get("end_reason") not in TASK_END_REASONS
or data.get("entrypoint") not in TASK_ENTRYPOINTS
or data.get("execution_surface") not in EXECUTION_SURFACES
or data.get("model_call_count_bucket") not in COUNT_BUCKETS
or data.get("outcome") not in TASK_OUTCOMES
or data.get("retry_count_bucket") not in COUNT_BUCKETS
or data.get("termination") not in TASK_TERMINATIONS
or data.get("tool_call_count_bucket") not in COUNT_BUCKETS
):
return None
return "hermes.task_run.finished", {
field: data[field] for field in sorted(expected_fields)
}
def execution_surface(kwargs: dict[str, Any]) -> str:
"""Normalize the safe session surface carried by the parent Relay scope."""
value = (
@ -166,6 +291,109 @@ def execution_surface(kwargs: dict[str, Any]) -> str:
return "unknown" if value == "unknown" else "other"
def task_start_fields(kwargs: dict[str, Any]) -> dict[str, str]:
"""Build the bounded fields recorded on a task scope start event."""
surface = execution_surface(kwargs)
return {
"entrypoint": task_entrypoint(kwargs, surface),
"execution_surface": surface,
}
def task_entrypoint(kwargs: dict[str, Any], surface: str | None = None) -> str:
"""Normalize the task dispatch owner without exporting source strings."""
declared = str(kwargs.get("entrypoint") or "").strip().lower()
if declared in TASK_ENTRYPOINTS:
return declared
resolved_surface = surface or execution_surface(kwargs)
if kwargs.get("parent_task_id") or kwargs.get("parent_session_id"):
return "delegated"
return {
"api": "api",
"batch": "batch",
"cli": "interactive",
"desktop": "interactive",
"gateway": "gateway_message",
"python": "python",
"scheduled_task": "scheduled_task",
"tui": "interactive",
"unknown": "unknown",
}.get(resolved_surface, "other")
def task_terminal_fields(
kwargs: dict[str, Any],
*,
duration_ms: int,
model_call_count: int,
tool_call_count: int,
retry_count: int,
) -> dict[str, str]:
"""Build the bounded terminal payload for one task scope."""
start_fields = task_start_fields(kwargs)
outcome, end_reason, termination = task_terminal_state(kwargs)
return {
**start_fields,
"duration_bucket": duration_bucket(duration_ms),
"end_reason": end_reason,
"model_call_count_bucket": count_bucket(model_call_count),
"outcome": outcome,
"retry_count_bucket": count_bucket(retry_count),
"termination": termination,
"tool_call_count_bucket": count_bucket(tool_call_count),
}
def task_terminal_state(kwargs: dict[str, Any]) -> tuple[str, str, str]:
"""Map Hermes terminal state to bounded task outcome dimensions."""
reason = str(kwargs.get("turn_exit_reason") or "").strip().lower()
if kwargs.get("interrupted") or "interrupt" in reason or "cancel" in reason:
return "cancelled", "user_cancelled", "user_cancelled"
if "timeout" in reason or "timed_out" in reason:
return "timed_out", "timed_out", "timed_out"
if "max_iterations" in reason or "budget_exhausted" in reason:
return "failed", "iteration_limit", "system_aborted"
if "approval" in reason and ("denied" in reason or "rejected" in reason):
return "failed", "approval_denied", "none"
if "guardrail" in reason:
return "failed", "guardrail_blocked", "system_aborted"
if reason == "system_aborted":
return "failed", "system_aborted", "system_aborted"
if kwargs.get("completed") is True:
return "success", "completed", "none"
if kwargs.get("failed") is True or (reason and reason != "unknown"):
return "failed", "failed", "none"
return "unknown", "unknown", "unknown"
def duration_bucket(duration_ms: int) -> str:
"""Bucket a non-negative task duration into a fixed low-cardinality range."""
value = max(0, int(duration_ms))
if value < 1_000:
return "lt_1s"
if value < 5_000:
return "1s_to_5s"
if value < 30_000:
return "5s_to_30s"
if value < 120_000:
return "30s_to_2m"
if value < 600_000:
return "2m_to_10m"
return "gte_10m"
def count_bucket(count: int) -> str:
"""Bucket a non-negative per-task count into a fixed range."""
value = max(0, int(count))
if value <= 2:
return str(value)
if value <= 5:
return "3_to_5"
if value <= 10:
return "6_to_10"
return "gte_11"
def provider_family(kwargs: dict[str, Any]) -> str:
"""Map a Hermes provider to a bounded product category."""
raw_provider = str(kwargs.get("provider") or "").strip().lower().replace("_", "-")

57
hermes_cli/observability/shared_metrics_subscriber.py
View File

@ -3,29 +3,64 @@
from __future__ import annotations
import logging
import threading
from typing import Any
from .shared_metrics import SharedMetricsStore
from .shared_metrics_contract import model_call_dimensions
from .shared_metrics_contract import model_call_dimensions, task_counter
from .relay_runtime import RUNTIME_INSTANCE_KEY
logger = logging.getLogger(__name__)
class SharedMetricsSubscriber:
"""Persist validated primary model-call counters from Relay events."""
"""Persist validated Hermes counters from Relay lifecycle events."""
def __init__(self, store: SharedMetricsStore, hermes_version: str) -> None:
def __init__(
self,
store: SharedMetricsStore,
hermes_version: str,
*,
runtime_id: str | None = None,
) -> None:
self.store = store
self._hermes_version = hermes_version or "unknown"
self._runtime_id = runtime_id
self._active = True
self._lock = threading.RLock()
def deactivate(self) -> None:
"""Stop accepting events before telemetry is disabled or torn down."""
with self._lock:
self._active = False
def __call__(self, event: Any) -> None:
if self._runtime_id is not None:
metadata = getattr(event, "metadata", None)
if (
not isinstance(metadata, dict)
or metadata.get(RUNTIME_INSTANCE_KEY) != self._runtime_id
):
return
dimensions = model_call_dimensions(event)
metric_name = "hermes.model_call.count"
if dimensions is None:
return
try:
self.store.record_model_call(dimensions, self._hermes_version)
except Exception:
logger.warning(
"Unable to persist the Hermes model-call metric",
exc_info=True,
)
task_metric = task_counter(event)
if task_metric is None:
return
metric_name, dimensions = task_metric
with self._lock:
if not self._active:
return
try:
self.store.record_counter(
metric_name,
dimensions,
self._hermes_version,
)
except Exception:
logger.warning(
"Unable to persist the Hermes shared metric: %s",
metric_name,
exc_info=True,
)

67
plugins/observability/nemo_relay/__init__.py
View File

@ -21,7 +21,7 @@ logger = logging.getLogger(__name__)
_INIT_FAILED = object()
_LOCK = threading.RLock()
_RUNTIME: "_Runtime | object | None" = None
_RUNTIMES: dict[str, "_Runtime | object"] = {}
_RELAY_LLM_SURFACE_BY_API_MODE = {
"anthropic_messages": "anthropic.messages",
"chat_completions": "openai.chat_completions",
@ -81,7 +81,7 @@ class _Runtime:
self.sessions: dict[str, _SessionState] = {}
self.subagent_contexts: dict[str, _SubagentContext] = {}
self.atof_exporter: Any = None
self._atof_subscriber_name = "hermes.nemo_relay.atof"
self._atof_subscriber_name = f"hermes.nemo_relay.atof.{self.host.runtime_id}"
self._plugin_activation: Any = None
self._shutdown_registered = False
self._plugin_config_initialized = self._configure_plugins_toml()
@ -257,7 +257,9 @@ class _Runtime:
model_name=str(kwargs.get("model") or self.settings.atif_model_name),
extra={"source": "hermes-agent", "plugin": "observability/nemo_relay"},
)
state.atif_subscriber_name = f"hermes.nemo_relay.atif.{session_id}"
state.atif_subscriber_name = (
f"hermes.nemo_relay.atif.{self.host.runtime_id}.{session_id}"
)
state.atif_exporter.register(state.atif_subscriber_name)
rich_metadata = _metadata(kwargs)
@ -295,6 +297,22 @@ class _Runtime:
**kwargs,
)
async def run_in_session_async(
self,
state: _SessionState,
callback: Callable[..., Any],
*args: Any,
**kwargs: Any,
) -> Any:
if state.relay_session is None:
raise RuntimeError("Hermes core Relay session is unavailable")
return await self.host.run_in_session_async(
state.relay_session,
callback,
*args,
**kwargs,
)
def export_atif(self, state: _SessionState) -> None:
if not self.settings.atif_enabled or state.atif_exporter is None:
return
@ -406,6 +424,7 @@ class _Runtime:
self.host.unregister_subagent(kwargs)
child_session_id = _child_session_id(kwargs)
if child_session_id:
self.close_session({"session_id": child_session_id})
self.subagent_contexts.pop(child_session_id, None)
self.mark("hermes.subagent.stop", kwargs)
@ -482,7 +501,7 @@ class _Runtime:
def _make_managed(impl: Callable[[Any], Any]) -> Any:
async def _managed_execute() -> Any:
result = self.run_in_session(
return await self.run_in_session_async(
state,
self.nemo_relay.llm.execute,
_relay_llm_surface(kwargs),
@ -500,9 +519,6 @@ class _Runtime:
metadata=_metadata(kwargs),
model_name=str(kwargs.get("model") or ""),
)
if inspect.isawaitable(result):
return await result
return result
return _managed_execute()
@ -519,11 +535,16 @@ class _Runtime:
return args
def _normalize(next_args: Any) -> Any:
return next_args if isinstance(next_args, dict) else args
normalized = next_args if isinstance(next_args, dict) else args
if not _json_semantically_equal(normalized, args):
raise RuntimeError(
"NeMo Relay changed tool arguments after Hermes authorization"
)
return args
def _make_managed(impl: Callable[[Any], Any]) -> Any:
async def _managed_execute() -> Any:
result = self.run_in_session(
return await self.run_in_session_async(
state,
self.nemo_relay.tools.execute,
tool_name,
@ -540,9 +561,6 @@ class _Runtime:
),
metadata=_metadata(kwargs),
)
if inspect.isawaitable(result):
return await result
return result
return _managed_execute()
@ -784,26 +802,28 @@ def on_tool_execution_middleware(**kwargs: Any) -> Any:
def _get_runtime() -> Optional[_Runtime]:
global _RUNTIME
profile_key = relay_runtime.current_profile_key()
with _LOCK:
if _RUNTIME is _INIT_FAILED:
runtime = _RUNTIMES.get(profile_key)
if runtime is _INIT_FAILED:
return None
if isinstance(_RUNTIME, _Runtime):
return _RUNTIME
if isinstance(runtime, _Runtime):
return runtime
try:
host = relay_runtime.get_runtime()
if host is None:
raise RuntimeError("Hermes core Relay runtime is unavailable")
_RUNTIME = _Runtime(
runtime = _Runtime(
nemo_relay=host.relay,
settings=_load_settings(),
host=host,
)
except Exception as exc:
logger.debug("NeMo Relay plugin disabled: init failed: %s", exc, exc_info=True)
_RUNTIME = _INIT_FAILED
_RUNTIMES[profile_key] = _INIT_FAILED
return None
return _RUNTIME
_RUNTIMES[profile_key] = runtime
return runtime
def _load_settings() -> _Settings:
@ -1278,8 +1298,9 @@ def _resolve_awaitable(value: Any) -> Any:
def reset_for_tests() -> None:
global _RUNTIME
with _LOCK:
if isinstance(_RUNTIME, _Runtime):
_RUNTIME.shutdown()
_RUNTIME = None
runtimes = list(_RUNTIMES.values())
_RUNTIMES.clear()
for runtime in runtimes:
if isinstance(runtime, _Runtime):
runtime.shutdown()

24
run_agent.py
View File

@ -6347,6 +6347,20 @@ class AIAgent:
reset_conversation_context,
set_conversation_context,
)
from hermes_cli.observability.relay_shared_metrics import (
finish_task_run,
start_task_run,
)
effective_task_id = task_id or str(uuid.uuid4())
task_context = {
"session_id": self.session_id or "",
"task_id": effective_task_id,
"platform": getattr(self, "platform", None) or "",
}
start_task_run(
**task_context,
parent_session_id=getattr(self, "_parent_session_id", None) or "",
)
# Publish the conversation id for ambient Nous Portal tagging. Every
# LLM call made inside this turn — main loop, compression, vision,
# web_extract, session_search, MoA slots, background-review forks
@ -6368,17 +6382,23 @@ class AIAgent:
# which may be observed from another thread.
with scoped_runtime_main({}):
try:
return run_conversation(
result = run_conversation(
self,
user_message,
system_message,
conversation_history,
task_id,
effective_task_id,
stream_callback,
persist_user_message,
persist_user_timestamp=persist_user_timestamp,
moa_config=moa_config,
)
except BaseException as exc:
finish_task_run(**task_context, error=exc)
raise
else:
finish_task_run(**task_context, result=result)
return result
finally:
reset_accounting_context(acct_token)
reset_conversation_context(token)

110
scripts/smoke_nemo_relay_shared_metrics.py
View File

@ -202,24 +202,62 @@ def _validate_store(database_path: Path) -> list[dict[str, Any]]:
}
for name, dimensions, value, packaged_value in rows
]
expected = [
{
"name": "hermes.model_call.count",
"dimensions": {
"call_role": "primary",
"locality": "local",
"model_family": "gpt",
"outcome": "success",
"provider_family": "custom",
},
"value": 1,
"packaged_value": 1,
}
]
if counters != expected:
by_name = {counter["name"]: counter for counter in counters}
if set(by_name) != {
"hermes.model_call.count",
"hermes.task_run.finished",
"hermes.task_run.started",
}:
raise AssertionError(
f"Unexpected SQLite counters:\n{json.dumps(counters, indent=2)}"
)
expected_model = {
"name": "hermes.model_call.count",
"dimensions": {
"call_role": "primary",
"locality": "local",
"model_family": "gpt",
"outcome": "success",
"provider_family": "custom",
},
"value": 1,
"packaged_value": 1,
}
if by_name["hermes.model_call.count"] != expected_model:
raise AssertionError(
f"Unexpected model counter: {by_name['hermes.model_call.count']}"
)
expected_start = {
"name": "hermes.task_run.started",
"dimensions": {
"entrypoint": "interactive",
"execution_surface": "cli",
},
"value": 1,
"packaged_value": 1,
}
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"]
expected_terminal_dimensions = {
"duration_bucket": terminal["dimensions"].get("duration_bucket"),
"end_reason": "completed",
"entrypoint": "interactive",
"execution_surface": "cli",
"model_call_count_bucket": "1",
"outcome": "success",
"retry_count_bucket": "0",
"termination": "none",
"tool_call_count_bucket": "0",
}
if (
terminal["dimensions"] != expected_terminal_dimensions
or terminal["value"] != 1
or terminal["packaged_value"] != 1
):
raise AssertionError(f"Unexpected task terminal counter: {terminal}")
return counters
@ -244,22 +282,38 @@ def _validate_package(outbox: Path, schema_path: Path) -> tuple[Path, dict[str,
raise AssertionError(
f"Exported package leaked prohibited value: {prohibited!r}"
)
expected_metric = {
"name": "hermes.model_call.count",
"type": "counter",
"dimensions": {
"call_role": "primary",
"locality": "local",
"model_family": "gpt",
"outcome": "success",
"provider_family": "custom",
},
"value": 1,
}
if package.get("metrics") != [expected_metric]:
metrics = {metric["name"]: metric for metric in package.get("metrics", [])}
if set(metrics) != {
"hermes.model_call.count",
"hermes.task_run.finished",
"hermes.task_run.started",
}:
raise AssertionError(
f"Unexpected package metrics:\n{json.dumps(package.get('metrics'), indent=2)}"
)
if metrics["hermes.model_call.count"]["dimensions"] != {
"call_role": "primary",
"locality": "local",
"model_family": "gpt",
"outcome": "success",
"provider_family": "custom",
}:
raise AssertionError(
f"Unexpected model metric: {metrics['hermes.model_call.count']}"
)
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",
"outcome": "success",
"retry_count_bucket": "0",
"termination": "none",
"tool_call_count_bucket": "0",
}:
raise AssertionError(f"Unexpected task terminal metric: {terminal}")
return package_path, package

238
tests/hermes_cli/test_relay_shared_metrics.py
View File

@ -7,6 +7,7 @@ import multiprocessing as mp
import os
import sqlite3
import stat
import threading
import uuid
from concurrent.futures import ThreadPoolExecutor
from copy import deepcopy
@ -17,17 +18,30 @@ from typing import Any
import pytest
from hermes_cli.observability.shared_metrics import SharedMetricsStore
from hermes_cli.observability.shared_metrics_contract import (
COUNT_BUCKETS,
DURATION_BUCKETS,
EXECUTION_SURFACES,
MODEL_FAMILIES,
MODEL_LOCALITIES,
MODEL_OUTCOMES,
PRIMARY_MODEL_CALL_ROLE,
PROVIDER_FAMILIES,
TASK_END_REASONS,
TASK_ENTRYPOINTS,
TASK_OUTCOMES,
TASK_TERMINATIONS,
count_bucket,
duration_bucket,
execution_surface,
model_call_outcome,
model_call_dimensions,
model_family,
model_locality,
provider_family,
task_counter,
task_start_fields,
task_terminal_fields,
task_terminal_state,
)
@ -55,6 +69,11 @@ def _package_dimension_schema() -> dict[str, object]:
return schema["$defs"]["model_call_counter"]["properties"]["dimensions"]
def _task_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 {
"call_role": PRIMARY_MODEL_CALL_ROLE,
@ -131,6 +150,21 @@ def test_package_schema_matches_the_model_call_contract():
assert set(properties["provider_family"]["enum"]) == PROVIDER_FAMILIES
def test_package_schema_matches_the_task_contract():
schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
start = _task_dimension_schema("task_started_counter")["properties"]
terminal = _task_dimension_schema("task_finished_counter")["properties"]
assert set(schema["$defs"]["execution_surface"]["enum"]) == EXECUTION_SURFACES
assert set(schema["$defs"]["task_entrypoint"]["enum"]) == TASK_ENTRYPOINTS
assert set(schema["$defs"]["duration_bucket"]["enum"]) == DURATION_BUCKETS
assert set(schema["$defs"]["count_bucket"]["enum"]) == COUNT_BUCKETS
assert start["entrypoint"] == {"$ref": "#/$defs/task_entrypoint"}
assert set(terminal["end_reason"]["enum"]) == TASK_END_REASONS
assert set(terminal["outcome"]["enum"]) == TASK_OUTCOMES
assert set(terminal["termination"]["enum"]) == TASK_TERMINATIONS
@pytest.mark.parametrize(
("provider", "expected"),
[
@ -231,6 +265,105 @@ def test_execution_surface_uses_the_hermes_platform_registry(platform, expected)
assert execution_surface({"platform": platform}) == expected
@pytest.mark.parametrize(
("platform", "expected"),
[
("cli", "interactive"),
("tui", "interactive"),
("whatsapp_cloud", "gateway_message"),
("cron", "scheduled_task"),
("api_server", "api"),
("private-surface", "other"),
],
)
def test_task_start_fields_use_bounded_surface_and_entrypoint(platform, expected):
fields = task_start_fields({"platform": platform})
assert fields["entrypoint"] == expected
assert fields["execution_surface"] in EXECUTION_SURFACES
def test_task_start_fields_identify_delegated_work_without_exporting_parent_id():
fields = task_start_fields({
"platform": "cli",
"parent_session_id": "private-parent-session",
})
assert fields == {
"entrypoint": "delegated",
"execution_surface": "cli",
}
assert "private-parent-session" not in json.dumps(fields)
@pytest.mark.parametrize(
("duration_ms", "expected"),
[
(0, "lt_1s"),
(999, "lt_1s"),
(1_000, "1s_to_5s"),
(5_000, "5s_to_30s"),
(30_000, "30s_to_2m"),
(120_000, "2m_to_10m"),
(600_000, "gte_10m"),
],
)
def test_duration_bucket_boundaries(duration_ms, expected):
assert duration_bucket(duration_ms) == expected
@pytest.mark.parametrize(
("count", "expected"),
[
(0, "0"),
(1, "1"),
(2, "2"),
(3, "3_to_5"),
(6, "6_to_10"),
(11, "gte_11"),
],
)
def test_count_bucket_boundaries(count, expected):
assert count_bucket(count) == expected
@pytest.mark.parametrize(
("event", "expected"),
[
(
{"completed": True, "turn_exit_reason": "text_response(stop)"},
("success", "completed", "none"),
),
(
{"failed": True, "turn_exit_reason": "all_retries_exhausted_no_response"},
("failed", "failed", "none"),
),
(
{"interrupted": True, "turn_exit_reason": "interrupted_by_user"},
("cancelled", "user_cancelled", "user_cancelled"),
),
(
{"turn_exit_reason": "budget_exhausted"},
("failed", "iteration_limit", "system_aborted"),
),
(
{"turn_exit_reason": "guardrail_halt"},
("failed", "guardrail_blocked", "system_aborted"),
),
(
{"failed": True, "turn_exit_reason": "provider_timeout"},
("timed_out", "timed_out", "timed_out"),
),
(
{"failed": True, "turn_exit_reason": "approval_denied"},
("failed", "approval_denied", "none"),
),
],
)
def test_task_terminal_state_is_bounded(event, expected):
assert task_terminal_state(event) == expected
def test_model_outcome_fails_closed_to_a_bounded_value():
assert model_call_outcome({"outcome": "private"}) == "failed"
@ -279,6 +412,52 @@ def test_subscriber_contract_rejects_unknown_fields_and_dimension_values():
assert model_call_dimensions(event) is None
def test_task_subscriber_contract_accepts_only_bounded_scope_events():
start = SimpleNamespace(
kind="scope",
category="function",
category_profile=None,
name="hermes.task_run",
scope_category="start",
metadata={"hermes.metrics.schema_version": "hermes.metrics.event.v1"},
data={"entrypoint": "interactive", "execution_surface": "cli"},
)
assert task_counter(start) == (
"hermes.task_run.started",
{"entrypoint": "interactive", "execution_surface": "cli"},
)
terminal_fields = task_terminal_fields(
{
"platform": "cli",
"completed": True,
"turn_exit_reason": "text_response(stop)",
},
duration_ms=6_000,
model_call_count=2,
tool_call_count=3,
retry_count=1,
)
end = SimpleNamespace(**{
**start.__dict__,
"scope_category": "end",
"data": terminal_fields,
})
assert task_counter(end) == (
"hermes.task_run.finished",
terminal_fields,
)
end.data["task_id"] = "must-not-pass"
assert task_counter(end) is None
end.data.pop("task_id")
end.data["outcome"] = "private"
assert task_counter(end) is None
end.data["outcome"] = "success"
end.metadata["prompt"] = "must-not-pass"
assert task_counter(end) 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:
@ -316,6 +495,36 @@ def test_pending_metrics_keep_the_version_recorded_at_event_time(tmp_path):
assert all(package["metrics"][0]["value"] == 1 for package in packages)
def test_store_exports_task_started_and_terminal_counters(tmp_path):
store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
store.record_counter(
"hermes.task_run.started",
{"entrypoint": "interactive", "execution_surface": "cli"},
"test-version",
)
terminal = task_terminal_fields(
{
"platform": "cli",
"completed": True,
"turn_exit_reason": "text_response(stop)",
},
duration_ms=2_000,
model_call_count=1,
tool_call_count=2,
retry_count=0,
)
store.record_counter("hermes.task_run.finished", terminal, "test-version")
[package_path] = store.create_and_export_package()
package = json.loads(package_path.read_text(encoding="utf-8"))
_schema_validator().validate(package)
assert {metric["name"] for metric in package["metrics"]} == {
"hermes.task_run.finished",
"hermes.task_run.started",
}
def test_package_schema_rejects_unknown_fields(tmp_path):
store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
store.record_model_call(_dimensions(), "test-version")
@ -411,6 +620,35 @@ def test_package_export_does_not_chase_concurrent_updates(tmp_path, monkeypatch)
assert store.counter_snapshot()[0]["packaged_value"] == 2
def test_concurrent_package_builders_commit_one_delta(tmp_path):
database_path = tmp_path / "metrics.sqlite3"
outbox_directory = tmp_path / "outbox"
store = SharedMetricsStore(database_path, outbox_directory)
store.record_model_call(_dimensions(), "test-version")
ready = threading.Barrier(2)
def export() -> list[Path]:
worker_store = SharedMetricsStore(database_path, outbox_directory)
ready.wait(timeout=5)
return worker_store.create_and_export_package()
with ThreadPoolExecutor(max_workers=2) as executor:
futures = [executor.submit(export) for _ in range(2)]
for future in futures:
future.result()
with sqlite3.connect(database_path) as connection:
[outbox_count] = connection.execute(
"SELECT COUNT(*) FROM package_outbox"
).fetchone()
[package_path] = list(outbox_directory.glob("*.json"))
package = json.loads(package_path.read_text(encoding="utf-8"))
assert outbox_count == 1
assert package["metrics"][0]["value"] == 1
assert store.counter_snapshot()[0]["packaged_value"] == 1
def test_concurrent_model_call_updates_are_transactional(tmp_path):
database_path = tmp_path / "metrics.sqlite3"
outbox_directory = tmp_path / "outbox"

802
tests/hermes_cli/test_relay_shared_metrics_runtime.py
View File

@ -3,6 +3,7 @@
from __future__ import annotations
import contextvars
import asyncio
import json
import threading
from pathlib import Path
@ -27,9 +28,10 @@ class _Relay:
self.events: list[tuple[Any, ...]] = []
self._callbacks: dict[str, Any] = {}
self._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")
self.ScopeType = SimpleNamespace(Agent="agent", Function="function")
self.LLMRequest = _Request
self.scope = SimpleNamespace(
push=self._scope_push,
@ -49,10 +51,39 @@ class _Relay:
handle = ("scope", name, self._scope_serial)
self._scope.set(handle)
self.events.append(("scope.push", name, scope_type, kwargs))
if scope_type == self.ScopeType.Function:
self._scope_starts[handle] = kwargs
event = SimpleNamespace(
kind="scope",
category="function",
name=name,
scope_category="start",
category_profile=None,
metadata=kwargs.get("metadata"),
data=kwargs.get("input"),
)
for callback in list(self._callbacks.values()):
callback(event)
return handle
def _scope_pop(self, handle: Any, **kwargs: Any) -> None:
self.events.append(("scope.pop", handle, kwargs))
start = self._scope_starts.pop(handle, None)
if start is not None:
event = SimpleNamespace(
kind="scope",
category="function",
name=handle[1],
scope_category="end",
category_profile=None,
metadata={
**(start.get("metadata") or {}),
**(kwargs.get("metadata") or {}),
},
data=kwargs.get("output"),
)
for callback in list(self._callbacks.values()):
callback(event)
def _scope_event(self, name: str, **kwargs: Any) -> None:
self.events.append(("scope.event", name, kwargs))
@ -139,11 +170,21 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa
assert plugins.has_hook("pre_api_request")
plugins.invoke_hook("on_session_start", **base)
plugins.invoke_hook("pre_llm_call", **base)
plugins.invoke_hook(
"pre_api_request",
**base,
request={"body": {"messages": ["sensitive-prompt"]}},
)
plugins.invoke_hook(
"post_tool_call",
**base,
tool_call_id="sensitive-tool-call",
tool_name="terminal",
args={"command": "sensitive-command"},
result={"output": "sensitive-tool-result"},
status="ok",
)
plugins.invoke_hook(
"api_request_error",
**base,
@ -170,14 +211,30 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa
},
response={"content": "sensitive-response"},
)
plugins.invoke_hook(
"on_session_end",
**base,
completed=True,
failed=False,
interrupted=False,
turn_exit_reason="text_response(stop)",
)
plugins.invoke_hook("on_session_finalize", session_id=base["session_id"])
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"]
session_starts = [
scope_starts = [
event for event in direct_runtime.events if event[0] == "scope.push"
]
assert len(session_starts) == 1
assert len(scope_starts) == 2
assert scope_starts[0][2] == direct_runtime.ScopeType.Agent
assert scope_starts[1][1] == "hermes.task_run"
assert scope_starts[1][2] == direct_runtime.ScopeType.Function
assert scope_starts[1][3]["handle"][1] == relay_runtime.SESSION_SCOPE
assert scope_starts[1][3]["input"] == {
"entrypoint": "interactive",
"execution_surface": "cli",
}
assert len(starts) == 1
assert len(ends) == 1
assert starts[0][2] == {}
@ -193,6 +250,9 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa
assert "sensitive-prompt" not in serialized_events
assert "sensitive-response" not in serialized_events
assert "sensitive-error" not in serialized_events
assert "sensitive-command" not in serialized_events
assert "sensitive-tool-result" not in serialized_events
assert "sensitive-tool-call" not in serialized_events
assert "gpt-sensitive-model-id" not in serialized_events
assert plugins.get_plugin_manager().list_plugins() == []
@ -200,9 +260,44 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa
packages = list((root / "outbox").glob("*.json"))
assert len(packages) == 1
package = json.loads(packages[0].read_text(encoding="utf-8"))
assert package["metrics"][0]["name"] == "hermes.model_call.count"
assert package["metrics"][0]["dimensions"]["model_family"] == "claude"
assert package["metrics"][0]["value"] == 1
metrics = {metric["name"]: metric for metric in package["metrics"]}
assert set(metrics) == {
"hermes.model_call.count",
"hermes.task_run.finished",
"hermes.task_run.started",
}
assert metrics["hermes.model_call.count"]["dimensions"]["model_family"] == "claude"
assert metrics["hermes.model_call.count"]["value"] == 1
assert metrics["hermes.task_run.started"] == {
"name": "hermes.task_run.started",
"type": "counter",
"dimensions": {
"entrypoint": "interactive",
"execution_surface": "cli",
},
"value": 1,
}
terminal = metrics["hermes.task_run.finished"]["dimensions"]
assert terminal["duration_bucket"] in {
"lt_1s",
"1s_to_5s",
"5s_to_30s",
"30s_to_2m",
"2m_to_10m",
"gte_10m",
}
assert {
key: value for key, value in terminal.items() if key != "duration_bucket"
} == {
"end_reason": "completed",
"entrypoint": "interactive",
"execution_surface": "cli",
"model_call_count_bucket": "1",
"outcome": "success",
"retry_count_bucket": "1",
"termination": "none",
"tool_call_count_bucket": "1",
}
def test_direct_runtime_is_disabled_by_default(tmp_path, monkeypatch):
@ -227,6 +322,7 @@ def test_direct_runtime_is_disabled_by_default(tmp_path, monkeypatch):
def test_core_runtime_is_fail_open_without_a_published_binding(monkeypatch, caplog):
relay_shared_metrics._reset_for_tests()
relay_runtime._reset_for_tests()
def missing_relay(name: str):
assert name == "nemo_relay"
raise ModuleNotFoundError(name)
@ -312,6 +408,334 @@ def test_core_runtime_creates_one_session_under_concurrent_access(direct_runtime
)
def test_core_runtime_isolates_same_session_id_by_profile(direct_runtime, tmp_path):
from hermes_constants import (
reset_hermes_home_override,
set_hermes_home_override,
)
profile_a = tmp_path / "profile-a"
profile_b = tmp_path / "profile-b"
token = set_hermes_home_override(profile_a)
try:
runtime_a = relay_runtime.get_runtime()
session_a = runtime_a.ensure_session({"session_id": "shared"})
finally:
reset_hermes_home_override(token)
token = set_hermes_home_override(profile_b)
try:
runtime_b = relay_runtime.get_runtime()
session_b = runtime_b.ensure_session({"session_id": "shared"})
finally:
reset_hermes_home_override(token)
assert runtime_a is not None
assert runtime_b is not None
assert runtime_a is not runtime_b
assert runtime_a.profile_key == str(profile_a.resolve())
assert runtime_b.profile_key == str(profile_b.resolve())
assert session_a is not session_b
assert session_a.handle != session_b.handle
def test_shared_metrics_policy_and_store_are_profile_scoped(tmp_path, monkeypatch):
from hermes_constants import (
get_hermes_home,
reset_hermes_home_override,
set_hermes_home_override,
)
fake = _Relay()
profile_a = tmp_path / "profile-a"
profile_b = tmp_path / "profile-b"
monkeypatch.setattr(relay_runtime, "_load_nemo_relay", lambda: fake)
monkeypatch.setattr(
"hermes_cli.config.load_config_readonly",
lambda: {
"telemetry": {
"shared_metrics": {"enabled": get_hermes_home() == profile_a}
}
},
)
relay_shared_metrics._reset_for_tests()
relay_runtime._reset_for_tests()
token = set_hermes_home_override(profile_a)
try:
assert relay_shared_metrics.enabled()
relay_shared_metrics.start_task_run(
session_id="shared",
task_id="task-a",
platform="cli",
)
relay_shared_metrics.finish_task_run(
session_id="shared",
task_id="task-a",
platform="cli",
result={"completed": True},
)
relay_shared_metrics._get_runtime().close_session({"session_id": "shared"})
finally:
reset_hermes_home_override(token)
token = set_hermes_home_override(profile_b)
try:
assert not relay_shared_metrics.enabled()
relay_shared_metrics.start_task_run(
session_id="shared",
task_id="task-b",
platform="cli",
)
finally:
reset_hermes_home_override(token)
assert list((profile_a / "telemetry" / "shared_metrics" / "outbox").glob("*.json"))
assert not (profile_b / "telemetry").exists()
relay_shared_metrics._reset_for_tests()
relay_runtime._reset_for_tests()
def test_shared_metrics_subscribers_isolate_two_enabled_profiles(tmp_path, monkeypatch):
from hermes_constants import (
reset_hermes_home_override,
set_hermes_home_override,
)
fake = _Relay()
profile_a = tmp_path / "profile-a"
profile_b = tmp_path / "profile-b"
monkeypatch.setattr(relay_runtime, "_load_nemo_relay", lambda: fake)
monkeypatch.setattr(
"hermes_cli.config.load_config_readonly",
lambda: {"telemetry": {"shared_metrics": {"enabled": True}}},
)
relay_shared_metrics._reset_for_tests()
relay_runtime._reset_for_tests()
for profile, task_id in ((profile_a, "task-a"), (profile_b, "task-b")):
token = set_hermes_home_override(profile)
try:
relay_shared_metrics.start_task_run(
session_id="shared",
task_id=task_id,
platform="cli",
)
relay_shared_metrics.finish_task_run(
session_id="shared",
task_id=task_id,
platform="cli",
result={"completed": True},
)
relay_shared_metrics._get_runtime().close_session(
{"session_id": "shared"}
)
finally:
reset_hermes_home_override(token)
for profile in (profile_a, profile_b):
packages = list(
(profile / "telemetry" / "shared_metrics" / "outbox").glob("*.json")
)
assert len(packages) == 1
package = json.loads(packages[0].read_text(encoding="utf-8"))
metrics = {metric["name"]: metric for metric in package["metrics"]}
assert metrics["hermes.task_run.started"]["value"] == 1
assert metrics["hermes.task_run.finished"]["value"] == 1
relay_shared_metrics._reset_for_tests()
relay_runtime._reset_for_tests()
def test_shared_metrics_isolates_same_task_id_across_sessions(direct_runtime):
runtime = relay_shared_metrics._get_runtime()
assert runtime is not None
task_a = runtime.start_task({
"session_id": "session-a",
"task_id": "shared-task",
"platform": "cli",
})
task_b = runtime.start_task({
"session_id": "session-b",
"task_id": "shared-task",
"platform": "gateway",
})
assert task_a is not None
assert task_b is not None
assert task_a is not task_b
assert task_a.handle != task_b.handle
runtime.finish_task({
"session_id": "session-a",
"task_id": "shared-task",
"platform": "cli",
"completed": True,
})
runtime.finish_task({
"session_id": "session-b",
"task_id": "shared-task",
"platform": "gateway",
"completed": True,
})
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"
]
assert len(task_starts) == 2
assert len(task_ends) == 2
def test_disabling_shared_metrics_stops_collection_and_shutdown_export(
tmp_path, monkeypatch
):
from hermes_cli.observability.shared_metrics import SharedMetricsStore
fake = _Relay()
profile = tmp_path / "profile"
policy = {"enabled": True}
monkeypatch.setenv("HERMES_HOME", str(profile))
monkeypatch.setattr(relay_runtime, "_load_nemo_relay", lambda: fake)
monkeypatch.setattr(
"hermes_cli.config.load_config_readonly",
lambda: {"telemetry": {"shared_metrics": dict(policy)}},
)
relay_shared_metrics._reset_for_tests()
relay_runtime._reset_for_tests()
relay_shared_metrics.start_task_run(
session_id="session",
task_id="task",
platform="cli",
)
runtime = relay_shared_metrics._get_runtime()
assert runtime is not None
policy["enabled"] = False
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"},
))
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
relay_shared_metrics.finish_task_run(
session_id="session",
task_id="task",
platform="cli",
result={"completed": True},
)
relay_shared_metrics._reset_for_tests()
root = profile / "telemetry" / "shared_metrics"
store = SharedMetricsStore(root / "metrics.sqlite3", root / "outbox")
assert [row["metric_name"] for row in store.counter_snapshot()] == [
"hermes.task_run.started"
]
assert list((root / "outbox").glob("*.json")) == []
relay_runtime._reset_for_tests()
def test_shared_metrics_retries_transient_initialization_failure(
direct_runtime, monkeypatch
):
real_store = relay_shared_metrics.SharedMetricsStore
attempts = 0
def flaky_store():
nonlocal attempts
attempts += 1
if attempts == 1:
raise OSError("transient store failure")
return real_store()
monkeypatch.setattr(relay_shared_metrics, "SharedMetricsStore", flaky_store)
relay_shared_metrics.start_task_run(
session_id="session",
task_id="first",
platform="cli",
)
relay_shared_metrics.start_task_run(
session_id="session",
task_id="second",
platform="cli",
)
assert attempts == 2
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_async_session_runner_awaits_inside_saved_relay_context(direct_runtime):
runtime = relay_runtime.get_runtime()
assert runtime is not None
session = runtime.ensure_session({"session_id": "async-session"})
assert session is not None
async def probe() -> Any:
await asyncio.sleep(0)
return direct_runtime._scope.get()
result = asyncio.run(runtime.run_in_session_async(session, probe))
assert result == session.handle
def test_shared_metrics_creates_one_task_under_concurrent_access(direct_runtime):
runtime = relay_shared_metrics._get_runtime()
assert runtime is not None
ready = threading.Barrier(8)
tasks: list[Any] = []
def start() -> None:
ready.wait(timeout=5)
tasks.append(
runtime.start_task({"session_id": "s1", "task_id": "t1", "platform": "cli"})
)
threads = [threading.Thread(target=start) for _ in range(8)]
for thread in threads:
thread.start()
for thread in threads:
thread.join(timeout=5)
assert all(not thread.is_alive() for thread in threads)
assert len({id(task) for task in tasks}) == 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_core_runtime_parents_subagent_session_without_exposing_ids(
direct_runtime,
):
@ -341,19 +765,39 @@ def test_core_runtime_parents_subagent_session_without_exposing_ids(
assert child_kwargs["handle"] == parent_handle
assert child_kwargs["metadata"] == {
relay_runtime.RUNTIME_SCHEMA_KEY: relay_runtime.RUNTIME_SCHEMA_VERSION,
relay_runtime.RUNTIME_INSTANCE_KEY: runtime.runtime_id,
"nemo_relay_scope_role": "subagent",
}
assert "sensitive-child" not in json.dumps(pushes)
assert "sensitive-subagent" not in json.dumps(pushes)
def test_core_runtime_closes_child_session_on_subagent_stop(direct_runtime):
runtime = relay_runtime.get_runtime()
assert runtime is not None
runtime.register_subagent({
"parent_session_id": "parent",
"child_session_id": "child",
})
child = runtime.ensure_session({"session_id": "child"})
assert child is not None
runtime.unregister_subagent({"child_session_id": "child"})
assert runtime.get_session("child") is None
child_closes = [
event
for event in direct_runtime.events
if event[0] == "scope.pop" and event[1] == child.handle
]
assert len(child_closes) == 1
def test_core_runtime_ignores_self_parenting_subagent_event(direct_runtime):
runtime = relay_runtime.get_runtime()
assert runtime is not None
runtime.register_subagent(
{"parent_session_id": "same", "child_session_id": "same"}
)
runtime.register_subagent({"parent_session_id": "same", "child_session_id": "same"})
session = runtime.ensure_session({"session_id": "same"})
assert session is not None
@ -377,6 +821,342 @@ def test_terminal_model_error_is_counted_as_failed(direct_runtime):
assert end[2]["outcome"] == "failed"
def test_task_terminal_counts_logical_calls_retries_and_unique_tools(direct_runtime):
base = {
"session_id": "s1",
"task_id": "t1",
"api_request_id": "r1",
"platform": "cli",
"provider": "nvidia",
"model": "nvidia/nemotron-3-super-120b-a12b",
}
plugins.invoke_hook("pre_llm_call", **base)
plugins.invoke_hook("pre_api_request", **base)
plugins.invoke_hook("api_request_error", **base, retryable=True)
plugins.invoke_hook("pre_api_request", **base)
plugins.invoke_hook("api_request_error", **base, retryable=True)
plugins.invoke_hook("pre_api_request", **base)
plugins.invoke_hook("api_request_error", **base, retryable=False)
for tool_call_id in ("tool-1", "tool-1", "tool-2"):
plugins.invoke_hook(
"post_tool_call",
**base,
tool_call_id=tool_call_id,
tool_name="terminal",
result={"output": "private"},
status="ok",
)
plugins.invoke_hook(
"on_session_end",
**base,
completed=False,
failed=True,
interrupted=False,
turn_exit_reason="all_retries_exhausted_no_response",
)
plugins.invoke_hook("on_session_finalize", session_id="s1")
model_starts = [event for event in direct_runtime.events if event[0] == "llm.call"]
model_ends = [
event for event in direct_runtime.events if event[0] == "llm.call_end"
]
assert len(model_starts) == 1
assert len(model_ends) == 1
assert model_ends[0][2]["outcome"] == "failed"
[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"] == {
"duration_bucket": task_end[2]["output"]["duration_bucket"],
"end_reason": "failed",
"entrypoint": "interactive",
"execution_surface": "cli",
"model_call_count_bucket": "1",
"outcome": "failed",
"retry_count_bucket": "2",
"termination": "none",
"tool_call_count_bucket": "2",
}
def test_outer_agent_boundary_closes_early_returns_and_exceptions(
direct_runtime,
monkeypatch,
):
from run_agent import AIAgent
agent = SimpleNamespace(
session_id="s1",
platform="cli",
_parent_session_id=None,
_session_db=None,
_conversation_root_id=lambda: "s1",
)
def early_failure(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
return {
"final_response": "private failure detail",
"completed": False,
"failed": True,
"interrupted": False,
}
monkeypatch.setattr(
"agent.conversation_loop.run_conversation",
early_failure,
)
result = AIAgent.run_conversation(agent, "private prompt", task_id="early")
assert result["failed"] is True
def raise_failure(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
raise RuntimeError("private exception detail")
monkeypatch.setattr(
"agent.conversation_loop.run_conversation",
raise_failure,
)
with pytest.raises(RuntimeError, match="private exception detail"):
AIAgent.run_conversation(agent, "private prompt", task_id="exception")
def raise_interrupt(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
raise KeyboardInterrupt
monkeypatch.setattr(
"agent.conversation_loop.run_conversation",
raise_interrupt,
)
with pytest.raises(KeyboardInterrupt):
AIAgent.run_conversation(agent, "private prompt", task_id="cancelled")
def raise_timeout(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
raise TimeoutError("private timeout detail")
monkeypatch.setattr(
"agent.conversation_loop.run_conversation",
raise_timeout,
)
with pytest.raises(TimeoutError, match="private timeout detail"):
AIAgent.run_conversation(agent, "private prompt", task_id="timed-out")
plugins.invoke_hook("on_session_finalize", session_id="s1")
task_ends = [
event[2]["output"]
for event in direct_runtime.events
if event[0] == "scope.pop" and event[1][1] == "hermes.task_run"
]
assert len(task_ends) == 4
assert task_ends[0]["outcome"] == "failed"
assert task_ends[0]["end_reason"] == "failed"
assert task_ends[0]["termination"] == "none"
assert task_ends[1]["outcome"] == "failed"
assert task_ends[1]["end_reason"] == "system_aborted"
assert task_ends[1]["termination"] == "system_aborted"
assert task_ends[2]["outcome"] == "cancelled"
assert task_ends[2]["end_reason"] == "user_cancelled"
assert task_ends[2]["termination"] == "user_cancelled"
assert task_ends[3]["outcome"] == "timed_out"
assert task_ends[3]["end_reason"] == "timed_out"
assert task_ends[3]["termination"] == "timed_out"
serialized = json.dumps(direct_runtime.events)
assert "private prompt" not in serialized
assert "private failure detail" not in serialized
assert "private exception detail" not in serialized
assert "private timeout detail" not in serialized
def test_outer_agent_boundary_preserves_a_returned_timeout_reason(
direct_runtime,
monkeypatch,
):
from run_agent import AIAgent
agent = SimpleNamespace(
session_id="s1",
platform="cli",
_parent_session_id=None,
_session_db=None,
_conversation_root_id=lambda: "s1",
)
monkeypatch.setattr(
"agent.conversation_loop.run_conversation",
lambda *_args, **_kwargs: {
"final_response": "private timeout response",
"completed": False,
"failed": True,
"failure_reason": "timeout",
},
)
AIAgent.run_conversation(agent, "private prompt", task_id="timed-out")
[task_end] = [
event[2]["output"]
for event in direct_runtime.events
if event[0] == "scope.pop" and event[1][1] == "hermes.task_run"
]
assert task_end["outcome"] == "timed_out"
assert task_end["end_reason"] == "timed_out"
assert task_end["termination"] == "timed_out"
serialized = json.dumps(direct_runtime.events)
assert "private prompt" not in serialized
assert "private timeout response" not in serialized
def test_session_finalize_closes_a_pending_task_as_system_aborted(direct_runtime):
plugins.invoke_hook(
"pre_llm_call",
session_id="s1",
task_id="t1",
platform="cli",
)
plugins.invoke_hook("on_session_finalize", session_id="s1")
[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"] == {
"duration_bucket": task_end[2]["output"]["duration_bucket"],
"end_reason": "system_aborted",
"entrypoint": "interactive",
"execution_surface": "cli",
"model_call_count_bucket": "0",
"outcome": "failed",
"retry_count_bucket": "0",
"termination": "system_aborted",
"tool_call_count_bucket": "0",
}
def test_sequential_tasks_in_one_session_aggregate_once_each(direct_runtime, tmp_path):
for task_id in ("t1", "t2"):
plugins.invoke_hook(
"pre_llm_call",
session_id="s1",
task_id=task_id,
platform="cli",
)
plugins.invoke_hook(
"on_session_end",
session_id="s1",
task_id=task_id,
platform="cli",
completed=True,
failed=False,
interrupted=False,
turn_exit_reason="text_response(stop)",
)
plugins.invoke_hook("on_session_finalize", session_id="s1")
outbox = tmp_path / "hermes-home" / "telemetry" / "shared_metrics" / "outbox"
[package_path] = list(outbox.glob("*.json"))
package = json.loads(package_path.read_text(encoding="utf-8"))
metrics = {metric["name"]: metric for metric in package["metrics"]}
assert metrics["hermes.task_run.started"]["value"] == 2
assert metrics["hermes.task_run.finished"]["value"] == 2
def test_task_ownership_survives_session_id_rotation(direct_runtime):
plugins.invoke_hook(
"pre_llm_call",
session_id="before-compression",
task_id="t1",
platform="cli",
)
plugins.invoke_hook(
"pre_api_request",
session_id="after-compression",
task_id="t1",
api_request_id="r1",
platform="cli",
provider="nvidia",
model="nvidia/nemotron-3-super-120b-a12b",
)
plugins.invoke_hook(
"post_api_request",
session_id="after-compression",
task_id="t1",
api_request_id="r1",
platform="cli",
provider="nvidia",
model="nvidia/nemotron-3-super-120b-a12b",
)
plugins.invoke_hook(
"on_session_end",
session_id="after-compression",
task_id="t1",
platform="cli",
completed=True,
failed=False,
interrupted=False,
turn_exit_reason="text_response(stop)",
)
plugins.invoke_hook("on_session_finalize", session_id="before-compression")
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"
]
model_ends = [
event for event in direct_runtime.events if event[0] == "llm.call_end"
]
assert len(task_starts) == 1
assert len(task_ends) == 1
assert len(model_ends) == 1
assert model_ends[0][2]["outcome"] == "success"
assert task_ends[0][2]["output"]["model_call_count_bucket"] == "1"
assert task_ends[0][2]["output"]["outcome"] == "success"
def test_gateway_and_delegated_entrypoints_flow_through_relay(direct_runtime):
tasks = [
{
"session_id": "gateway-session",
"task_id": "gateway-task",
"platform": "whatsapp_cloud",
},
{
"session_id": "child-session",
"task_id": "delegated-task",
"platform": "cli",
"parent_session_id": "private-parent-session",
},
]
for task in tasks:
plugins.invoke_hook("pre_llm_call", **task)
plugins.invoke_hook(
"on_session_end",
**task,
completed=True,
failed=False,
interrupted=False,
turn_exit_reason="text_response(stop)",
)
starts = [
event[3]["input"]
for event in direct_runtime.events
if event[0] == "scope.push" and event[1] == "hermes.task_run"
]
assert starts == [
{"entrypoint": "gateway_message", "execution_surface": "gateway"},
{"entrypoint": "delegated", "execution_surface": "cli"},
]
assert "private-parent-session" not in json.dumps(direct_runtime.events)
def test_persistence_failure_does_not_escape_the_hook(
direct_runtime,
monkeypatch,
@ -388,7 +1168,7 @@ def test_persistence_failure_does_not_escape_the_hook(
def fail_record(*_args: Any, **_kwargs: Any) -> None:
raise OSError("store unavailable")
monkeypatch.setattr(runtime.subscriber.store, "record_model_call", fail_record)
monkeypatch.setattr(runtime.subscriber.store, "record_counter", fail_record)
plugins.invoke_hook(
"pre_api_request",
session_id="s1",
@ -406,7 +1186,7 @@ def test_persistence_failure_does_not_escape_the_hook(
model="gpt-5",
)
assert "Unable to persist the Hermes model-call metric" in caplog.text
assert "Unable to persist the Hermes shared metric" in caplog.text
def test_close_does_not_reopen_a_session_after_scope_start_failure(

105
tests/plugins/test_nemo_relay_plugin.py
View File

@ -34,7 +34,7 @@ class _FakeNemoRelay:
self._scope_context = contextvars.ContextVar(
"fake_nemo_relay_scope", default=None
)
self.ScopeType = SimpleNamespace(Agent="agent")
self.ScopeType = SimpleNamespace(Agent="agent", Function="function")
self.scope = SimpleNamespace(
push=self._scope_push,
pop=self._scope_pop,
@ -49,6 +49,7 @@ class _FakeNemoRelay:
call=self._tool_call,
call_end=self._tool_call_end,
execute=self._tool_execute,
request_intercepts=self._tool_request_intercepts,
)
self.plugin = SimpleNamespace(
initialize=self._plugin_initialize,
@ -126,10 +127,14 @@ class _FakeNemoRelay:
def _tool_execute(self, name, args, func, **kwargs):
self.events.append(("tool.execute.start", name, args, kwargs))
result = func({"intercepted": True, **args})
result = func(args)
self.events.append(("tool.execute.end", name, result, kwargs))
return result
def _tool_request_intercepts(self, name, args):
self.events.append(("tool.request_intercepts", name, args))
return {"intercepted": True, **args}
def _make_atof_exporter(self, config):
return _FakeAtofExporter(self.events, config)
@ -411,8 +416,11 @@ def test_shared_metrics_and_rich_plugin_share_one_core_session(
if item[0] == "scope.push" and item[1] == relay_runtime.SESSION_SCOPE
]
assert len(session_pushes) == 1
register_metrics = fake.events.index(
("subscribers.register", "hermes.nemo_relay.shared_metrics")
register_metrics = next(
index
for index, item in enumerate(fake.events)
if item[0] == "subscribers.register"
and item[1].startswith("hermes.nemo_relay.shared_metrics.")
)
register_atif = next(
index for index, item in enumerate(fake.events) if item[0] == "atif.register"
@ -575,6 +583,15 @@ def test_nemo_relay_plugin_reparents_child_session_scope_for_embedded_atif(monke
assert child_kwargs["metadata"]["parent_session_id"] == "parent-session"
assert runtime.sessions["child-session"].parent_session_id == "parent-session"
plugin.on_subagent_stop(
parent_session_id="parent-session",
child_session_id="child-session",
child_status="completed",
)
assert "child-session" not in runtime.sessions
assert runtime.host.get_session("child-session") is None
def test_nemo_relay_plugin_skips_embedded_child_atif_file_by_default(tmp_path, monkeypatch):
fake = _FakeNemoRelay()
@ -698,10 +715,15 @@ def test_nemo_relay_plugin_activates_and_owns_dynamic_plugins(tmp_path, monkeypa
request={"messages": []},
next_call=lambda request: {"request": request},
)
tool_result = plugin.on_tool_execution_middleware(
tool_args = relay_runtime.apply_tool_request_intercepts(
session_id="s1",
tool_name="fixture-tool",
args={"value": 1},
)
tool_result = plugin.on_tool_execution_middleware(
session_id="s1",
tool_name="fixture-tool",
args=tool_args,
next_call=lambda args: {"args": args},
)
assert llm_result["request"]["intercepted"] is True
@ -900,7 +922,7 @@ def test_nemo_relay_managed_tool_returns_post_interceptor_result(tmp_path, monke
def execute(name, args, func, **kwargs):
fake.events.append(("tool.execute.start", name, args, kwargs))
raw = func({"intercepted": True, **args})
raw = func(args)
result = {"compressed": True, "raw": raw}
fake.events.append(("tool.execute.end", name, result, kwargs))
return result
@ -918,10 +940,66 @@ def test_nemo_relay_managed_tool_returns_post_interceptor_result(tmp_path, monke
assert result == {
"compressed": True,
"raw": {"tool_output": {"intercepted": True, "value": 1}},
"raw": {"tool_output": {"value": 1}},
}
def test_relay_tool_request_rewrite_precedes_hermes_authorization_boundary(
tmp_path,
monkeypatch,
):
from hermes_cli.middleware import apply_tool_request_middleware
fake = _FakeNemoRelay()
plugin = _fresh_plugin(monkeypatch, fake)
_enable_dynamic_plugin(tmp_path, monkeypatch)
plugin.on_session_start(session_id="s1")
result = apply_tool_request_middleware(
"fixture-tool",
{"value": 1},
session_id="s1",
tool_call_id="tool-1",
)
assert result.payload == {"intercepted": True, "value": 1}
assert result.trace[0] == {"source": "nemo_relay"}
def test_managed_tool_refuses_post_authorization_argument_rewrite(
tmp_path,
monkeypatch,
):
fake = _FakeNemoRelay()
def execute(name, args, func, **kwargs):
del name, kwargs
return func({**args, "after_approval": True})
fake.tools.execute = execute
plugin = _fresh_plugin(monkeypatch, fake)
_enable_dynamic_plugin(tmp_path, monkeypatch)
dispatched = False
def next_call(args):
nonlocal dispatched
dispatched = True
return args
with pytest.raises(
RuntimeError,
match="changed tool arguments after Hermes authorization",
):
plugin.on_tool_execution_middleware(
session_id="s1",
tool_name="fixture-tool",
args={"value": 1},
next_call=next_call,
)
assert not dispatched
def test_nemo_relay_plugin_activates_before_registering_managed_middleware(tmp_path, monkeypatch):
fake = _FakeNemoRelay()
plugin = _fresh_plugin(monkeypatch, fake)
@ -1744,6 +1822,11 @@ mode = "observe_only"
seen_args.update(args)
return {"raw": True, "args": args}
approved_args = relay_runtime.apply_tool_request_intercepts(
session_id="s1",
tool_name="terminal",
args={"command": "pwd"},
)
response = plugin.on_tool_execution_middleware(
session_id="s1",
task_id="t1",
@ -1751,7 +1834,7 @@ mode = "observe_only"
api_request_id="api-1",
tool_name="terminal",
tool_call_id="tool-1",
args={"command": "pwd"},
args=approved_args,
next_call=next_call,
)
@ -1768,7 +1851,7 @@ def test_nemo_relay_adaptive_tool_execution_preserves_downstream_error(tmp_path,
def native_like_execute(name, args, func, **kwargs):
fake.events.append(("tool.execute.start", name, args, kwargs))
try:
return func({"intercepted": True, **args})
return func(args)
except Exception as exc:
raise RuntimeError(f"internal error: {type(exc).__name__}: {exc}") from None
@ -1827,7 +1910,7 @@ def test_nemo_relay_adaptive_tool_execution_keeps_wrapped_relay_error_after_down
def translated_execute(name, args, func, **kwargs):
try:
return func({"intercepted": True, **args})
return func(args)
except Exception:
raise relay_error
@ -1859,7 +1942,7 @@ def test_nemo_relay_adaptive_tool_execution_keeps_relay_translated_error(tmp_pat
def translated_execute(name, args, func, **kwargs):
try:
return func({"intercepted": True, **args})
return func(args)
except Exception:
raise relay_error

34
tests/test_model_tools.py
View File

@ -326,6 +326,40 @@ class TestPreToolCallBlocking:
assert "post_tool_call" in hook_calls
assert "transform_tool_result" in hook_calls
def test_relay_rewrite_is_visible_to_pre_tool_authorization(self, monkeypatch):
observed = {}
def rewrite(**kwargs):
assert kwargs["tool_name"] == "read_file"
return {**kwargs["args"], "path": "approved.txt"}
def fake_invoke_hook(hook_name, **kwargs):
if hook_name == "pre_tool_call":
observed["pre_tool_args"] = kwargs["args"]
return []
def dispatch(_name, args, **_kwargs):
observed["dispatch_args"] = args
return json.dumps({"ok": True})
monkeypatch.setattr(
"hermes_cli.observability.relay_runtime.apply_tool_request_intercepts",
rewrite,
)
monkeypatch.setattr("hermes_cli.plugins.invoke_hook", fake_invoke_hook)
monkeypatch.setattr("hermes_cli.plugins.has_hook", lambda name: True)
monkeypatch.setattr("model_tools.registry.dispatch", dispatch)
handle_function_call(
"read_file",
{"path": "original.txt"},
task_id="t1",
session_id="s1",
)
assert observed["pre_tool_args"]["path"] == "approved.txt"
assert observed["dispatch_args"]["path"] == "approved.txt"
def test_run_agent_pattern_fires_pre_tool_call_exactly_once(self, monkeypatch):
"""End-to-end regression for the double-fire bug.

38
tests/tools/test_zombie_process_cleanup.py
View File

@ -317,9 +317,15 @@ class TestGatewayCleanupWiring:
class TestDelegationCleanup:
"""Verify subagent delegation cleans up child agents."""
def test_run_single_child_calls_close(self):
def test_run_single_child_calls_close(self, monkeypatch, tmp_path):
"""_run_single_child finally block should call close() on child."""
from unittest.mock import MagicMock
from hermes_constants import (
get_hermes_home,
reset_hermes_home_override,
set_hermes_home_override,
)
from hermes_cli.observability import relay_runtime
from tools.delegate_tool import _run_single_child
parent = MagicMock()
@ -327,18 +333,34 @@ class TestDelegationCleanup:
parent._active_children_lock = threading.Lock()
child = MagicMock()
child.session_id = "child-session"
child._delegate_saved_tool_names = ["tool1"]
child.run_conversation.side_effect = RuntimeError("test abort")
observed = {}
def run_conversation(**_kwargs):
observed["hermes_home"] = get_hermes_home()
raise RuntimeError("test abort")
child.run_conversation.side_effect = run_conversation
relay_host = MagicMock()
monkeypatch.setattr(relay_runtime, "get_runtime", lambda **kwargs: relay_host)
parent._active_children.append(child)
result = _run_single_child(
task_index=0,
goal="test goal",
child=child,
parent_agent=parent,
)
profile_home = tmp_path / "profile-a"
token = set_hermes_home_override(profile_home)
try:
result = _run_single_child(
task_index=0,
goal="test goal",
child=child,
parent_agent=parent,
)
finally:
reset_hermes_home_override(token)
child.close.assert_called_once()
assert observed["hermes_home"] == profile_home
relay_host.close_session.assert_called_once_with({"session_id": "child-session"})
assert child not in parent._active_children
assert result["status"] == "error"

21
tools/delegate_tool.py
View File

@ -18,6 +18,7 @@ never the child's intermediate tool calls or reasoning.
"""
import enum
import contextvars
import json
import logging
@ -2008,7 +2009,11 @@ def _run_single_child(
stream_callback=_relay_child_text,
)
_child_future = _timeout_executor.submit(_run_with_thread_capture)
_child_context = contextvars.copy_context()
_child_future = _timeout_executor.submit(
_child_context.run,
_run_with_thread_capture,
)
try:
result = _child_future.result(timeout=child_timeout)
except Exception as _timeout_exc:
@ -2399,6 +2404,18 @@ def _run_single_child(
except Exception:
logger.debug("Failed to close child agent after delegation")
# The child owns its Relay scope lifetime. Close it here, on the worker
# that ran the child, before the parent emits its terminal report.
try:
from hermes_cli.observability import relay_runtime
runtime = relay_runtime.get_runtime(create=False)
child_session_id = str(getattr(child, "session_id", "") or "")
if runtime is not None and child_session_id:
runtime.close_session({"session_id": child_session_id})
except Exception:
logger.debug("Failed to close child Relay session after delegation")
def _recover_tasks_from_json_string(
tasks: Any,
@ -2626,7 +2643,9 @@ def delegate_task(
with DaemonThreadPoolExecutor(max_workers=max_children) as executor:
futures = {}
for i, t, child in children:
child_context = contextvars.copy_context()
future = executor.submit(
child_context.run,
_run_single_child,
task_index=i,
goal=t["goal"],