diff --git a/agent/turn_context.py b/agent/turn_context.py index 8807407278c9c..8f703383d6140 100644 --- a/agent/turn_context.py +++ b/agent/turn_context.py @@ -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] = [] diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py index ccbaac9f6190f..d4293bebd523c 100644 --- a/agent/turn_finalizer.py +++ b/agent/turn_finalizer.py @@ -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 "", ) diff --git a/docs/observability/relay-shared-metrics.md b/docs/observability/relay-shared-metrics.md index 6c85c84cda434..3943ee5596e45 100644 --- a/docs/observability/relay-shared-metrics.md +++ b/docs/observability/relay-shared-metrics.md @@ -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. diff --git a/hermes_cli/hooks.py b/hermes_cli/hooks.py index d3f86bd00e802..2ba0857c93c29 100644 --- a/hermes_cli/hooks.py +++ b/hermes_cli/hooks.py @@ -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": { diff --git a/hermes_cli/middleware.py b/hermes_cli/middleware.py index 8795952a2b7de..8045595c35d20 100644 --- a/hermes_cli/middleware.py +++ b/hermes_cli/middleware.py @@ -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, diff --git a/hermes_cli/observability/relay_runtime.py b/hermes_cli/observability/relay_runtime.py index d52b44d07cb1d..b0d78c976bb2e 100644 --- a/hermes_cli/observability/relay_runtime.py +++ b/hermes_cli/observability/relay_runtime.py @@ -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() diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index c8306966f1e7c..ba1b4bb507e1f 100644 --- a/hermes_cli/observability/relay_shared_metrics.py +++ b/hermes_cli/observability/relay_shared_metrics.py @@ -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() diff --git a/hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json b/hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json index f7306597bc679..6c067d0eb8e7d 100644 --- a/hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json +++ b/hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json @@ -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" + ] } } } diff --git a/hermes_cli/observability/shared_metrics.py b/hermes_cli/observability/shared_metrics.py index 4848ba827593d..228b239342007 100644 --- a/hermes_cli/observability/shared_metrics.py +++ b/hermes_cli/observability/shared_metrics.py @@ -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, ), diff --git a/hermes_cli/observability/shared_metrics_contract.py b/hermes_cli/observability/shared_metrics_contract.py index 7e3c5d336e831..41cfc2eaa5cdb 100644 --- a/hermes_cli/observability/shared_metrics_contract.py +++ b/hermes_cli/observability/shared_metrics_contract.py @@ -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("_", "-") diff --git a/hermes_cli/observability/shared_metrics_subscriber.py b/hermes_cli/observability/shared_metrics_subscriber.py index be5aa6b53b0b4..bbd0442a5201b 100644 --- a/hermes_cli/observability/shared_metrics_subscriber.py +++ b/hermes_cli/observability/shared_metrics_subscriber.py @@ -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, + ) diff --git a/plugins/observability/nemo_relay/__init__.py b/plugins/observability/nemo_relay/__init__.py index d539e0eb198a1..f33359d729992 100644 --- a/plugins/observability/nemo_relay/__init__.py +++ b/plugins/observability/nemo_relay/__init__.py @@ -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() diff --git a/run_agent.py b/run_agent.py index 7280e29422605..8a42d08a6f5c9 100644 --- a/run_agent.py +++ b/run_agent.py @@ -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) diff --git a/scripts/smoke_nemo_relay_shared_metrics.py b/scripts/smoke_nemo_relay_shared_metrics.py index 76e0a59538d40..93f4562a99de1 100644 --- a/scripts/smoke_nemo_relay_shared_metrics.py +++ b/scripts/smoke_nemo_relay_shared_metrics.py @@ -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 diff --git a/tests/hermes_cli/test_relay_shared_metrics.py b/tests/hermes_cli/test_relay_shared_metrics.py index a3bc18e8fbf37..2a6ba2b2b902c 100644 --- a/tests/hermes_cli/test_relay_shared_metrics.py +++ b/tests/hermes_cli/test_relay_shared_metrics.py @@ -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" diff --git a/tests/hermes_cli/test_relay_shared_metrics_runtime.py b/tests/hermes_cli/test_relay_shared_metrics_runtime.py index bbd612a4d5c60..bb87cf7d00d9f 100644 --- a/tests/hermes_cli/test_relay_shared_metrics_runtime.py +++ b/tests/hermes_cli/test_relay_shared_metrics_runtime.py @@ -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( diff --git a/tests/plugins/test_nemo_relay_plugin.py b/tests/plugins/test_nemo_relay_plugin.py index c752c2fe6e0bc..e8599b54f1a5f 100644 --- a/tests/plugins/test_nemo_relay_plugin.py +++ b/tests/plugins/test_nemo_relay_plugin.py @@ -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 diff --git a/tests/test_model_tools.py b/tests/test_model_tools.py index 469b8a6921e95..e6b0bfd51f33b 100644 --- a/tests/test_model_tools.py +++ b/tests/test_model_tools.py @@ -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. diff --git a/tests/tools/test_zombie_process_cleanup.py b/tests/tools/test_zombie_process_cleanup.py index a8b745f541a90..6e0d41148d0ca 100644 --- a/tests/tools/test_zombie_process_cleanup.py +++ b/tests/tools/test_zombie_process_cleanup.py @@ -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" diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 12f94a180a8ef..94eab58fb60c2 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -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"],