From 3bd338d2a9e572e85c77fbfff57b5d319d72c83f Mon Sep 17 00:00:00 2001 From: Alex Fournier Date: Sun, 19 Jul 2026 08:54:27 -0400 Subject: [PATCH] feat(observability): add Relay shared metrics pipeline Signed-off-by: Alex Fournier --- plugins/observability/nemo_relay/README.md | 74 +- plugins/observability/nemo_relay/__init__.py | 689 ++++++++++++++---- .../hermes.shared_metrics.v1.schema.json | 153 ++++ .../nemo_relay/shared_metrics.py | 376 ++++++++++ .../nemo_relay/shared_metrics_contract.py | 251 +++++++ .../nemo_relay/shared_metrics_subscriber.py | 31 + pyproject.toml | 3 +- scripts/smoke_nemo_relay_shared_metrics.py | 379 ++++++++++ tests/plugins/test_nemo_relay_plugin.py | 609 ++++++++++++++-- .../plugins/test_nemo_relay_shared_metrics.py | 462 ++++++++++++ uv.lock | 2 +- 11 files changed, 2824 insertions(+), 205 deletions(-) create mode 100644 plugins/observability/nemo_relay/schemas/hermes.shared_metrics.v1.schema.json create mode 100644 plugins/observability/nemo_relay/shared_metrics.py create mode 100644 plugins/observability/nemo_relay/shared_metrics_contract.py create mode 100644 plugins/observability/nemo_relay/shared_metrics_subscriber.py create mode 100644 scripts/smoke_nemo_relay_shared_metrics.py create mode 100644 tests/plugins/test_nemo_relay_shared_metrics.py diff --git a/plugins/observability/nemo_relay/README.md b/plugins/observability/nemo_relay/README.md index a1b90c4da4fa..60db027072e8 100644 --- a/plugins/observability/nemo_relay/README.md +++ b/plugins/observability/nemo_relay/README.md @@ -84,17 +84,79 @@ wheel from this checkout, then install the official NeMo Relay runtime extra: ```bash uv build --wheel python -m pip install --force-reinstall dist/hermes_agent-*.whl -python -m pip install "nemo-relay>=0.5,<1.0" +python -m pip install "nemo-relay>=0.5.0,<0.6.0" hermes plugins enable observability/nemo_relay ``` The plugin fails open when `nemo-relay` is not installed. Install a supported -NeMo Relay 0.x distribution beginning with 0.5: +NeMo Relay 0.5.x distribution: ```bash -pip install "nemo-relay>=0.5,<1.0" +pip install "nemo-relay>=0.5.0,<0.6.0" ``` +## Shared Metrics Proof Mode + +The Phase 1 telemetry proof adds a separately gated metrics-only mode. Enable +the bundled plugin and the mode in the same `config.yaml`: + +```yaml +plugins: + enabled: + - observability/nemo_relay + entries: + observability/nemo_relay: + shared_metrics: + enabled: true +``` + +The proof reads this setting when the plugin runtime initializes. Restart a +long-running Hermes agent or gateway after changing it. Live revocation and +local-state reset controls are follow-on requirements before production +rollout. + +This first vertical slice maps Hermes's existing API-request hooks to one Relay +LLM lifecycle per logical model call and aggregates terminal primary calls into +`hermes.model_call.count`. Retries sharing an API request ID remain one logical +call. The counter uses bounded provider family, model family, locality, and +outcome dimensions; raw model IDs and request or response content are never +included in the metrics event. + +The subscriber stores cumulative counters and a random opaque install ID in +`$HERMES_HOME/telemetry/shared_metrics/metrics.sqlite3`. After Relay's flush +barrier, Hermes commits un-packaged deltas to an immutable outbox record and +atomically writes a `hermes.shared_metrics.v1` JSON package under +`$HERMES_HOME/telemetry/shared_metrics/outbox/`. Repeating export without new +model calls reuses pending package IDs and does not package a count twice. The +proof mode does not perform network I/O. Task, tool, approval, and skill metrics +remain follow-on slices. The closed package contract ships with the plugin at +`schemas/hermes.shared_metrics.v1.schema.json`. + +When shared metrics are enabled without a separate rich-observability +configuration, the plugin does not emit its existing content-bearing turn, +model, tool, approval, or subagent events. ATOF, ATIF, adaptive components, and +dynamic plugins remain separate explicit configuration choices. + +### End-to-End Smoke Test + +The repository includes a deterministic smoke runner that exercises the real +Hermes CLI and native NeMo Relay binding without external model credentials. It +starts a loopback OpenAI-compatible model server, creates an isolated +`HERMES_HOME`, runs one `hermes chat` turn, and validates the resulting SQLite +counter and schema-conformant JSON package: + +```bash +./.venv/bin/python scripts/smoke_nemo_relay_shared_metrics.py \ + --relay-python ../nemo-relay/python +``` + +The NeMo Relay Python binding must already be built. By default, the runner +expects the Hermes checkout as its working directory and looks for a sibling +`nemo-relay` checkout. Use `--hermes-repo`, `--relay-python`, or `--output-dir` +to override those locations. The generated profile, captured CLI output, +SQLite database, and immutable package are retained in the artifact directory +printed after a successful run. + ## Export Configuration The plugin can configure exporters directly from `HERMES_NEMO_RELAY_*` @@ -253,12 +315,12 @@ For the full generic Hermes middleware contract, see ## Canonical Local Examples -The observe-only examples in this section use a supported NeMo Relay 0.x -distribution beginning with 0.5 and a local Ollama model served through the +The observe-only examples in this section use a supported NeMo Relay 0.5.x +distribution and a local Ollama model served through the OpenAI-compatible API. ```bash -pip install "nemo-relay>=0.5,<1.0" +pip install "nemo-relay>=0.5.0,<0.6.0" export HERMES_HOME=/tmp/hermes-nemo-relay-docs/hermes-home mkdir -p "$HERMES_HOME" diff --git a/plugins/observability/nemo_relay/__init__.py b/plugins/observability/nemo_relay/__init__.py index 56c6140ca58c..895320b18830 100644 --- a/plugins/observability/nemo_relay/__init__.py +++ b/plugins/observability/nemo_relay/__init__.py @@ -4,6 +4,7 @@ from __future__ import annotations import atexit import asyncio +import contextvars import inspect import json import logging @@ -12,9 +13,23 @@ import threading import tomllib from collections.abc import Callable from dataclasses import dataclass, field +from functools import wraps from pathlib import Path from typing import Any, Optional +from .shared_metrics import SharedMetricsStore +from .shared_metrics_contract import ( + MODEL_CALL_SCOPE as _SHARED_METRICS_MODEL_CALL_SCOPE, + SCHEMA_KEY as _SHARED_METRICS_SCHEMA_KEY, + SCHEMA_VERSION as _SHARED_METRICS_SCHEMA_VERSION, + SESSION_SCOPE as _SHARED_METRICS_SESSION_SCOPE, + SUBSCRIBER_NAME as _SHARED_METRICS_SUBSCRIBER_NAME, + execution_surface as _execution_surface, + model_call_fields as _model_call_fields, + model_call_outcome as _model_call_outcome, +) +from .shared_metrics_subscriber import SharedMetricsSubscriber + logger = logging.getLogger(__name__) _INIT_FAILED = object() @@ -30,13 +45,27 @@ _RELAY_LLM_SURFACE_BY_API_MODE = { @dataclass class _SessionState: session_id: str + lock: threading.RLock = field(default_factory=threading.RLock, repr=False) + closing: bool = False handle: Any = None + metrics_handle: Any = None + metrics_context: contextvars.Context | None = None atif_exporter: Any = None atif_subscriber_name: str = "" is_embedded_subagent: bool = False parent_session_id: str = "" llm_spans: dict[str, Any] = field(default_factory=dict) tool_spans: dict[str, Any] = field(default_factory=dict) + metrics_model_calls: dict[str, "_SharedMetricsModelCall"] = field( + default_factory=dict + ) + + +@dataclass +class _SharedMetricsModelCall: + handle: Any + task_id: str + fields: dict[str, str] @dataclass @@ -48,6 +77,7 @@ class _SubagentParent: @dataclass class _Settings: + shared_metrics_enabled: bool = False plugins_toml_path: str = "" plugins_config: dict[str, Any] | None = None dynamic_plugins: list[dict[str, Any]] = field(default_factory=list) @@ -66,21 +96,125 @@ class _Settings: atif_model_name: str = "unknown" +def _with_session_state_lock(method: Callable[..., Any]) -> Callable[..., Any]: + """Serialize one session without blocking unrelated sessions.""" + + @wraps(method) + def wrapped( + self: "_Runtime", + event: dict[str, Any], + *args: Any, + **kwargs: Any, + ) -> Any: + with self._state_lock: + state = self.sessions.get(_session_id(event)) + if state is None: + return None + with state.lock: + if state.closing: + return None + return method(self, event, state, *args, **kwargs) + + return wrapped + + +def _with_ensured_session_lock(method: Callable[..., Any]) -> Callable[..., Any]: + """Create a missing session, then serialize work scoped to it.""" + + @wraps(method) + def wrapped( + self: "_Runtime", + event: dict[str, Any], + *args: Any, + **kwargs: Any, + ) -> Any: + state = self.ensure_session(event) + with state.lock: + if state.closing: + return None + return method(self, event, state, *args, **kwargs) + + return wrapped + + class _Runtime: def __init__(self, nemo_relay: Any, settings: _Settings) -> None: self.nemo_relay = nemo_relay self.settings = settings + self._state_lock = threading.RLock() + self._plugin_lifecycle_lock = threading.RLock() self.sessions: dict[str, _SessionState] = {} self.subagent_parents: dict[str, _SubagentParent] = {} self.atof_exporter: Any = None self._atof_subscriber_name = "hermes.nemo_relay.atof" self._plugin_activation: Any = None self._shutdown_registered = False + self.shared_metrics: SharedMetricsSubscriber | None = None + if settings.shared_metrics_enabled: + try: + from hermes_cli import __version__ + + self.shared_metrics = SharedMetricsSubscriber( + SharedMetricsStore(), + __version__, + ) + except Exception: + logger.warning( + "NeMo Relay shared metrics disabled: local store initialization failed", + exc_info=True, + ) + self._shared_metrics_registered = False + self._configure_shared_metrics() self._plugin_config_initialized = self._configure_plugins_toml() self._plugin_config_needs_reinit = False if not self._plugin_config_initialized: self._activate_direct_fallbacks() + def _configure_shared_metrics(self) -> None: + if self.shared_metrics is None: + return + subscribers = getattr(self.nemo_relay, "subscribers", None) + register = getattr(subscribers, "register", None) + if not callable(register): + logger.warning( + "NeMo Relay shared metrics disabled: subscriber registration is unavailable" + ) + self.shared_metrics = None + return + try: + register(_SHARED_METRICS_SUBSCRIBER_NAME, self.shared_metrics) + except Exception as exc: + logger.warning( + "NeMo Relay shared metrics subscriber registration failed: %s", exc + ) + self.shared_metrics = None + return + self._shared_metrics_registered = True + self._ensure_shutdown_registered() + + def export_shared_metrics(self) -> list[Path]: + """Commit and export model-call deltas after Relay subscriber flush.""" + if self.shared_metrics is None: + return [] + try: + return self.shared_metrics.store.create_and_export_package() + except Exception: + logger.warning("Hermes shared-metrics package export failed", exc_info=True) + return [] + + def rich_observability_enabled(self) -> bool: + if not self.settings.shared_metrics_enabled: + return True + return bool( + self.settings.atof_enabled + or self.settings.atif_enabled + or _enabled_component_config( + self.settings.plugins_config, + "observability", + ) + is not None + ) + def _configure_plugins_toml(self) -> bool: if not self.settings.plugins_config: return False @@ -150,7 +284,9 @@ class _Runtime: # before its awaitable resolves, including error results. self._plugin_activation = None self._plugin_config_initialized = False - self._plugin_config_needs_reinit = bool(self.settings.plugins_config) + self._plugin_config_needs_reinit = bool( + self.settings.plugins_config + ) else: failures.append("dynamic plugin activation has no close method") else: @@ -234,71 +370,268 @@ class _Runtime: self.atof_exporter = None def ensure_session(self, kwargs: dict[str, Any]) -> _SessionState: - self._maybe_reinitialize_plugins_toml() - session_id = _session_id(kwargs) - state = self.sessions.get(session_id) - if state is not None: - return state + with self._plugin_lifecycle_lock: + self._maybe_reinitialize_plugins_toml() + with self._state_lock: + session_id = _session_id(kwargs) + state = self.sessions.get(session_id) + if state is not None: + self._ensure_shared_metrics_session(state, kwargs) + return state - state = _SessionState(session_id=session_id) - if self.settings.atif_enabled and not self._plugins_toml_owns_exporter("atif"): - state.atif_exporter = self.nemo_relay.AtifExporter( - session_id, - self.settings.atif_agent_name, - self.settings.atif_agent_version, - model_name=str(kwargs.get("model") or self.settings.atif_model_name), - extra={"source": "hermes-agent", "plugin": "observability/nemo_relay"}, + state = _SessionState(session_id=session_id) + if self.rich_observability_enabled(): + if ( + self.settings.atif_enabled + and not self._plugins_toml_owns_exporter("atif") + ): + state.atif_exporter = self.nemo_relay.AtifExporter( + session_id, + self.settings.atif_agent_name, + self.settings.atif_agent_version, + 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_exporter.register(state.atif_subscriber_name) + + subagent_parent = self.subagent_parents.get(session_id) + metadata = _metadata(kwargs) + parent_handle = None + if subagent_parent is not None: + parent_handle = subagent_parent.parent_handle + metadata = {**metadata, **subagent_parent.metadata} + state.is_embedded_subagent = True + state.parent_session_id = subagent_parent.parent_session_id + + state.handle = self.nemo_relay.scope.push( + f"hermes-session-{session_id}", + self.nemo_relay.ScopeType.Agent, + handle=parent_handle, + data={"session_id": session_id}, + metadata=metadata, + ) + + self._ensure_shared_metrics_session(state, kwargs) + self.sessions[session_id] = state + return state + + def _ensure_shared_metrics_session( + self, + state: _SessionState, + kwargs: dict[str, Any], + ) -> None: + if ( + state.closing + or not self._shared_metrics_registered + or state.metrics_handle is not None + ): + return + metrics_context = contextvars.Context() + try: + state.metrics_handle = metrics_context.run( + self.nemo_relay.scope.push, + _SHARED_METRICS_SESSION_SCOPE, + self.nemo_relay.ScopeType.Agent, + input={"execution_surface": _execution_surface(kwargs)}, + metadata={_SHARED_METRICS_SCHEMA_KEY: _SHARED_METRICS_SCHEMA_VERSION}, ) - state.atif_subscriber_name = f"hermes.nemo_relay.atif.{session_id}" - state.atif_exporter.register(state.atif_subscriber_name) + except Exception: + logger.warning( + "NeMo Relay shared-metrics session start failed", exc_info=True + ) + return + state.metrics_context = metrics_context - subagent_parent = self.subagent_parents.get(session_id) - metadata = _metadata(kwargs) - parent_handle = None - if subagent_parent is not None: - parent_handle = subagent_parent.parent_handle - metadata = {**metadata, **subagent_parent.metadata} - state.is_embedded_subagent = True - state.parent_session_id = subagent_parent.parent_session_id + def _run_in_metrics_context( + self, + state: _SessionState, + callback: Callable[..., Any], + *args: Any, + **kwargs: Any, + ) -> Any: + """Run a native lifecycle call against the isolated metrics stack.""" + if state.metrics_context is None: + raise RuntimeError("shared-metrics scope context is unavailable") - state.handle = self.nemo_relay.scope.push( - f"hermes-session-{session_id}", - self.nemo_relay.ScopeType.Agent, - handle=parent_handle, - data={"session_id": session_id}, - metadata=metadata, + def invoke() -> Any: + # Relay's manual LLM helpers call the native API directly. Re-sync + # this Context's stack before entering them so scope-local policy + # cannot leak in from the rich-observability stack on this thread. + self.nemo_relay.get_scope_stack() + return callback(*args, **kwargs) + + return state.metrics_context.run(invoke) + + @_with_ensured_session_lock + def start_model_call(self, kwargs: dict[str, Any], state: _SessionState) -> None: + if not self._shared_metrics_registered or state.metrics_handle is None: + return + request_id = str(kwargs.get("api_request_id") or "") + if not request_id: + return + fields = _model_call_fields(kwargs) + model_family = fields.pop("model_family") + existing = state.metrics_model_calls.get(request_id) + if existing is not None: + # A logical call can span retries or provider fallback. Attribute its + # terminal provider fields to the most recent attempt without opening + # a second Relay lifecycle. Relay's model_name remains the logical + # model family recorded when the lifecycle started. + existing.fields = fields + return + request = self.nemo_relay.LLMRequest({}, {}) + try: + handle = self._run_in_metrics_context( + state, + self.nemo_relay.llm.call, + _SHARED_METRICS_MODEL_CALL_SCOPE, + request, + handle=state.metrics_handle, + metadata={_SHARED_METRICS_SCHEMA_KEY: _SHARED_METRICS_SCHEMA_VERSION}, + model_name=model_family, + ) + except Exception: + logger.warning( + "NeMo Relay shared-metrics model-call start failed", exc_info=True + ) + return + state.metrics_model_calls[request_id] = _SharedMetricsModelCall( + handle=handle, + task_id=str(kwargs.get("task_id") or ""), + fields=fields, ) - self.sessions[session_id] = state - return state + + def _finish_model_call( + self, + state: _SessionState, + request_id: str, + outcome: str, + ) -> None: + model_call = state.metrics_model_calls.pop(request_id, None) + if model_call is None: + return + try: + self._run_in_metrics_context( + state, + self.nemo_relay.llm.call_end, + model_call.handle, + {**model_call.fields, "outcome": outcome}, + metadata={_SHARED_METRICS_SCHEMA_KEY: _SHARED_METRICS_SCHEMA_VERSION}, + ) + except Exception: + logger.warning( + "NeMo Relay shared-metrics model-call end failed", exc_info=True + ) + + @_with_session_state_lock + def end_model_call(self, kwargs: dict[str, Any], state: _SessionState) -> None: + request_id = str(kwargs.get("api_request_id") or "") + model_call = state.metrics_model_calls.get(request_id) + if model_call is None: + return + fields = _model_call_fields(kwargs) + fields.pop("model_family") + model_call.fields = fields + self._finish_model_call(state, request_id, _model_call_outcome(kwargs)) + + @_with_session_state_lock + def end_pending_model_calls( + self, + kwargs: dict[str, Any], + state: _SessionState, + ) -> None: + self._end_pending_model_calls(state, kwargs) + + def _end_pending_model_calls( + self, + state: _SessionState, + kwargs: dict[str, Any], + ) -> None: + task_id = str(kwargs.get("task_id") or "") + request_ids = [ + request_id + for request_id, model_call in state.metrics_model_calls.items() + if not task_id or model_call.task_id == task_id + ] + outcome = "cancelled" if kwargs.get("interrupted") else "failed" + for request_id in request_ids: + self._finish_model_call(state, request_id, outcome) def export_atif(self, state: _SessionState) -> None: if not self.settings.atif_enabled or state.atif_exporter is None: return - if state.is_embedded_subagent and self.settings.atif_subagent_export_mode != "all": + if ( + state.is_embedded_subagent + and self.settings.atif_subagent_export_mode != "all" + ): return output_dir = self.settings.atif_output_directory if not output_dir: return Path(output_dir).mkdir(parents=True, exist_ok=True) - filename = self.settings.atif_filename_template.format(session_id=state.session_id) - Path(output_dir, filename).write_text(state.atif_exporter.export_json(), encoding="utf-8") + filename = self.settings.atif_filename_template.format( + session_id=state.session_id + ) + Path(output_dir, filename).write_text( + state.atif_exporter.export_json(), encoding="utf-8" + ) + + def _clear_static_plugins_after_last_session(self) -> None: + """Clear static plugin state without holding the session-map lock.""" + with self._plugin_lifecycle_lock: + with self._state_lock: + if self.sessions or self._plugin_activation is not None: + return + should_clear = self._plugin_config_initialized + if not should_clear and self.settings.plugins_config: + self._plugin_config_needs_reinit = True + if should_clear: + self._clear_plugins_toml() def close_session(self, kwargs: dict[str, Any]) -> None: session_id = _session_id(kwargs) - self.subagent_parents.pop(session_id, None) - state = self.sessions.pop(session_id, None) + with self._state_lock: + self.subagent_parents.pop(session_id, None) + state = self.sessions.get(session_id) if state is None: return failures: list[str] = [] - if state.handle is not None: - try: - self.nemo_relay.scope.pop(state.handle, output=_jsonable(kwargs)) - except Exception as exc: - failures.append(f"session scope pop failed: {exc}") + with state.lock: + if state.closing: + return + state.closing = True + self._end_pending_model_calls(state, {"session_id": session_id}) + if state.metrics_handle is not None and state.metrics_context is not None: + try: + self._run_in_metrics_context( + state, + self.nemo_relay.scope.pop, + state.metrics_handle, + output={}, + metadata={ + _SHARED_METRICS_SCHEMA_KEY: _SHARED_METRICS_SCHEMA_VERSION + }, + ) + except Exception as exc: + failures.append(f"shared-metrics session scope pop failed: {exc}") + if state.handle is not None: + try: + self.nemo_relay.scope.pop(state.handle, output=_jsonable(kwargs)) + except Exception as exc: + failures.append(f"session scope pop failed: {exc}") try: _flush_relay_subscribers(self.nemo_relay) except Exception as exc: failures.append(f"subscriber flush failed: {exc}") + self.export_shared_metrics() try: self.export_atif(state) except Exception as exc: @@ -308,21 +641,13 @@ class _Runtime: state.atif_exporter.deregister(state.atif_subscriber_name) except Exception as exc: failures.append(f"ATIF deregister failed: {exc}") - if ( - self._plugin_config_initialized - and self._plugin_activation is None - and not self.sessions - ): - try: - self._clear_plugins_toml() - except Exception as exc: - failures.append(f"plugin configuration clear failed: {exc}") - elif ( - self.settings.plugins_config - and self._plugin_activation is None - and not self.sessions - ): - self._plugin_config_needs_reinit = True + with self._state_lock: + if self.sessions.get(session_id) is state: + self.sessions.pop(session_id, None) + try: + self._clear_static_plugins_after_last_session() + except Exception as exc: + failures.append(f"plugin configuration clear failed: {exc}") if failures: logger.warning( "NeMo Relay session %s teardown completed with errors: %s", @@ -333,17 +658,38 @@ class _Runtime: def shutdown(self) -> None: """Close active sessions and the process-lifetime plugin activation.""" failures: list[str] = [] - for session_id in list(self.sessions): + with self._state_lock: + session_ids = list(self.sessions) + for session_id in session_ids: try: - self.close_session({"session_id": session_id, "reason": "runtime_shutdown"}) + self.close_session({ + "session_id": session_id, + "reason": "runtime_shutdown", + }) except Exception as exc: failures.append(f"session {session_id} close failed: {exc}") - if self._plugin_config_initialized: - try: - self._clear_plugins_toml() - except Exception as exc: - failures.append(f"plugin runtime close failed: {exc}") + with self._plugin_lifecycle_lock: + if self._plugin_config_initialized: + try: + self._clear_plugins_toml() + except Exception as exc: + failures.append(f"plugin runtime close failed: {exc}") self._clear_atof() + if self._shared_metrics_registered: + try: + _flush_relay_subscribers(self.nemo_relay) + except Exception as exc: + failures.append(f"shared-metrics subscriber flush failed: {exc}") + self.export_shared_metrics() + try: + subscribers = getattr(self.nemo_relay, "subscribers", None) + deregister = getattr(subscribers, "deregister", None) + if callable(deregister): + deregister(_SHARED_METRICS_SUBSCRIBER_NAME) + except Exception as exc: + failures.append(f"shared-metrics subscriber deregister failed: {exc}") + finally: + self._shared_metrics_registered = False if self._shutdown_registered and self._plugin_activation is None: atexit.unregister(self.shutdown) self._shutdown_registered = False @@ -388,14 +734,17 @@ class _Runtime: def managed_llm_enabled(self) -> bool: return ( (self.settings.adaptive_enabled or self._plugin_activation is not None) - and callable(getattr(getattr(self.nemo_relay, "llm", None), "execute", None)) + and callable( + getattr(getattr(self.nemo_relay, "llm", None), "execute", None) + ) and callable(getattr(self.nemo_relay, "LLMRequest", None)) ) def managed_tool_enabled(self) -> bool: return ( - (self.settings.adaptive_enabled or self._plugin_activation is not None) - and callable(getattr(getattr(self.nemo_relay, "tools", None), "execute", None)) + self.settings.adaptive_enabled or self._plugin_activation is not None + ) and callable( + getattr(getattr(self.nemo_relay, "tools", None), "execute", None) ) def _run_managed_with_downstream_preservation( @@ -433,7 +782,9 @@ class _Runtime: try: managed_result = _resolve_awaitable(make_managed_execute(_impl)) except Exception as exc: - if downstream_error is not None and _is_relay_wrapped_callback_error(exc, callback_error): + if downstream_error is not None and _is_relay_wrapped_callback_error( + exc, callback_error + ): raise downstream_error raise if ( @@ -463,14 +814,12 @@ class _Runtime: request, impl, handle=state.handle, - data=_jsonable( - { - "turn_id": kwargs.get("turn_id"), - "api_request_id": kwargs.get("api_request_id"), - "api_call_count": kwargs.get("api_call_count"), - "mode": self.settings.adaptive_mode, - } - ), + data=_jsonable({ + "turn_id": kwargs.get("turn_id"), + "api_request_id": kwargs.get("api_request_id"), + "api_call_count": kwargs.get("api_call_count"), + "mode": self.settings.adaptive_mode, + }), metadata=_metadata(kwargs), model_name=str(kwargs.get("model") or ""), ) @@ -481,7 +830,11 @@ class _Runtime: return _managed_execute() return self._run_managed_with_downstream_preservation( - next_call, _normalize, _llm_response_payload, _make_managed, preserve_raw_response=True + next_call, + _normalize, + _llm_response_payload, + _make_managed, + preserve_raw_response=True, ) def execute_tool(self, kwargs: dict[str, Any]) -> Any: @@ -502,14 +855,12 @@ class _Runtime: args, impl, handle=state.handle, - data=_jsonable( - { - "turn_id": kwargs.get("turn_id"), - "api_request_id": kwargs.get("api_request_id"), - "tool_call_id": kwargs.get("tool_call_id"), - "mode": self.settings.adaptive_mode, - } - ), + data=_jsonable({ + "turn_id": kwargs.get("turn_id"), + "api_request_id": kwargs.get("api_request_id"), + "tool_call_id": kwargs.get("tool_call_id"), + "mode": self.settings.adaptive_mode, + }), metadata=_metadata(kwargs), ) if inspect.isawaitable(result): @@ -555,8 +906,16 @@ def on_session_start(**kwargs: Any) -> None: def on_session_end(**kwargs: Any) -> None: runtime = _get_runtime() - if runtime is not None: - _safe(lambda: (runtime.mark("hermes.session.end", kwargs), runtime.export_atif(runtime.ensure_session(kwargs)))) + if runtime is None: + return + _safe(lambda: runtime.end_pending_model_calls(kwargs)) + if runtime.rich_observability_enabled(): + _safe( + lambda: ( + runtime.mark("hermes.session.end", kwargs), + runtime.export_atif(runtime.ensure_session(kwargs)), + ) + ) def on_session_finalize(**kwargs: Any) -> None: @@ -573,13 +932,13 @@ def on_session_reset(**kwargs: Any) -> None: def on_pre_llm_call(**kwargs: Any) -> None: runtime = _get_runtime() - if runtime is not None: + if runtime is not None and runtime.rich_observability_enabled(): _safe(lambda: runtime.mark("hermes.turn.start", kwargs)) def on_post_llm_call(**kwargs: Any) -> None: runtime = _get_runtime() - if runtime is not None: + if runtime is not None and runtime.rich_observability_enabled(): _safe(lambda: runtime.mark("hermes.turn.end", kwargs)) @@ -587,19 +946,27 @@ def on_pre_api_request(**kwargs: Any) -> None: runtime = _get_runtime() if runtime is None: return + _safe(lambda: runtime.start_model_call(kwargs)) + if not runtime.rich_observability_enabled(): + return if runtime.managed_llm_enabled(): return def _record() -> None: state = runtime.ensure_session(kwargs) request_payload = kwargs.get("request") - request_body = request_payload.get("body") if isinstance(request_payload, dict) else {} + request_body = ( + request_payload.get("body") if isinstance(request_payload, dict) else {} + ) request = runtime.nemo_relay.LLMRequest({}, _jsonable(request_body)) span = runtime.nemo_relay.llm.call( str(kwargs.get("provider") or "llm"), request, handle=state.handle, - data=_jsonable({"turn_id": kwargs.get("turn_id"), "api_request_id": kwargs.get("api_request_id")}), + data=_jsonable({ + "turn_id": kwargs.get("turn_id"), + "api_request_id": kwargs.get("api_request_id"), + }), metadata=_metadata(kwargs), model_name=str(kwargs.get("model") or ""), ) @@ -612,6 +979,9 @@ def on_post_api_request(**kwargs: Any) -> None: runtime = _get_runtime() if runtime is None: return + _safe(lambda: runtime.end_model_call({**kwargs, "outcome": "success"})) + if not runtime.rich_observability_enabled(): + return if runtime.managed_llm_enabled(): return @@ -624,7 +994,10 @@ def on_post_api_request(**kwargs: Any) -> None: runtime.nemo_relay.llm.call_end( span, _jsonable(kwargs.get("response") or {}), - data=_jsonable({"usage": kwargs.get("usage"), "finish_reason": kwargs.get("finish_reason")}), + data=_jsonable({ + "usage": kwargs.get("usage"), + "finish_reason": kwargs.get("finish_reason"), + }), metadata=_metadata(kwargs), ) @@ -633,7 +1006,7 @@ def on_post_api_request(**kwargs: Any) -> None: def on_api_request_error(**kwargs: Any) -> None: runtime = _get_runtime() - if runtime is None: + if runtime is None or not runtime.rich_observability_enabled(): return if runtime.managed_llm_enabled(): return @@ -658,6 +1031,8 @@ def on_pre_tool_call(**kwargs: Any) -> None: runtime = _get_runtime() if runtime is None: return + if not runtime.rich_observability_enabled(): + return if runtime.managed_tool_enabled(): return @@ -667,7 +1042,10 @@ def on_pre_tool_call(**kwargs: Any) -> None: str(kwargs.get("tool_name") or "tool"), _jsonable(kwargs.get("args") or {}), handle=state.handle, - data=_jsonable({"turn_id": kwargs.get("turn_id"), "api_request_id": kwargs.get("api_request_id")}), + data=_jsonable({ + "turn_id": kwargs.get("turn_id"), + "api_request_id": kwargs.get("api_request_id"), + }), metadata=_metadata(kwargs), tool_call_id=str(kwargs.get("tool_call_id") or ""), ) @@ -680,6 +1058,8 @@ def on_post_tool_call(**kwargs: Any) -> None: runtime = _get_runtime() if runtime is None: return + if not runtime.rich_observability_enabled(): + return if runtime.managed_tool_enabled(): return @@ -692,7 +1072,10 @@ def on_post_tool_call(**kwargs: Any) -> None: runtime.nemo_relay.tools.call_end( span, _jsonable(kwargs.get("result")), - data=_jsonable({"status": kwargs.get("status"), "duration_ms": kwargs.get("duration_ms")}), + data=_jsonable({ + "status": kwargs.get("status"), + "duration_ms": kwargs.get("duration_ms"), + }), metadata=_metadata(kwargs), ) @@ -701,25 +1084,29 @@ def on_post_tool_call(**kwargs: Any) -> None: def on_pre_approval_request(**kwargs: Any) -> None: runtime = _get_runtime() - if runtime is not None: + if runtime is None: + return + if runtime.rich_observability_enabled(): _safe(lambda: runtime.mark("hermes.approval.request", kwargs)) def on_post_approval_response(**kwargs: Any) -> None: runtime = _get_runtime() - if runtime is not None: + if runtime is None: + return + if runtime.rich_observability_enabled(): _safe(lambda: runtime.mark("hermes.approval.response", kwargs)) def on_subagent_start(**kwargs: Any) -> None: runtime = _get_runtime() - if runtime is not None: + if runtime is not None and runtime.rich_observability_enabled(): _safe(lambda: runtime.mark_subagent_start(kwargs)) def on_subagent_stop(**kwargs: Any) -> None: runtime = _get_runtime() - if runtime is not None: + if runtime is not None and runtime.rich_observability_enabled(): _safe(lambda: runtime.mark_subagent_stop(kwargs)) @@ -761,7 +1148,9 @@ def _get_runtime() -> Optional[_Runtime]: try: _RUNTIME = _Runtime(nemo_relay=nemo_runtime, settings=_load_settings()) except Exception as exc: - logger.debug("NeMo Relay plugin disabled: init failed: %s", exc, exc_info=True) + logger.debug( + "NeMo Relay plugin disabled: init failed: %s", exc, exc_info=True + ) _RUNTIME = _INIT_FAILED return None return _RUNTIME @@ -772,6 +1161,7 @@ def _load_settings() -> _Settings: plugins_config = _load_plugins_config(plugins_toml_path) adaptive_config = _enabled_component_config(plugins_config, "adaptive") return _Settings( + shared_metrics_enabled=_shared_metrics_enabled(), plugins_toml_path=plugins_toml_path, plugins_config=plugins_config, dynamic_plugins=_dynamic_plugin_specs(plugins_config, plugins_toml_path), @@ -783,7 +1173,8 @@ def _load_settings() -> _Settings: atof_mode=_env("HERMES_NEMO_RELAY_ATOF_MODE") or "append", atif_enabled=_env_bool("HERMES_NEMO_RELAY_ATIF_ENABLED"), atif_output_directory=_env("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY"), - atif_filename_template=_env("HERMES_NEMO_RELAY_ATIF_FILENAME_TEMPLATE") or "hermes-atif-{session_id}.json", + atif_filename_template=_env("HERMES_NEMO_RELAY_ATIF_FILENAME_TEMPLATE") + or "hermes-atif-{session_id}.json", atif_subagent_export_mode=_atif_subagent_export_mode(), atif_agent_name=_env("HERMES_NEMO_RELAY_ATIF_AGENT_NAME") or "Hermes Agent", atif_agent_version=_env("HERMES_NEMO_RELAY_ATIF_AGENT_VERSION") or "unknown", @@ -791,6 +1182,31 @@ def _load_settings() -> _Settings: ) +def _shared_metrics_enabled() -> bool: + try: + from hermes_cli.config import load_config_readonly + + config = load_config_readonly() or {} + except Exception: + logger.debug( + "Unable to read Hermes shared-metrics configuration", exc_info=True + ) + return False + if not isinstance(config, dict): + return False + plugins = config.get("plugins") + if not isinstance(plugins, dict): + return False + entries = plugins.get("entries") + if not isinstance(entries, dict): + return False + entry = entries.get("observability/nemo_relay") or entries.get("nemo_relay") + if not isinstance(entry, dict): + return False + shared_metrics = entry.get("shared_metrics") + return isinstance(shared_metrics, dict) and shared_metrics.get("enabled") is True + + def _static_plugin_config(plugins_config: dict[str, Any]) -> dict[str, Any]: """Return Relay's base config without embedding- or gateway-host fields.""" return { @@ -863,13 +1279,15 @@ def _dynamic_plugin_specs( continue if not isinstance(manifest_ref, str) or not manifest_ref.strip(): logger.warning( - "Invalid NeMo Relay dynamic_plugins[%d]: manifest_ref is required", index + "Invalid NeMo Relay dynamic_plugins[%d]: manifest_ref is required", + index, ) invalid = True continue if not isinstance(config, dict): logger.warning( - "Invalid NeMo Relay dynamic_plugins[%d]: config must be an object", index + "Invalid NeMo Relay dynamic_plugins[%d]: config must be an object", + index, ) invalid = True continue @@ -886,7 +1304,9 @@ def _dynamic_plugin_specs( spec: dict[str, Any] = { "plugin_id": plugin_id.strip(), "kind": kind, - "manifest_ref": _config_relative_path(manifest_ref.strip(), plugins_toml_path), + "manifest_ref": _config_relative_path( + manifest_ref.strip(), plugins_toml_path + ), "config": config, } if environment_ref is not None: @@ -908,7 +1328,9 @@ def _config_relative_path(value: str, plugins_toml_path: str) -> str: path = Path(value) if path.is_absolute(): return str(path) - config_path = Path(plugins_toml_path) if plugins_toml_path else Path.cwd() / "plugins.toml" + config_path = ( + Path(plugins_toml_path) if plugins_toml_path else Path.cwd() / "plugins.toml" + ) if not config_path.is_absolute(): config_path = Path.cwd() / config_path return os.path.abspath(config_path.parent / path) @@ -998,7 +1420,9 @@ def _child_session_id(kwargs: dict[str, Any]) -> str: return str(kwargs.get("child_session_id") or "") -def _subagent_child_metadata(kwargs: dict[str, Any], parent_metadata: dict[str, Any]) -> dict[str, Any]: +def _subagent_child_metadata( + kwargs: dict[str, Any], parent_metadata: dict[str, Any] +) -> dict[str, Any]: child_session_id = _child_session_id(kwargs) metadata = { "session_id": child_session_id, @@ -1023,7 +1447,10 @@ def _subagent_child_metadata(kwargs: dict[str, Any], parent_metadata: dict[str, def _api_key(kwargs: dict[str, Any]) -> str: - return str(kwargs.get("api_request_id") or f"{_session_id(kwargs)}:{kwargs.get('api_call_count') or 'api'}") + return str( + kwargs.get("api_request_id") + or f"{_session_id(kwargs)}:{kwargs.get('api_call_count') or 'api'}" + ) def _tool_key(kwargs: dict[str, Any]) -> str: @@ -1104,7 +1531,9 @@ def _json_semantically_equal(left: Any, right: Any) -> bool: """Compare JSON-compatible values without conflating booleans and numbers.""" try: options = {"ensure_ascii": False, "sort_keys": True, "separators": (",", ":")} - return json.dumps(_jsonable(left), **options) == json.dumps(_jsonable(right), **options) + return json.dumps(_jsonable(left), **options) == json.dumps( + _jsonable(right), **options + ) except (TypeError, ValueError): return False @@ -1119,12 +1548,16 @@ def _original_downstream_error(exc: Exception) -> BaseException: # Hermes wraps downstream execution failures in a local/private exception # class, so detect the wrapper by shape instead of importing it here. original = getattr(exc, "original", None) - if exc.__class__.__name__ == "_DownstreamExecutionError" and isinstance(original, BaseException): + if exc.__class__.__name__ == "_DownstreamExecutionError" and isinstance( + original, BaseException + ): return original return exc -def _is_relay_wrapped_callback_error(exc: Exception, callback_error: Exception | None) -> bool: +def _is_relay_wrapped_callback_error( + exc: Exception, callback_error: Exception | None +) -> bool: # NeMo Relay re-wraps a failing callback as ``RuntimeError("internal error: # : ")``. Match by prefix rather than exact equality so a # trailing traceback/suffix in a future Relay version doesn't silently defeat @@ -1165,13 +1598,25 @@ def _llm_response_payload(response: Any) -> Any: if reasoning is not None: assistant_message["reasoning_content"] = _jsonable(reasoning) elif isinstance(payload, dict): - assistant_message["content"] = payload.get("content") or payload.get("output_text") or "" + assistant_message["content"] = ( + payload.get("content") or payload.get("output_text") or "" + ) return { - "model": _value(response, "model", payload.get("model") if isinstance(payload, dict) else None), + "model": _value( + response, + "model", + payload.get("model") if isinstance(payload, dict) else None, + ), "assistant_message": assistant_message, "finish_reason": finish_reason, - "usage": _jsonable(_value(response, "usage", payload.get("usage") if isinstance(payload, dict) else None)), + "usage": _jsonable( + _value( + response, + "usage", + payload.get("usage") if isinstance(payload, dict) else None, + ) + ), } @@ -1181,16 +1626,14 @@ def _tool_calls_payload(tool_calls: Any) -> list[dict[str, Any]]: normalized: list[dict[str, Any]] = [] for call in tool_calls: function = _value(call, "function") - normalized.append( - { - "id": _value(call, "id"), - "type": _value(call, "type", "function") or "function", - "function": { - "name": _value(function, "name"), - "arguments": _value(function, "arguments"), - }, - } - ) + normalized.append({ + "id": _value(call, "id"), + "type": _value(call, "type", "function") or "function", + "function": { + "name": _value(function, "name"), + "arguments": _value(function, "arguments"), + }, + }) return normalized diff --git a/plugins/observability/nemo_relay/schemas/hermes.shared_metrics.v1.schema.json b/plugins/observability/nemo_relay/schemas/hermes.shared_metrics.v1.schema.json new file mode 100644 index 000000000000..f7306597bc67 --- /dev/null +++ b/plugins/observability/nemo_relay/schemas/hermes.shared_metrics.v1.schema.json @@ -0,0 +1,153 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "urn:hermes-agent:schema:shared-metrics:v1", + "title": "Hermes Shared Metrics Package v1", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "package_id", + "install_id", + "period_start", + "period_end", + "generated_at", + "resource", + "metrics" + ], + "properties": { + "schema_version": { + "const": "hermes.shared_metrics.v1" + }, + "package_id": { + "$ref": "#/$defs/uuid" + }, + "install_id": { + "$ref": "#/$defs/uuid" + }, + "period_start": { + "type": "string", + "format": "date-time" + }, + "period_end": { + "type": "string", + "format": "date-time" + }, + "generated_at": { + "type": "string", + "format": "date-time" + }, + "resource": { + "type": "object", + "additionalProperties": false, + "required": [ + "hermes_version" + ], + "properties": { + "hermes_version": { + "type": "string", + "minLength": 1, + "maxLength": 64 + } + } + }, + "metrics": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/$defs/model_call_counter" + } + } + }, + "$defs": { + "uuid": { + "type": "string", + "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" + }, + "model_call_counter": { + "type": "object", + "additionalProperties": false, + "required": [ + "name", + "type", + "dimensions", + "value" + ], + "properties": { + "name": { + "const": "hermes.model_call.count" + }, + "type": { + "const": "counter" + }, + "dimensions": { + "type": "object", + "additionalProperties": false, + "required": [ + "call_role", + "locality", + "model_family", + "outcome", + "provider_family" + ], + "properties": { + "call_role": { + "const": "primary" + }, + "locality": { + "enum": [ + "local", + "remote", + "unknown" + ] + }, + "model_family": { + "enum": [ + "claude", + "deepseek", + "gemini", + "gemma", + "glm", + "gpt", + "grok", + "kimi", + "llama", + "minimax", + "mimo", + "mistral", + "nemotron", + "nova", + "o1", + "o3", + "o4", + "qwen", + "step", + "trinity", + "unknown" + ] + }, + "outcome": { + "enum": [ + "cancelled", + "failed", + "success" + ] + }, + "provider_family": { + "enum": [ + "aggregator", + "custom", + "direct", + "local", + "unknown" + ] + } + } + }, + "value": { + "type": "integer", + "minimum": 1 + } + } + } + } +} diff --git a/plugins/observability/nemo_relay/shared_metrics.py b/plugins/observability/nemo_relay/shared_metrics.py new file mode 100644 index 000000000000..4848ba827593 --- /dev/null +++ b/plugins/observability/nemo_relay/shared_metrics.py @@ -0,0 +1,376 @@ +"""Durable aggregation and local export for Hermes shared metrics.""" + +from __future__ import annotations + +import json +import sqlite3 +import uuid +from collections.abc import Iterator +from contextlib import contextmanager +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +from hermes_cli.sqlite_util import write_txn +from hermes_constants import get_hermes_home +from utils import atomic_json_write + + +_PACKAGE_SCHEMA_VERSION = "hermes.shared_metrics.v1" +_MODEL_CALL_METRIC = "hermes.model_call.count" +_STORE_SCHEMA_VERSION = "1" +_BUSY_TIMEOUT_MS = 250 + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _isoformat(value: datetime) -> str: + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +class SharedMetricsStore: + """Persist model-call counters and export immutable delta packages.""" + + def __init__( + self, + database_path: Path | None = None, + outbox_directory: Path | None = None, + ) -> None: + root = get_hermes_home() / "telemetry" / "shared_metrics" + self.database_path = database_path or root / "metrics.sqlite3" + self.outbox_directory = outbox_directory or root / "outbox" + self._ensure_private_directory(self.database_path.parent) + self._ensure_private_directory(self.outbox_directory) + self._ensure_schema() + + def record_model_call( + self, + dimensions: dict[str, str], + hermes_version: str, + ) -> None: + """Increment the terminal model-call counter for the current UTC day.""" + dimensions_json = json.dumps( + dimensions, + sort_keys=True, + separators=(",", ":"), + ) + period_start = _utc_now().date().isoformat() + with self._connection() as connection: + connection.execute( + """ + INSERT INTO counter_aggregates( + period_start, + metric_name, + hermes_version, + dimensions_json, + value, + packaged_value + ) VALUES (?, ?, ?, ?, 1, 0) + ON CONFLICT( + period_start, + metric_name, + hermes_version, + dimensions_json + ) + DO UPDATE SET value = value + 1 + """, + ( + period_start, + _MODEL_CALL_METRIC, + hermes_version or "unknown", + dimensions_json, + ), + ) + + def create_and_export_package(self) -> list[Path]: + """Commit one pending delta package, then atomically export the outbox.""" + pending_periods = self._pending_period_count() + for _ in range(pending_periods): + if self._create_package() is None: + break + return self._export_pending_packages() + + def counter_snapshot(self) -> list[dict[str, Any]]: + """Return cumulative counters for focused tests and local inspection.""" + with self._connection() as connection: + rows = connection.execute( + """ + SELECT + period_start, + metric_name, + hermes_version, + dimensions_json, + value, + packaged_value + FROM counter_aggregates + ORDER BY period_start, hermes_version, metric_name, dimensions_json + """ + ).fetchall() + return [ + { + "period_start": row["period_start"], + "metric_name": row["metric_name"], + "hermes_version": row["hermes_version"], + "dimensions": json.loads(row["dimensions_json"]), + "value": row["value"], + "packaged_value": row["packaged_value"], + } + for row in rows + ] + + @contextmanager + def _connection(self) -> Iterator[sqlite3.Connection]: + connection = sqlite3.connect( + self.database_path, + timeout=_BUSY_TIMEOUT_MS / 1000, + ) + try: + try: + self.database_path.chmod(0o600) + except OSError: + pass + connection.row_factory = sqlite3.Row + connection.execute(f"PRAGMA busy_timeout={_BUSY_TIMEOUT_MS}") + with connection: + yield connection + finally: + connection.close() + + @staticmethod + def _ensure_private_directory(path: Path) -> None: + path.mkdir(parents=True, exist_ok=True, mode=0o700) + try: + path.chmod(0o700) + except OSError: + pass + + def _ensure_schema(self) -> None: + with self._connection() as connection: + # Serialize first-run creation and upgrades across Hermes processes. + with write_txn(connection): + self._ensure_schema_in_transaction(connection) + + @staticmethod + def _ensure_schema_in_transaction(connection: sqlite3.Connection) -> None: + connection.execute( + """ + CREATE TABLE IF NOT EXISTS telemetry_state ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ) + """ + ) + schema_row = connection.execute( + "SELECT value FROM telemetry_state WHERE key = 'schema_version'" + ).fetchone() + if schema_row is not None and str(schema_row["value"]) != _STORE_SCHEMA_VERSION: + raise RuntimeError( + "Unsupported shared-metrics store schema version: " + f"{schema_row['value']}" + ) + connection.execute( + """ + CREATE TABLE IF NOT EXISTS counter_aggregates ( + period_start TEXT NOT NULL, + metric_name TEXT NOT NULL, + hermes_version TEXT NOT NULL, + dimensions_json TEXT NOT NULL, + value INTEGER NOT NULL, + packaged_value INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY ( + period_start, + metric_name, + hermes_version, + dimensions_json + ) + ) + """ + ) + connection.execute( + """ + CREATE TABLE IF NOT EXISTS package_outbox ( + package_id TEXT PRIMARY KEY, + period_start TEXT NOT NULL, + period_end TEXT NOT NULL, + payload_json TEXT NOT NULL, + created_at TEXT NOT NULL, + exported_at TEXT + ) + """ + ) + connection.execute( + """ + INSERT OR IGNORE INTO telemetry_state(key, value) + VALUES ('schema_version', ?) + """, + (_STORE_SCHEMA_VERSION,), + ) + + def _install_id(self, connection: sqlite3.Connection) -> str: + row = connection.execute( + "SELECT value FROM telemetry_state WHERE key = 'install_id'" + ).fetchone() + if row is not None: + return str(row["value"]) + candidate = str(uuid.uuid4()) + connection.execute( + "INSERT OR IGNORE INTO telemetry_state(key, value) VALUES ('install_id', ?)", + (candidate,), + ) + row = connection.execute( + "SELECT value FROM telemetry_state WHERE key = 'install_id'" + ).fetchone() + if row is None: + raise RuntimeError("Unable to create the shared-metrics install identity") + return str(row["value"]) + + def _pending_period_count(self) -> int: + with self._connection() as connection: + row = connection.execute( + """ + SELECT COUNT(*) AS period_count + FROM ( + SELECT period_start, hermes_version + FROM counter_aggregates + WHERE value > packaged_value + GROUP BY period_start, hermes_version + ) + """ + ).fetchone() + return int(row["period_count"]) if row is not None else 0 + + def _create_package(self) -> dict[str, Any] | None: + now = _utc_now() + with self._connection() as connection: + with write_txn(connection): + return self._create_package_in_transaction(connection, now) + + def _create_package_in_transaction( + self, + connection: sqlite3.Connection, + now: datetime, + ) -> dict[str, Any] | None: + period_row = connection.execute( + """ + SELECT period_start, hermes_version + FROM counter_aggregates + WHERE value > packaged_value + ORDER BY period_start, hermes_version + LIMIT 1 + """ + ).fetchone() + period_value = period_row["period_start"] if period_row is not None else None + if not period_value: + return None + + rows = connection.execute( + """ + SELECT metric_name, dimensions_json, value, packaged_value + FROM counter_aggregates + WHERE period_start = ? + AND hermes_version = ? + AND value > packaged_value + ORDER BY metric_name, dimensions_json + """, + (period_value, period_row["hermes_version"]), + ).fetchall() + period_start = datetime.fromisoformat(str(period_value)).replace( + tzinfo=timezone.utc + ) + period_end = period_start + timedelta(days=1) + package_id = str(uuid.uuid4()) + payload = { + "schema_version": _PACKAGE_SCHEMA_VERSION, + "package_id": package_id, + "install_id": self._install_id(connection), + "period_start": _isoformat(period_start), + "period_end": _isoformat(period_end), + "generated_at": _isoformat(now), + "resource": {"hermes_version": period_row["hermes_version"]}, + "metrics": [ + { + "name": row["metric_name"], + "type": "counter", + "dimensions": json.loads(row["dimensions_json"]), + "value": row["value"] - row["packaged_value"], + } + for row in rows + ], + } + payload_json = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ) + connection.execute( + """ + INSERT INTO package_outbox( + package_id, + period_start, + period_end, + payload_json, + created_at + ) VALUES (?, ?, ?, ?, ?) + """, + ( + package_id, + payload["period_start"], + payload["period_end"], + payload_json, + payload["generated_at"], + ), + ) + for row in rows: + connection.execute( + """ + UPDATE counter_aggregates + SET packaged_value = value + WHERE period_start = ? + AND metric_name = ? + AND hermes_version = ? + AND dimensions_json = ? + """, + ( + period_value, + row["metric_name"], + period_row["hermes_version"], + row["dimensions_json"], + ), + ) + return payload + + def _export_pending_packages(self) -> list[Path]: + with self._connection() as connection: + rows = connection.execute( + """ + SELECT package_id, payload_json + FROM package_outbox + WHERE exported_at IS NULL + ORDER BY created_at, package_id + """ + ).fetchall() + + exported: list[Path] = [] + for row in rows: + package_id = str(row["package_id"]) + path = self.outbox_directory / f"{package_id}.json" + atomic_json_write( + path, + json.loads(row["payload_json"]), + indent=2, + sort_keys=True, + mode=0o600, + ) + with self._connection() as connection: + connection.execute( + """ + UPDATE package_outbox + SET exported_at = ? + WHERE package_id = ? AND exported_at IS NULL + """, + (_isoformat(_utc_now()), package_id), + ) + exported.append(path) + return exported diff --git a/plugins/observability/nemo_relay/shared_metrics_contract.py b/plugins/observability/nemo_relay/shared_metrics_contract.py new file mode 100644 index 000000000000..d7937a6483ab --- /dev/null +++ b/plugins/observability/nemo_relay/shared_metrics_contract.py @@ -0,0 +1,251 @@ +"""Bounded product contract for the first Hermes shared-metrics slice.""" + +from __future__ import annotations + +import re +from functools import lru_cache +from typing import Any + +SCHEMA_KEY = "hermes.metrics.schema_version" +SCHEMA_VERSION = "hermes.metrics.event.v1" +SESSION_SCOPE = "hermes.session" +MODEL_CALL_SCOPE = "hermes.model_call" +SUBSCRIBER_NAME = "hermes.nemo_relay.shared_metrics" +PRIMARY_MODEL_CALL_ROLE = "primary" + +EXECUTION_SURFACES = frozenset({ + "api", + "batch", + "cli", + "desktop", + "gateway", + "python", + "scheduled_task", + "tui", + "other", + "unknown", +}) +PROVIDER_FAMILIES = frozenset({"aggregator", "custom", "direct", "local", "unknown"}) +MODEL_LOCALITIES = frozenset({"local", "remote", "unknown"}) +MODEL_OUTCOMES = frozenset({"cancelled", "failed", "success"}) + +# Shared metrics use an explicit family allowlist rather than raw model IDs or +# dynamically sourced catalog values. The latter would make the exported schema +# drift independently of this contract. +MODEL_FAMILIES = frozenset({ + "claude", + "deepseek", + "gemini", + "gemma", + "glm", + "gpt", + "grok", + "kimi", + "llama", + "minimax", + "mimo", + "mistral", + "nemotron", + "nova", + "qwen", + "step", + "trinity", + "o1", + "o3", + "o4", + "unknown", +}) + +_MODEL_FAMILY_PATTERN = re.compile( + r"(?:^|[/_.:-])(" + + "|".join( + re.escape(family) + for family in sorted(MODEL_FAMILIES - {"unknown"}, key=len, reverse=True) + ) + + r")(?=$|[/_.:-]|\d)" +) + +# These providers route across model families but are not marked as aggregators +# in Hermes's execution metadata because that flag has narrower routing/catalog +# semantics there. +_TELEMETRY_AGGREGATOR_OVERRIDES = frozenset({ + "copilot-acp", + "github-copilot", + "moa", + "nous", +}) + +# Hermes intentionally resolves these local runtimes through the generic custom +# provider path, so canonical provider metadata cannot distinguish them alone. +_LOCAL_CUSTOM_PROVIDER_ALIASES = frozenset({"mlx", "ollama"}) + + +def model_call_dimensions(event: Any) -> dict[str, str] | None: + """Return package dimensions for one valid primary model-call end 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} + 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 "") != "llm" + or str(getattr(event, "name", "") or "") != MODEL_CALL_SCOPE + or str(getattr(event, "scope_category", "") or "") != "end" + ): + return None + category_profile = getattr(event, "category_profile", None) + if not isinstance(category_profile, dict) or set(category_profile) != { + "model_name" + }: + return None + event_model_family = category_profile.get("model_name") + if event_model_family not in MODEL_FAMILIES: + return None + data = getattr(event, "data", None) + expected_fields = { + "call_role", + "locality", + "outcome", + "provider_family", + } + if not isinstance(data, dict) or set(data) != expected_fields: + return None + if ( + data.get("call_role") != PRIMARY_MODEL_CALL_ROLE + or data.get("locality") not in MODEL_LOCALITIES + or data.get("outcome") not in MODEL_OUTCOMES + or data.get("provider_family") not in PROVIDER_FAMILIES + ): + return None + return { + "call_role": PRIMARY_MODEL_CALL_ROLE, + "locality": data["locality"], + "model_family": event_model_family, + "outcome": data["outcome"], + "provider_family": data["provider_family"], + } + + +def execution_surface(kwargs: dict[str, Any]) -> str: + """Normalize the safe session surface carried by the parent Relay scope.""" + value = ( + str(kwargs.get("execution_surface") or kwargs.get("platform") or "unknown") + .strip() + .lower() + ) + if value in EXECUTION_SURFACES: + return value + if value == "api_server": + return "api" + if value in {"cron", "scheduler", "scheduled"}: + return "scheduled_task" + try: + from hermes_cli.platforms import get_all_platforms + + if value in get_all_platforms(): + return "gateway" + except Exception: + pass + if value in {"discord", "email", "slack", "telegram", "teams", "whatsapp"}: + return "gateway" + return "unknown" if value == "unknown" else "other" + + +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("_", "-") + if not raw_provider: + return "unknown" + if raw_provider in _LOCAL_CUSTOM_PROVIDER_ALIASES: + return "local" + if raw_provider == "custom" or raw_provider.startswith(("custom-", "custom:")): + return "custom" + provider, is_aggregator, is_known = _provider_metadata(raw_provider) + if provider in {"lmstudio", "local"}: + return "local" + if is_aggregator or provider in _TELEMETRY_AGGREGATOR_OVERRIDES: + return "aggregator" + if provider == "custom": + return "custom" + return "direct" if is_known else "unknown" + + +def _provider_metadata(provider: str) -> tuple[str, bool, bool]: + """Resolve provider identity without refreshing remote provider metadata.""" + try: + from hermes_cli.models import normalize_provider as normalize_model_provider + from hermes_cli.providers import HERMES_OVERLAYS, normalize_provider + + canonical = normalize_provider(normalize_model_provider(provider)) + overlay = HERMES_OVERLAYS.get(canonical) + return ( + canonical, + bool(overlay and overlay.is_aggregator), + canonical in _known_provider_ids(), + ) + except Exception: + return provider, False, False + + +@lru_cache(maxsize=1) +def _known_provider_ids() -> frozenset[str]: + """Cache Hermes's static provider catalog for the process lifetime.""" + try: + from hermes_cli.provider_catalog import provider_catalog_by_slug + + return frozenset(provider_catalog_by_slug()) + except Exception: + return frozenset() + + +def model_locality(kwargs: dict[str, Any]) -> str: + """Classify local endpoints without exporting their URL.""" + return _model_locality(kwargs, provider_family(kwargs)) + + +def _model_locality(kwargs: dict[str, Any], provider_category: str) -> str: + base_url = kwargs.get("base_url") + if isinstance(base_url, str) and base_url: + try: + from agent.model_metadata import is_local_endpoint + + if is_local_endpoint(base_url): + return "local" + except Exception: + pass + if provider_category == "local": + return "local" + if provider_category in {"aggregator", "direct"}: + return "remote" + return "unknown" + + +def model_call_fields(kwargs: dict[str, Any]) -> dict[str, str]: + """Build the bounded producer fields for one logical model call.""" + provider_category = provider_family(kwargs) + return { + "call_role": PRIMARY_MODEL_CALL_ROLE, + "locality": _model_locality(kwargs, provider_category), + "model_family": model_family(kwargs), + "provider_family": provider_category, + } + + +def model_family(kwargs: dict[str, Any]) -> str: + """Map a raw model identifier to an allowlisted family.""" + declared_family = str(kwargs.get("model_family") or "").strip().lower() + if declared_family in MODEL_FAMILIES - {"unknown"}: + return declared_family + model = str(kwargs.get("model") or "").lower() + match = _MODEL_FAMILY_PATTERN.search(model) + return match.group(1) if match is not None else "unknown" + + +def model_call_outcome(kwargs: dict[str, Any]) -> str: + """Fail closed when a terminal model-call outcome is not recognized.""" + value = str(kwargs.get("outcome") or "").lower() + return value if value in MODEL_OUTCOMES else "failed" diff --git a/plugins/observability/nemo_relay/shared_metrics_subscriber.py b/plugins/observability/nemo_relay/shared_metrics_subscriber.py new file mode 100644 index 000000000000..be5aa6b53b0b --- /dev/null +++ b/plugins/observability/nemo_relay/shared_metrics_subscriber.py @@ -0,0 +1,31 @@ +"""Relay subscriber for the persisted Hermes shared-metrics slice.""" + +from __future__ import annotations + +import logging +from typing import Any + +from .shared_metrics import SharedMetricsStore +from .shared_metrics_contract import model_call_dimensions + +logger = logging.getLogger(__name__) + + +class SharedMetricsSubscriber: + """Persist validated primary model-call counters from Relay events.""" + + def __init__(self, store: SharedMetricsStore, hermes_version: str) -> None: + self.store = store + self._hermes_version = hermes_version or "unknown" + + def __call__(self, event: Any) -> None: + dimensions = model_call_dimensions(event) + 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, + ) diff --git a/pyproject.toml b/pyproject.toml index faf5b6efcf4e..19950019f4c0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -205,7 +205,7 @@ vision = [] # extra that exposes a Starlette-backed server surface so pip/uv can't resolve # a vulnerable pre-1.0.1 transitive. Bump in lockstep with uv.lock. mcp = ["mcp==1.26.0", "starlette==1.0.1"] # starlette: CVE-2026-48710 -nemo-relay = ["nemo-relay>=0.5,<1.0"] +nemo-relay = ["nemo-relay>=0.5.0,<0.6.0"] homeassistant = ["aiohttp==3.14.1"] sms = ["aiohttp==3.14.1"] teams = ["microsoft-teams-apps==2.0.13.4", "aiohttp==3.14.1"] # aiohttp 3.14.1: CVE-2026-34993(RCE)/47265 + 34513/34518/34519/34520/34525 @@ -351,6 +351,7 @@ plugins = [ "**/plugin.yaml", "**/plugin.yml", "**/README.md", + "**/schemas/*.json", ] [tool.setuptools.packages.find] diff --git a/scripts/smoke_nemo_relay_shared_metrics.py b/scripts/smoke_nemo_relay_shared_metrics.py new file mode 100644 index 000000000000..8ea4dde1b200 --- /dev/null +++ b/scripts/smoke_nemo_relay_shared_metrics.py @@ -0,0 +1,379 @@ +"""Run a real Hermes CLI turn and validate the Relay shared-metrics output.""" + +from __future__ import annotations + +import argparse +import json +import os +import sqlite3 +import subprocess +import sys +import tempfile +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any + + +PROMPT_CANARY = "relay-smoke-sensitive-prompt" +MODEL_CANARY = "gpt-relay-smoke-sensitive-model" +RESPONSE_CANARY = "relay-smoke-sensitive-response" + + +class _ModelHandler(BaseHTTPRequestHandler): + """Minimal OpenAI-compatible model server for one deterministic turn.""" + + protocol_version = "HTTP/1.1" + requests: list[dict[str, Any]] = [] + + def do_GET(self) -> None: # noqa: N802 + if self.path.rstrip("/") != "/v1/models": + self.send_error(404) + return + self._write_json({ + "object": "list", + "data": [ + { + "id": MODEL_CANARY, + "object": "model", + "created": 0, + "owned_by": "smoke-test", + } + ], + }) + + def do_POST(self) -> None: # noqa: N802 + if self.path.rstrip("/") != "/v1/chat/completions": + self.send_error(404) + return + length = int(self.headers.get("Content-Length", "0")) + request = json.loads(self.rfile.read(length) or b"{}") + type(self).requests.append(request) + if request.get("stream"): + self._write_stream() + else: + self._write_json({ + "id": "chatcmpl-relay-smoke", + "object": "chat.completion", + "created": int(time.time()), + "model": MODEL_CANARY, + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": RESPONSE_CANARY, + }, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 1, + "total_tokens": 11, + }, + }) + + def log_message(self, format: str, *args: Any) -> None: + return + + def _write_json(self, payload: dict[str, Any]) -> None: + body = json.dumps(payload).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.send_header("Connection", "close") + self.end_headers() + self.wfile.write(body) + self.close_connection = True + + def _write_stream(self) -> None: + now = int(time.time()) + chunks = [ + { + "id": "chatcmpl-relay-smoke", + "object": "chat.completion.chunk", + "created": now, + "model": MODEL_CANARY, + "choices": [ + { + "index": 0, + "delta": { + "role": "assistant", + "content": RESPONSE_CANARY, + }, + "finish_reason": None, + } + ], + }, + { + "id": "chatcmpl-relay-smoke", + "object": "chat.completion.chunk", + "created": now, + "model": MODEL_CANARY, + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + }, + { + "id": "chatcmpl-relay-smoke", + "object": "chat.completion.chunk", + "created": now, + "model": MODEL_CANARY, + "choices": [], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 1, + "total_tokens": 11, + }, + }, + ] + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.send_header("Connection", "close") + self.end_headers() + for chunk in chunks: + self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode("utf-8")) + self.wfile.flush() + self.wfile.write(b"data: [DONE]\n\n") + self.wfile.flush() + self.close_connection = True + + +def _arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--hermes-repo", + type=Path, + default=Path.cwd(), + help="Hermes source checkout containing .venv/bin/hermes", + ) + parser.add_argument( + "--relay-python", + type=Path, + default=None, + help="NeMo Relay checkout's python directory", + ) + parser.add_argument( + "--output-dir", + type=Path, + default=None, + help="Directory for the isolated HERMES_HOME and captured output", + ) + return parser.parse_args() + + +def _write_config(home: Path, port: int) -> None: + home.mkdir(parents=True, exist_ok=True) + (home / "config.yaml").write_text( + f"""model: + default: {MODEL_CANARY} + provider: custom + base_url: http://127.0.0.1:{port}/v1 + api_mode: chat_completions + api_key: no-key-required +security: + tirith_enabled: false +plugins: + enabled: + - observability/nemo_relay + entries: + observability/nemo_relay: + shared_metrics: + enabled: true +""", + encoding="utf-8", + ) + + +def _validate_store(database_path: Path) -> list[dict[str, Any]]: + if not database_path.is_file(): + raise AssertionError(f"Metrics database was not created: {database_path}") + with sqlite3.connect(database_path) as connection: + rows = connection.execute( + """ + SELECT metric_name, dimensions_json, value, packaged_value + FROM counter_aggregates + ORDER BY metric_name, dimensions_json + """ + ).fetchall() + counters = [ + { + "name": name, + "dimensions": json.loads(dimensions), + "value": value, + "packaged_value": packaged_value, + } + 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: + raise AssertionError( + f"Unexpected SQLite counters:\n{json.dumps(counters, indent=2)}" + ) + return counters + + +def _validate_package(outbox: Path, schema_path: Path) -> tuple[Path, dict[str, Any]]: + packages = sorted(outbox.glob("*.json")) + if len(packages) != 1: + raise AssertionError(f"Expected one package in {outbox}, found {len(packages)}") + package_path = packages[0] + package = json.loads(package_path.read_text(encoding="utf-8")) + try: + import jsonschema + except ImportError as exc: + raise RuntimeError( + "The Hermes development environment requires jsonschema" + ) from exc + schema = json.loads(schema_path.read_text(encoding="utf-8")) + jsonschema.validate(package, schema) + + serialized = json.dumps(package) + for prohibited in (PROMPT_CANARY, MODEL_CANARY, RESPONSE_CANARY): + if prohibited in serialized: + 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]: + raise AssertionError( + f"Unexpected package metrics:\n{json.dumps(package.get('metrics'), indent=2)}" + ) + return package_path, package + + +def main() -> int: + args = _arguments() + hermes_repo = args.hermes_repo.resolve() + relay_python = ( + args.relay_python.resolve() + if args.relay_python + else (hermes_repo.parent / "nemo-relay" / "python").resolve() + ) + hermes = hermes_repo / ".venv" / "bin" / "hermes" + if not hermes.is_file(): + raise SystemExit(f"Hermes executable not found: {hermes}") + if not any((relay_python / "nemo_relay").glob("_native.*")): + raise SystemExit( + "Built NeMo Relay Python binding not found under " + f"{relay_python}; run the Relay Python build first" + ) + + if args.output_dir: + root = args.output_dir.resolve() + if root.exists(): + raise SystemExit(f"Refusing to replace existing output directory: {root}") + root.mkdir(parents=True) + else: + root = Path(tempfile.mkdtemp(prefix="hermes-relay-shared-metrics-")) + home = root / "hermes-home" + workdir = root / "workspace" + workdir.mkdir() + home.mkdir() + (home / ".no-bundled-skills").touch() + + _ModelHandler.requests = [] + server = ThreadingHTTPServer(("127.0.0.1", 0), _ModelHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + _write_config(home, server.server_port) + env = os.environ.copy() + env["HERMES_HOME"] = str(home) + env["PYTHONPATH"] = os.pathsep.join([ + str(relay_python), + env.get("PYTHONPATH", ""), + ]).rstrip(os.pathsep) + result = subprocess.run( + [ + str(hermes), + "chat", + "--query", + PROMPT_CANARY, + "--provider", + "custom", + "--model", + MODEL_CANARY, + "--quiet", + "--ignore-rules", + "--toolsets", + "search", + "--max-turns", + "2", + ], + cwd=workdir, + env=env, + text=True, + capture_output=True, + timeout=120, + ) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + (root / "hermes.stdout.txt").write_text(result.stdout, encoding="utf-8") + (root / "hermes.stderr.txt").write_text(result.stderr, encoding="utf-8") + if result.returncode != 0: + raise AssertionError( + f"Hermes exited with {result.returncode}\n" + f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}" + ) + if not _ModelHandler.requests: + raise AssertionError("Hermes did not call the local model endpoint") + request = _ModelHandler.requests[0] + if request.get("model") != MODEL_CANARY: + raise AssertionError(f"Unexpected model request: {request.get('model')!r}") + if PROMPT_CANARY not in json.dumps(request.get("messages", [])): + raise AssertionError("Hermes model request did not contain the prompt canary") + if RESPONSE_CANARY not in result.stdout: + raise AssertionError("Hermes did not print the mock model response") + + telemetry = home / "telemetry" / "shared_metrics" + counters = _validate_store(telemetry / "metrics.sqlite3") + package_path, package = _validate_package( + telemetry / "outbox", + hermes_repo + / "plugins" + / "observability" + / "nemo_relay" + / "schemas" + / "hermes.shared_metrics.v1.schema.json", + ) + + print("Hermes -> NeMo Relay shared-metrics smoke test passed") + print(f"Artifact directory: {root}") + print(f"Model requests: {len(_ModelHandler.requests)}") + print(f"SQLite counters: {json.dumps(counters, indent=2)}") + print(f"Export package: {package_path}") + print(json.dumps(package, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/plugins/test_nemo_relay_plugin.py b/tests/plugins/test_nemo_relay_plugin.py index 61958194e585..eabb1a510b4a 100644 --- a/tests/plugins/test_nemo_relay_plugin.py +++ b/tests/plugins/test_nemo_relay_plugin.py @@ -4,10 +4,12 @@ from __future__ import annotations import asyncio import builtins +import contextvars import gc import importlib import json import sys +import threading import warnings from pathlib import Path from types import SimpleNamespace @@ -25,6 +27,10 @@ PLUGIN_DIR = REPO_ROOT / "plugins" / "observability" / "nemo_relay" class _FakeNemoRelay: def __init__(self): self.events = [] + self._llm_handles = {} + self._subscriber_callbacks = {} + self._handle_serial = 0 + self._scope_context = contextvars.ContextVar("fake_relay_scope", default=None) self.ScopeType = SimpleNamespace(Agent="agent") self.scope = SimpleNamespace( push=self._scope_push, @@ -46,14 +52,20 @@ class _FakeNemoRelay: clear=self._plugin_clear, activate_dynamic_plugins=self._plugin_activate_dynamic, ) - self.subscribers = SimpleNamespace(flush=self._flush_subscribers) + self.subscribers = SimpleNamespace( + register=self._register_subscriber, + deregister=self._deregister_subscriber, + flush=self._flush_subscribers, + ) self.LLMRequest = _FakeLLMRequest self.AtofExporterConfig = _FakeAtofExporterConfig self.AtofExporterMode = SimpleNamespace(Append="append", Overwrite="overwrite") self.AtofExporter = self._make_atof_exporter self.AtifExporter = self._make_atif_exporter + self.get_scope_stack = self._get_scope_stack def _scope_push(self, name, scope_type, **kwargs): + self._scope_context.set(name) handle = ("scope", name) self.events.append(("scope.push", name, scope_type, kwargs)) return handle @@ -64,17 +76,46 @@ class _FakeNemoRelay: def _scope_event(self, name, **kwargs): self.events.append(("scope.event", name, kwargs)) + def _get_scope_stack(self): + current = self._scope_context.get() + self.events.append(("scope_stack.sync", current)) + return current + def _llm_call(self, name, request, **kwargs): + self.events.append(("llm.call.context", self._scope_context.get())) handle = ("llm", name) + if handle in self._llm_handles: + self._handle_serial += 1 + handle = ("llm", name, self._handle_serial) + self._llm_handles[handle] = kwargs self.events.append(("llm.call", name, request.content, kwargs)) return handle def _llm_call_end(self, handle, response, **kwargs): + self.events.append(("llm.call_end.context", self._scope_context.get())) self.events.append(("llm.call_end", handle, response, kwargs)) + started = self._llm_handles.pop(handle, {}) + self._emit( + _FakeEvent( + kind="scope", + category="llm", + name=handle[1], + scope_category="end", + data=response, + category_profile={"model_name": started.get("model_name")}, + metadata={ + **(started.get("metadata") or {}), + **(kwargs.get("metadata") or {}), + "otel.status_code": "OK", + }, + ) + ) def _llm_execute(self, name, request, func, **kwargs): self.events.append(("llm.execute.start", name, request.content, kwargs)) - result = func(_FakeLLMRequest(request.headers, {"intercepted": True, **request.content})) + result = func( + _FakeLLMRequest(request.headers, {"intercepted": True, **request.content}) + ) self.events.append(("llm.execute.end", name, result, kwargs)) return result @@ -96,7 +137,9 @@ class _FakeNemoRelay: return _FakeAtofExporter(self.events, config) def _make_atif_exporter(self, session_id, agent_name, agent_version, **kwargs): - return _FakeAtifExporter(self.events, session_id, agent_name, agent_version, kwargs) + return _FakeAtifExporter( + self.events, session_id, agent_name, agent_version, kwargs + ) async def _plugin_initialize(self, config): self.events.append(("plugin.initialize", config)) @@ -112,6 +155,39 @@ class _FakeNemoRelay: def _flush_subscribers(self): self.events.append(("subscribers.flush",)) + def _register_subscriber(self, name, callback): + self.events.append(("subscribers.register", name)) + self._subscriber_callbacks[name] = callback + + def _deregister_subscriber(self, name): + self.events.append(("subscribers.deregister", name)) + return self._subscriber_callbacks.pop(name, None) is not None + + def _emit(self, event): + for callback in list(self._subscriber_callbacks.values()): + callback(event) + + +class _FakeEvent: + def __init__( + self, + *, + kind, + name, + category=None, + category_profile=None, + data=None, + metadata=None, + scope_category=None, + ): + self.kind = kind + self.name = name + self.category = category + self.category_profile = category_profile + self.data = data + self.metadata = metadata + self.scope_category = scope_category + class _FakePluginActivation: def __init__(self, events): @@ -141,10 +217,20 @@ class _FakeAtofExporter: self.config = config def register(self, name): - self.events.append(("atof.register", name, self.config.output_directory, self.config.filename)) + self.events.append(( + "atof.register", + name, + self.config.output_directory, + self.config.filename, + )) def deregister(self, name): - self.events.append(("atof.deregister", name, self.config.output_directory, self.config.filename)) + self.events.append(( + "atof.deregister", + name, + self.config.output_directory, + self.config.filename, + )) return True @@ -165,7 +251,10 @@ class _FakeAtifExporter: def export_json(self): self.events.append(("atif.export", self.session_id)) - return json.dumps({"session_id": self.session_id, "agent_name": self.agent_name}) + return json.dumps({ + "session_id": self.session_id, + "agent_name": self.agent_name, + }) def _fresh_plugin(monkeypatch, fake): @@ -223,6 +312,25 @@ mode = "test" return plugins_toml +def _enable_shared_metrics(tmp_path, monkeypatch) -> Path: + hermes_home = tmp_path / "hermes-home" + hermes_home.mkdir() + (hermes_home / "config.yaml").write_text( + """ +plugins: + enabled: + - observability/nemo_relay + entries: + observability/nemo_relay: + shared_metrics: + enabled: true +""", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + return hermes_home + + def test_manifest_fields(): data = yaml.safe_load((PLUGIN_DIR / "plugin.yaml").read_text()) assert data["name"] == "nemo_relay" @@ -266,13 +374,218 @@ def test_nemo_relay_plugin_uses_nemo_relay_runtime(monkeypatch): assert any(event[0] == "scope.push" for event in fake_relay.events) +def test_shared_metrics_default_off_does_not_create_state(tmp_path, monkeypatch): + hermes_home = tmp_path / "hermes-home" + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + fake = _FakeNemoRelay() + plugin = _fresh_plugin(monkeypatch, fake) + + plugin.on_session_start(session_id="sensitive-session", model="sensitive-model") + + assert not any(event[0] == "subscribers.register" for event in fake.events) + assert not (hermes_home / "telemetry").exists() + + +def test_shared_metrics_counts_one_logical_model_call_across_retries( + tmp_path, monkeypatch +): + hermes_home = _enable_shared_metrics(tmp_path, monkeypatch) + fake = _FakeNemoRelay() + plugin = _fresh_plugin(monkeypatch, fake) + base = { + "session_id": "sensitive-session", + "task_id": "task-1", + "api_request_id": "request-1", + "platform": "cli", + "provider": "custom", + "model": "gpt-sensitive-model-id", + "base_url": "http://127.0.0.1:11434/v1", + } + + plugin.on_session_start(**base) + plugin.on_pre_api_request( + **base, + request={"body": {"messages": ["sensitive-prompt"]}}, + ) + plugin.on_api_request_error( + **base, + retryable=True, + error={"message": "sensitive-error"}, + ) + plugin.on_pre_api_request( + **base, + request={"body": {"messages": ["sensitive-prompt"]}}, + ) + plugin.on_post_api_request( + **base, + response={"content": "sensitive-response"}, + ) + plugin.on_session_finalize(session_id=base["session_id"]) + + model_starts = [ + event for event in fake.events if event[:2] == ("llm.call", "hermes.model_call") + ] + model_ends = [ + event + for event in fake.events + if event[0] == "llm.call_end" and event[1][1] == "hermes.model_call" + ] + assert len(model_starts) == 1 + assert len(model_ends) == 1 + assert [ + event[1] + for event in fake.events + if event[0] in {"llm.call.context", "llm.call_end.context"} + ] == ["hermes.session", "hermes.session"] + assert model_starts[0][2] == {} + assert model_starts[0][3]["model_name"] == "gpt" + assert model_ends[0][2] == { + "call_role": "primary", + "locality": "local", + "outcome": "success", + "provider_family": "custom", + } + serialized_events = json.dumps(fake.events) + assert "sensitive-prompt" not in serialized_events + assert "sensitive-response" not in serialized_events + assert "sensitive-error" not in serialized_events + assert "gpt-sensitive-model-id" not in serialized_events + runtime = plugin._get_runtime() + assert runtime is not None + assert runtime.shared_metrics is not None + assert runtime.shared_metrics.store.counter_snapshot()[0]["value"] == 1 + assert ( + len( + list( + (hermes_home / "telemetry" / "shared_metrics" / "outbox").glob("*.json") + ) + ) + == 1 + ) + + +def test_shared_metrics_closes_unfinished_model_call_as_failed(tmp_path, monkeypatch): + _enable_shared_metrics(tmp_path, monkeypatch) + fake = _FakeNemoRelay() + plugin = _fresh_plugin(monkeypatch, fake) + base = { + "session_id": "s1", + "task_id": "task-1", + "api_request_id": "request-1", + "provider": "anthropic", + "model": "claude-sonnet", + } + + plugin.on_pre_api_request(**base) + plugin.on_api_request_error(**base, retryable=False, error={"message": "private"}) + plugin.on_session_end(session_id="s1", task_id="task-1", interrupted=False) + plugin.on_session_finalize(session_id="s1") + + [model_end] = [ + event + for event in fake.events + if event[0] == "llm.call_end" and event[1][1] == "hermes.model_call" + ] + assert model_end[2]["outcome"] == "failed" + runtime = plugin._get_runtime() + assert runtime is not None + assert ( + runtime.shared_metrics.store.counter_snapshot()[0]["dimensions"]["outcome"] + == "failed" + ) + + +def test_shared_metrics_persistence_failure_is_fail_open(tmp_path, monkeypatch, caplog): + _enable_shared_metrics(tmp_path, monkeypatch) + fake = _FakeNemoRelay() + plugin = _fresh_plugin(monkeypatch, fake) + runtime = plugin._get_runtime() + assert runtime is not None + assert runtime.shared_metrics is not None + + def fail_record(*_args, **_kwargs): + raise OSError("store unavailable") + + monkeypatch.setattr(runtime.shared_metrics.store, "record_model_call", fail_record) + plugin.on_pre_api_request( + session_id="s1", + task_id="t1", + api_request_id="r1", + provider="openai", + model="gpt-5", + ) + plugin.on_post_api_request( + session_id="s1", + task_id="t1", + api_request_id="r1", + provider="openai", + model="gpt-5", + ) + + assert "Unable to persist the Hermes model-call metric" in caplog.text + + +def test_shared_metrics_close_does_not_reopen_a_failed_session_scope( + tmp_path, monkeypatch +): + _enable_shared_metrics(tmp_path, monkeypatch) + fake = _FakeNemoRelay() + original_push = fake.scope.push + push_attempts = 0 + + def fail_first_metrics_scope(*args, **kwargs): + nonlocal push_attempts + push_attempts += 1 + if push_attempts == 1: + raise RuntimeError("simulated metrics scope failure") + return original_push(*args, **kwargs) + + fake.scope.push = fail_first_metrics_scope + plugin = _fresh_plugin(monkeypatch, fake) + runtime = plugin._get_runtime() + assert runtime is not None + runtime.ensure_session({"session_id": "s1"}) + state = runtime.sessions["s1"] + assert state.metrics_handle is None + + close_started = threading.Event() + allow_close = threading.Event() + original_drain = runtime._end_pending_model_calls + + def block_drain(current_state, kwargs): + assert current_state.closing is True + close_started.set() + assert allow_close.wait(timeout=5) + original_drain(current_state, kwargs) + + monkeypatch.setattr(runtime, "_end_pending_model_calls", block_drain) + close_thread = threading.Thread( + target=runtime.close_session, + args=({"session_id": "s1"},), + ) + close_thread.start() + assert close_started.wait(timeout=5) + + runtime.ensure_session({"session_id": "s1"}) + assert push_attempts == 1 + + allow_close.set() + close_thread.join(timeout=5) + assert not close_thread.is_alive() + assert "s1" not in runtime.sessions + + def test_nemo_relay_plugin_emits_llm_tool_and_exports_atif(tmp_path, monkeypatch): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) monkeypatch.setenv("HERMES_NEMO_RELAY_ATOF_ENABLED", "1") - monkeypatch.setenv("HERMES_NEMO_RELAY_ATOF_OUTPUT_DIRECTORY", str(tmp_path / "atof")) + monkeypatch.setenv( + "HERMES_NEMO_RELAY_ATOF_OUTPUT_DIRECTORY", str(tmp_path / "atof") + ) monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "1") - monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif")) + monkeypatch.setenv( + "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif") + ) base = { "session_id": "s1", @@ -286,15 +599,26 @@ def test_nemo_relay_plugin_emits_llm_tool_and_exports_atif(tmp_path, monkeypatch api_request_id="api-1", provider="openai", model="demo-model", - request={"method": "POST", "body": {"messages": [{"role": "user", "content": "hi"}]}}, + request={ + "method": "POST", + "body": {"messages": [{"role": "user", "content": "hi"}]}, + }, ) plugin.on_post_api_request( **base, api_request_id="api-1", response={"assistant_message": {"role": "assistant", "content": "hello"}}, ) - plugin.on_pre_tool_call(**base, tool_name="read_file", tool_call_id="tool-1", args={"path": "x"}) - plugin.on_post_tool_call(**base, tool_name="read_file", tool_call_id="tool-1", result='{"ok": true}', status="ok") + plugin.on_pre_tool_call( + **base, tool_name="read_file", tool_call_id="tool-1", args={"path": "x"} + ) + plugin.on_post_tool_call( + **base, + tool_name="read_file", + tool_call_id="tool-1", + result='{"ok": true}', + status="ok", + ) plugin.on_session_end(**base, completed=True, interrupted=False) plugin.on_session_finalize(**base, reason="shutdown") @@ -336,7 +660,9 @@ def test_nemo_relay_plugin_closes_api_span_on_error(monkeypatch): call_end = next(event for event in fake.events if event[0] == "llm.call_end") assert call_end[1] == ("llm", "openai") - assert call_end[2] == {"error": {"type": "RateLimitError", "message": "rate limited"}} + assert call_end[2] == { + "error": {"type": "RateLimitError", "message": "rate limited"} + } assert call_end[3]["data"]["reason"] == "rate_limit" assert not plugin._get_runtime().sessions["s1"].llm_spans @@ -345,8 +671,12 @@ def test_nemo_relay_plugin_emits_approval_marks(monkeypatch): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) - plugin.on_pre_approval_request(session_id="s1", approval_id="approval-1", tool_name="shell") - plugin.on_post_approval_response(session_id="s1", approval_id="approval-1", approved=True) + plugin.on_pre_approval_request( + session_id="s1", approval_id="approval-1", tool_name="shell" + ) + plugin.on_post_approval_response( + session_id="s1", approval_id="approval-1", approved=True + ) mark_names = [event[1] for event in fake.events if event[0] == "scope.event"] assert "hermes.approval.request" in mark_names @@ -357,13 +687,17 @@ def test_nemo_relay_plugin_emits_unmatched_fallback_marks(monkeypatch): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) - plugin.on_post_api_request(session_id="s1", api_request_id="missing-api", response={"ok": True}) + plugin.on_post_api_request( + session_id="s1", api_request_id="missing-api", response={"ok": True} + ) plugin.on_api_request_error( session_id="s1", api_request_id="missing-api", error={"type": "TimeoutError", "message": "timed out"}, ) - plugin.on_post_tool_call(session_id="s1", tool_call_id="missing-tool", result={"ok": True}) + plugin.on_post_tool_call( + session_id="s1", tool_call_id="missing-tool", result={"ok": True} + ) mark_names = [event[1] for event in fake.events if event[0] == "scope.event"] assert "hermes.api.response.unmatched" in mark_names @@ -399,12 +733,20 @@ def test_nemo_relay_plugin_metadata_promotes_trajectory_and_subagent_ids(monkeyp telemetry_schema_version="hermes.observer.v1", ) - turn_mark = next(event for event in fake.events if event[0] == "scope.event" and event[1] == "hermes.turn.start") + turn_mark = next( + event + for event in fake.events + if event[0] == "scope.event" and event[1] == "hermes.turn.start" + ) turn_metadata = turn_mark[2]["metadata"] assert turn_metadata["session_id"] == "parent-session" assert turn_metadata["trajectory_id"] == "parent-session" - start_mark = next(event for event in fake.events if event[0] == "scope.event" and event[1] == "hermes.subagent.start") + start_mark = next( + event + for event in fake.events + if event[0] == "scope.event" and event[1] == "hermes.subagent.start" + ) start_metadata = start_mark[2]["metadata"] assert start_metadata["parent_session_id"] == "parent-session" assert start_metadata["parent_trajectory_id"] == "parent-session" @@ -413,7 +755,11 @@ def test_nemo_relay_plugin_metadata_promotes_trajectory_and_subagent_ids(monkeyp assert start_metadata["child_subagent_id"] == "child-sa" assert start_metadata["child_role"] == "leaf" - stop_mark = next(event for event in fake.events if event[0] == "scope.event" and event[1] == "hermes.subagent.stop") + stop_mark = next( + event + for event in fake.events + if event[0] == "scope.event" and event[1] == "hermes.subagent.stop" + ) assert stop_mark[2]["metadata"]["child_status"] == "completed" @@ -446,11 +792,15 @@ def test_nemo_relay_plugin_reparents_child_session_scope_for_embedded_atif(monke assert child_kwargs["metadata"]["parent_session_id"] == "parent-session" -def test_nemo_relay_plugin_skips_embedded_child_atif_file_by_default(tmp_path, monkeypatch): +def test_nemo_relay_plugin_skips_embedded_child_atif_file_by_default( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "1") - monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif")) + monkeypatch.setenv( + "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif") + ) plugin.on_session_start(session_id="parent-session") plugin.on_subagent_start( @@ -468,11 +818,15 @@ def test_nemo_relay_plugin_skips_embedded_child_atif_file_by_default(tmp_path, m assert not (tmp_path / "atif" / "hermes-atif-child-session.json").exists() -def test_nemo_relay_plugin_can_write_embedded_child_atif_file_in_all_mode(tmp_path, monkeypatch): +def test_nemo_relay_plugin_can_write_embedded_child_atif_file_in_all_mode( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "1") - monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif")) + monkeypatch.setenv( + "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif") + ) monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_SUBAGENT_EXPORT_MODE", "all") plugin.on_session_start(session_id="parent-session") @@ -525,7 +879,9 @@ output_directory = "{atif_dir}" assert atif_dir.is_dir() -def test_nemo_relay_plugin_clears_plugins_toml_on_final_session_finalize_and_reinitializes(tmp_path, monkeypatch): +def test_nemo_relay_plugin_clears_plugins_toml_on_final_session_finalize_and_reinitializes( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) plugins_toml = tmp_path / "plugins.toml" @@ -555,7 +911,9 @@ def test_nemo_relay_plugin_activates_and_owns_dynamic_plugins(tmp_path, monkeypa plugin = _fresh_plugin(monkeypatch, fake) _enable_dynamic_plugin(tmp_path, monkeypatch) monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "1") - monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif")) + monkeypatch.setenv( + "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif") + ) plugin.on_session_start(session_id="s1") runtime = plugin._get_runtime() @@ -583,7 +941,9 @@ def test_nemo_relay_plugin_activates_and_owns_dynamic_plugins(tmp_path, monkeypa plugin.on_session_finalize(session_id="s2", reason="shutdown") assert sum(event[0] == "plugin.activate_dynamic" for event in fake.events) == 1 - activation = next(event for event in fake.events if event[0] == "plugin.activate_dynamic") + activation = next( + event for event in fake.events if event[0] == "plugin.activate_dynamic" + ) assert "dynamic_plugins" not in activation[1] assert activation[2] == [ { @@ -600,7 +960,9 @@ def test_nemo_relay_plugin_activates_and_owns_dynamic_plugins(tmp_path, monkeypa runtime.shutdown() event_names = [event[0] for event in fake.events] - assert event_names.index("atif.deregister") < event_names.index("plugin.activation.close") + assert event_names.index("atif.deregister") < event_names.index( + "plugin.activation.close" + ) def test_nemo_relay_rejects_gateway_dynamic_config_with_actionable_diagnostic( @@ -633,7 +995,9 @@ mode = "test" assert "Use Hermes-owned [[dynamic_plugins]]" in caplog.text -def test_nemo_relay_explicit_dynamic_paths_resolve_from_plugins_toml(tmp_path, monkeypatch): +def test_nemo_relay_explicit_dynamic_paths_resolve_from_plugins_toml( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) config_dir = tmp_path / "config" @@ -655,7 +1019,9 @@ environment_ref = "../environments/worker-fixture" plugin.on_session_start(session_id="s1") - activation = next(event for event in fake.events if event[0] == "plugin.activate_dynamic") + activation = next( + event for event in fake.events if event[0] == "plugin.activate_dynamic" + ) assert activation[2] == [ { "plugin_id": "worker-fixture", @@ -729,7 +1095,9 @@ def test_nemo_relay_managed_llm_uses_wire_protocol_for_interceptor_dispatch( assert "rewritten_for" not in result -def test_nemo_relay_managed_llm_returns_post_next_interceptor_result(tmp_path, monkeypatch): +def test_nemo_relay_managed_llm_returns_post_next_interceptor_result( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() raw_response = SimpleNamespace( model="fixture", @@ -792,7 +1160,9 @@ def test_nemo_relay_managed_tool_returns_post_interceptor_result(tmp_path, monke } -def test_nemo_relay_plugin_activates_before_registering_managed_middleware(tmp_path, monkeypatch): +def test_nemo_relay_plugin_activates_before_registering_managed_middleware( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) _enable_dynamic_plugin(tmp_path, monkeypatch) @@ -885,16 +1255,21 @@ manifest_ref = "{(tmp_path / "invalid" / "relay-plugin.toml").as_posix()}" assert "no dynamic plugins will be activated" in caplog.text -def test_nemo_relay_plugin_registers_shutdown_after_dynamic_retry(tmp_path, monkeypatch): +def test_nemo_relay_plugin_registers_shutdown_after_dynamic_retry( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() activation_attempts = 0 async def _flaky_activate(config, dynamic_plugins): nonlocal activation_attempts activation_attempts += 1 - fake.events.append( - ("plugin.activate_dynamic.attempt", activation_attempts, config, dynamic_plugins) - ) + fake.events.append(( + "plugin.activate_dynamic.attempt", + activation_attempts, + config, + dynamic_plugins, + )) if activation_attempts == 1: raise RuntimeError("temporary activation failure") return _FakePluginActivation(fake.events) @@ -938,7 +1313,9 @@ def test_nemo_relay_plugin_attempts_activation_close_after_subscriber_flush_fail event_names = [event[0] for event in fake.events] assert event_names.count("subscribers.flush.failed") == 2 flush_indices = [ - index for index, name in enumerate(event_names) if name == "subscribers.flush.failed" + index + for index, name in enumerate(event_names) + if name == "subscribers.flush.failed" ] assert max(flush_indices) < event_names.index("plugin.activation.close") assert runtime._plugin_activation is None @@ -961,7 +1338,9 @@ def test_nemo_relay_plugin_continues_shutdown_after_atif_export_failure( plugin = _fresh_plugin(monkeypatch, fake) _enable_dynamic_plugin(tmp_path, monkeypatch) monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "1") - monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif")) + monkeypatch.setenv( + "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "atif") + ) plugin.on_session_start(session_id="s1") runtime = plugin._get_runtime() assert runtime is not None @@ -970,13 +1349,19 @@ def test_nemo_relay_plugin_continues_shutdown_after_atif_export_failure( runtime.shutdown() event_names = [event[0] for event in fake.events] - assert event_names.index("atif.export.failed") < event_names.index("atif.deregister") - assert event_names.index("atif.deregister") < event_names.index("plugin.activation.close") + assert event_names.index("atif.export.failed") < event_names.index( + "atif.deregister" + ) + assert event_names.index("atif.deregister") < event_names.index( + "plugin.activation.close" + ) assert runtime._plugin_activation is None assert "ATIF export failed: disk full" in caplog.text -def test_nemo_relay_plugin_keeps_plugins_toml_active_while_other_sessions_remain(tmp_path, monkeypatch): +def test_nemo_relay_plugin_keeps_plugins_toml_active_while_other_sessions_remain( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) plugins_toml = tmp_path / "plugins.toml" @@ -1002,7 +1387,9 @@ enabled = true assert event_names.count("plugin.clear") == 1 -def test_nemo_relay_plugin_reinitializes_plugins_toml_inside_active_event_loop(tmp_path, monkeypatch): +def test_nemo_relay_plugin_reinitializes_plugins_toml_inside_active_event_loop( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) plugins_toml = tmp_path / "plugins.toml" @@ -1037,7 +1424,9 @@ enabled = true assert "hermes-session-s2" in scope_push_names -def test_nemo_relay_plugin_retries_plugins_toml_after_clear_failure(tmp_path, monkeypatch): +def test_nemo_relay_plugin_retries_plugins_toml_after_clear_failure( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() initialize_calls = 0 @@ -1078,7 +1467,9 @@ enabled = true assert "hermes-session-s2" in scope_push_names -def test_nemo_relay_plugin_disables_direct_atif_when_plugins_toml_owns_atif(tmp_path, monkeypatch): +def test_nemo_relay_plugin_disables_direct_atif_when_plugins_toml_owns_atif( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) plugins_toml = tmp_path / "plugins.toml" @@ -1098,7 +1489,9 @@ output_directory = "{(tmp_path / "managed-atif").as_posix()}" ) monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml)) monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "1") - monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "direct-atif")) + monkeypatch.setenv( + "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "direct-atif") + ) plugin.on_session_start(session_id="s1") plugin.on_session_finalize(session_id="s1", reason="shutdown") @@ -1110,7 +1503,9 @@ output_directory = "{(tmp_path / "managed-atif").as_posix()}" assert not (tmp_path / "direct-atif" / "hermes-atif-s1.json").exists() -def test_nemo_relay_plugin_keeps_direct_atif_when_plugins_toml_init_fails(tmp_path, monkeypatch): +def test_nemo_relay_plugin_keeps_direct_atif_when_plugins_toml_init_fails( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() async def _failing_initialize(config): @@ -1136,7 +1531,9 @@ output_directory = "{(tmp_path / "managed-atif").as_posix()}" ) monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml)) monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "1") - monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "direct-atif")) + monkeypatch.setenv( + "HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "direct-atif") + ) plugin.on_session_start(session_id="s1") plugin.on_session_finalize(session_id="s1", reason="shutdown") @@ -1182,7 +1579,9 @@ output_directory = "{(tmp_path / "managed-atof").as_posix()}" ) monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml)) monkeypatch.setenv("HERMES_NEMO_RELAY_ATOF_ENABLED", "1") - monkeypatch.setenv("HERMES_NEMO_RELAY_ATOF_OUTPUT_DIRECTORY", str(tmp_path / "direct-atof")) + monkeypatch.setenv( + "HERMES_NEMO_RELAY_ATOF_OUTPUT_DIRECTORY", str(tmp_path / "direct-atof") + ) plugin.on_session_start(session_id="s1") plugin.on_session_finalize(session_id="s1", reason="shutdown") @@ -1197,7 +1596,9 @@ output_directory = "{(tmp_path / "managed-atof").as_posix()}" assert event_names.count("atof.deregister") == 1 -def test_nemo_relay_adaptive_llm_execution_middleware_preserves_raw_response(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_llm_execution_middleware_preserves_raw_response( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) plugins_toml = tmp_path / "plugins.toml" @@ -1225,7 +1626,9 @@ mode = "observe_only" SimpleNamespace( id="tool-1", type="function", - function=SimpleNamespace(name="terminal", arguments='{"command":"pwd"}'), + function=SimpleNamespace( + name="terminal", arguments='{"command":"pwd"}' + ), ) ], reasoning_content="need a tool", @@ -1259,7 +1662,9 @@ mode = "observe_only" assert response.model == "demo-model" assert response.choices == [raw_choice] assert seen_request["intercepted"] is True - execute_start = next(event for event in fake.events if event[0] == "llm.execute.start") + execute_start = next( + event for event in fake.events if event[0] == "llm.execute.start" + ) assert execute_start[3]["data"]["mode"] == "observe_only" execute_end = next(event for event in fake.events if event[0] == "llm.execute.end") assert execute_end[2] == { @@ -1281,13 +1686,19 @@ mode = "observe_only" } -def test_nemo_relay_adaptive_llm_execution_preserves_downstream_error(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_llm_execution_preserves_downstream_error( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() def native_like_execute(name, request, func, **kwargs): fake.events.append(("llm.execute.start", name, request.content, kwargs)) try: - return func(_FakeLLMRequest(request.headers, {"intercepted": True, **request.content})) + return func( + _FakeLLMRequest( + request.headers, {"intercepted": True, **request.content} + ) + ) except Exception as exc: raise RuntimeError(f"internal error: {type(exc).__name__}: {exc}") from None @@ -1327,9 +1738,15 @@ def test_nemo_relay_adaptive_llm_execution_preserves_downstream_error_with_relay def native_like_execute(name, request, func, **kwargs): try: - return func(_FakeLLMRequest(request.headers, {"intercepted": True, **request.content})) + return func( + _FakeLLMRequest( + request.headers, {"intercepted": True, **request.content} + ) + ) except Exception as exc: - raise RuntimeError(f"internal error: {type(exc).__name__}: {exc} (retried 3x)") from None + raise RuntimeError( + f"internal error: {type(exc).__name__}: {exc} (retried 3x)" + ) from None fake.llm.execute = native_like_execute plugin = _fresh_plugin(monkeypatch, fake) @@ -1356,7 +1773,9 @@ def test_nemo_relay_adaptive_llm_execution_preserves_downstream_error_with_relay assert caught.value.status_code == 403 -def test_nemo_relay_adaptive_llm_execution_keeps_unrelated_internal_error(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_llm_execution_keeps_unrelated_internal_error( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() relay_error = RuntimeError("internal error: relay setup failed") @@ -1384,11 +1803,17 @@ def test_nemo_relay_adaptive_llm_execution_keeps_wrapped_relay_error_after_downs tmp_path, monkeypatch ): fake = _FakeNemoRelay() - relay_error = RuntimeError("internal error: RuntimeError: relay policy blocked after downstream") + relay_error = RuntimeError( + "internal error: RuntimeError: relay policy blocked after downstream" + ) def translated_execute(name, request, func, **kwargs): try: - return func(_FakeLLMRequest(request.headers, {"intercepted": True, **request.content})) + return func( + _FakeLLMRequest( + request.headers, {"intercepted": True, **request.content} + ) + ) except Exception: raise relay_error @@ -1411,7 +1836,9 @@ def test_nemo_relay_adaptive_llm_execution_keeps_wrapped_relay_error_after_downs assert caught.value is relay_error -def test_nemo_relay_adaptive_llm_execution_keeps_relay_translated_error(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_llm_execution_keeps_relay_translated_error( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() class RelayPolicyError(Exception): @@ -1421,7 +1848,11 @@ def test_nemo_relay_adaptive_llm_execution_keeps_relay_translated_error(tmp_path def translated_execute(name, request, func, **kwargs): try: - return func(_FakeLLMRequest(request.headers, {"intercepted": True, **request.content})) + return func( + _FakeLLMRequest( + request.headers, {"intercepted": True, **request.content} + ) + ) except Exception: raise relay_error @@ -1446,7 +1877,9 @@ def test_nemo_relay_adaptive_llm_execution_keeps_relay_translated_error(tmp_path assert caught.value is relay_error -def test_nemo_relay_downstream_unwrap_matches_real_middleware_wrapper_shape(monkeypatch): +def test_nemo_relay_downstream_unwrap_matches_real_middleware_wrapper_shape( + monkeypatch, +): # Regression guard against core/plugin drift. The synthetic tests above model # the downstream-error wrapper with a local class, so they keep passing even # if core middleware renames its private ``_DownstreamExecutionError`` or drops @@ -1510,7 +1943,9 @@ def _adaptive_llm_execute_mode(tmp_path, monkeypatch, plugins_toml_text: str) -> next_call=lambda request: {"raw": request}, ) - execute_start = next(event for event in fake.events if event[0] == "llm.execute.start") + execute_start = next( + event for event in fake.events if event[0] == "llm.execute.start" + ) return execute_start[3]["data"]["mode"] @@ -1534,7 +1969,9 @@ version = 1 assert mode == "observe_only" -def test_nemo_relay_adaptive_llm_execution_middleware_accepts_legacy_top_level_mode(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_llm_execution_middleware_accepts_legacy_top_level_mode( + tmp_path, monkeypatch +): mode = _adaptive_llm_execute_mode( tmp_path, monkeypatch, @@ -1552,7 +1989,9 @@ mode = "route" assert mode == "route" -def test_nemo_relay_adaptive_llm_execution_middleware_prefers_tool_parallelism_mode(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_llm_execution_middleware_prefers_tool_parallelism_mode( + tmp_path, monkeypatch +): mode = _adaptive_llm_execute_mode( tmp_path, monkeypatch, @@ -1573,7 +2012,9 @@ mode = "schedule" assert mode == "schedule" -def test_nemo_relay_llm_execution_middleware_calls_through_without_adaptive(monkeypatch): +def test_nemo_relay_llm_execution_middleware_calls_through_without_adaptive( + monkeypatch, +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) @@ -1589,7 +2030,9 @@ def test_nemo_relay_llm_execution_middleware_calls_through_without_adaptive(monk assert not any(event[0] == "llm.execute.start" for event in fake.events) -def test_nemo_relay_adaptive_tool_execution_middleware_preserves_raw_response(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_tool_execution_middleware_preserves_raw_response( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) plugins_toml = tmp_path / "plugins.toml" @@ -1627,12 +2070,16 @@ mode = "observe_only" assert response == {"raw": True, "args": {"command": "pwd", "intercepted": True}} assert seen_args["intercepted"] is True - execute_start = next(event for event in fake.events if event[0] == "tool.execute.start") + execute_start = next( + event for event in fake.events if event[0] == "tool.execute.start" + ) assert execute_start[3]["data"]["mode"] == "observe_only" assert execute_start[3]["data"]["tool_call_id"] == "tool-1" -def test_nemo_relay_adaptive_tool_execution_preserves_downstream_error(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_tool_execution_preserves_downstream_error( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() def native_like_execute(name, args, func, **kwargs): @@ -1666,7 +2113,9 @@ def test_nemo_relay_adaptive_tool_execution_preserves_downstream_error(tmp_path, assert caught.value.status_code == 403 -def test_nemo_relay_adaptive_tool_execution_keeps_unrelated_internal_error(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_tool_execution_keeps_unrelated_internal_error( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() relay_error = RuntimeError("internal error: relay setup failed") @@ -1693,7 +2142,9 @@ def test_nemo_relay_adaptive_tool_execution_keeps_wrapped_relay_error_after_down tmp_path, monkeypatch ): fake = _FakeNemoRelay() - relay_error = RuntimeError("internal error: RuntimeError: relay policy blocked after downstream") + relay_error = RuntimeError( + "internal error: RuntimeError: relay policy blocked after downstream" + ) def translated_execute(name, args, func, **kwargs): try: @@ -1719,7 +2170,9 @@ def test_nemo_relay_adaptive_tool_execution_keeps_wrapped_relay_error_after_down assert caught.value is relay_error -def test_nemo_relay_adaptive_tool_execution_keeps_relay_translated_error(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_tool_execution_keeps_relay_translated_error( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() class RelayPolicyError(Exception): @@ -1753,7 +2206,9 @@ def test_nemo_relay_adaptive_tool_execution_keeps_relay_translated_error(tmp_pat assert caught.value is relay_error -def test_nemo_relay_tool_execution_middleware_calls_through_without_adaptive(monkeypatch): +def test_nemo_relay_tool_execution_middleware_calls_through_without_adaptive( + monkeypatch, +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) @@ -1768,7 +2223,9 @@ def test_nemo_relay_tool_execution_middleware_calls_through_without_adaptive(mon assert not any(event[0] == "tool.execute.start" for event in fake.events) -def test_nemo_relay_adaptive_execution_skips_duplicate_observer_spans(tmp_path, monkeypatch): +def test_nemo_relay_adaptive_execution_skips_duplicate_observer_spans( + tmp_path, monkeypatch +): fake = _FakeNemoRelay() plugin = _fresh_plugin(monkeypatch, fake) plugins_toml = tmp_path / "plugins.toml" @@ -1800,8 +2257,12 @@ mode = "observe_only" request={"body": {"messages": [{"role": "user", "content": "hi"}]}}, ) plugin.on_post_api_request(**base, response={"ok": True}) - plugin.on_pre_tool_call(**base, tool_name="terminal", tool_call_id="tool-1", args={"command": "pwd"}) - plugin.on_post_tool_call(**base, tool_name="terminal", tool_call_id="tool-1", result={"ok": True}) + plugin.on_pre_tool_call( + **base, tool_name="terminal", tool_call_id="tool-1", args={"command": "pwd"} + ) + plugin.on_post_tool_call( + **base, tool_name="terminal", tool_call_id="tool-1", result={"ok": True} + ) plugin.on_llm_execution_middleware( **base, diff --git a/tests/plugins/test_nemo_relay_shared_metrics.py b/tests/plugins/test_nemo_relay_shared_metrics.py new file mode 100644 index 000000000000..5c2c1a87b814 --- /dev/null +++ b/tests/plugins/test_nemo_relay_shared_metrics.py @@ -0,0 +1,462 @@ +"""Focused tests for the Hermes shared-metrics durable store.""" + +from __future__ import annotations + +import json +import multiprocessing as mp +import os +import sqlite3 +import stat +import uuid +from concurrent.futures import ThreadPoolExecutor +from copy import deepcopy +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest +from plugins.observability.nemo_relay.shared_metrics import SharedMetricsStore +from plugins.observability.nemo_relay.shared_metrics_contract import ( + MODEL_FAMILIES, + MODEL_LOCALITIES, + MODEL_OUTCOMES, + PRIMARY_MODEL_CALL_ROLE, + PROVIDER_FAMILIES, + execution_surface, + model_call_outcome, + model_call_dimensions, + model_family, + model_locality, + provider_family, +) + + +SCHEMA_PATH = ( + Path(__file__).resolve().parents[2] + / "plugins" + / "observability" + / "nemo_relay" + / "schemas" + / "hermes.shared_metrics.v1.schema.json" +) + + +def _schema_validator(): + jsonschema = pytest.importorskip("jsonschema") + schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) + jsonschema.Draft202012Validator.check_schema(schema) + return jsonschema.Draft202012Validator( + schema, + format_checker=jsonschema.FormatChecker(), + ) + + +def _package_dimension_schema() -> dict[str, object]: + schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) + return schema["$defs"]["model_call_counter"]["properties"]["dimensions"] + + +def _dimensions() -> dict[str, str]: + return { + "call_role": PRIMARY_MODEL_CALL_ROLE, + "locality": "remote", + "model_family": "claude", + "outcome": "success", + "provider_family": "direct", + } + + +def _record_model_calls_in_process( + database_path: str, + outbox_directory: str, + count: int, + start_barrier: Any | None = None, +) -> None: + if start_barrier is not None: + start_barrier.wait() + store = SharedMetricsStore(Path(database_path), Path(outbox_directory)) + for _ in range(count): + store.record_model_call(_dimensions(), "test-version") + + +def test_model_call_counter_survives_restart_and_exports_only_new_deltas(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") + store.record_model_call(_dimensions(), "test-version") + + first_paths = store.create_and_export_package() + + assert len(first_paths) == 1 + first_package = json.loads(first_paths[0].read_text(encoding="utf-8")) + _schema_validator().validate(first_package) + uuid.UUID(first_package["package_id"]) + uuid.UUID(first_package["install_id"]) + assert first_package["schema_version"] == "hermes.shared_metrics.v1" + assert first_package["resource"] == {"hermes_version": "test-version"} + assert first_package["metrics"] == [ + { + "name": "hermes.model_call.count", + "type": "counter", + "dimensions": _dimensions(), + "value": 2, + } + ] + + restarted = SharedMetricsStore(database_path, outbox_directory) + assert restarted.counter_snapshot()[0]["value"] == 2 + assert restarted.counter_snapshot()[0]["packaged_value"] == 2 + assert restarted.create_and_export_package() == [] + assert len(list(outbox_directory.glob("*.json"))) == 1 + + restarted.record_model_call(_dimensions(), "test-version") + second_paths = restarted.create_and_export_package() + + assert len(second_paths) == 1 + second_package = json.loads(second_paths[0].read_text(encoding="utf-8")) + assert second_package["package_id"] != first_package["package_id"] + assert second_package["install_id"] == first_package["install_id"] + assert second_package["metrics"][0]["value"] == 1 + assert restarted.counter_snapshot()[0]["value"] == 3 + assert restarted.counter_snapshot()[0]["packaged_value"] == 3 + + +def test_package_schema_matches_the_model_call_contract(): + properties = _package_dimension_schema()["properties"] + + assert properties["call_role"] == {"const": PRIMARY_MODEL_CALL_ROLE} + assert set(properties["locality"]["enum"]) == MODEL_LOCALITIES + assert set(properties["model_family"]["enum"]) == MODEL_FAMILIES + assert set(properties["outcome"]["enum"]) == MODEL_OUTCOMES + assert set(properties["provider_family"]["enum"]) == PROVIDER_FAMILIES + + +@pytest.mark.parametrize( + ("provider", "expected"), + [ + ("", "unknown"), + ("not-a-hermes-provider", "unknown"), + ("custom", "custom"), + ("custom-local", "custom"), + ("custom:private-endpoint", "custom"), + ("lmstudio", "local"), + ("lm_studio", "local"), + ("ollama", "local"), + ("nous", "aggregator"), + ("openrouter", "aggregator"), + ("kilo", "aggregator"), + ("copilot-acp", "aggregator"), + ("huggingface", "aggregator"), + ("novita", "aggregator"), + ("anthropic", "direct"), + ("google", "direct"), + ("openai-api", "direct"), + ], +) +def test_provider_family_uses_bounded_product_categories(provider, expected): + assert provider_family({"provider": provider}) == expected + + +def test_provider_family_does_not_resolve_live_provider_metadata(monkeypatch): + def fail_live_lookup(_provider): + raise AssertionError("telemetry must not refresh provider metadata") + + monkeypatch.setattr("hermes_cli.providers.get_provider", fail_live_lookup) + assert provider_family({"provider": "anthropic"}) == "direct" + + +def test_locality_uses_the_endpoint_only_for_local_classification(): + kwargs = { + "provider": "custom", + "base_url": "http://127.0.0.1:11434/v1", + } + + assert provider_family(kwargs) == "custom" + assert model_locality(kwargs) == "local" + + +@pytest.mark.parametrize( + ("model", "expected"), + [ + ("google/gemma-3", "gemma"), + ("x-ai/grok-4", "grok"), + ("minimax/minimax-m2.5", "minimax"), + ("xiaomi/mimo-v2", "mimo"), + ("amazon/nova-pro", "nova"), + ("stepfun/step-3.5", "step"), + ("arcee-ai/trinity-large", "trinity"), + ], +) +def test_model_family_covers_families_evidenced_by_the_hermes_catalog(model, expected): + assert model_family({"model": model}) == expected + + +@pytest.mark.parametrize( + "model", + [ + "private-gptish-model", + "innovation-private", + "mimosa-private", + "stepstone-private", + "supernova-private", + ], +) +def test_model_family_requires_identifier_boundaries(model): + assert model_family({"model": model}) == "unknown" + + +def test_model_family_accepts_only_allowlisted_declared_metadata(): + assert model_family({"model": "private", "model_family": "qwen"}) == "qwen" + assert model_family({"model": "private", "model_family": "private"}) == "unknown" + + +@pytest.mark.parametrize( + ("platform", "expected"), + [ + ("", "unknown"), + ("cli", "cli"), + ("api_server", "api"), + ("cron", "scheduled_task"), + ("whatsapp_cloud", "gateway"), + ("private-surface", "other"), + ], +) +def test_execution_surface_uses_the_hermes_platform_registry(platform, expected): + assert execution_surface({"platform": platform}) == expected + + +def test_model_outcome_fails_closed_to_a_bounded_value(): + assert model_call_outcome({"outcome": "private"}) == "failed" + + +def test_unlisted_model_collapses_to_a_bounded_value(): + assert model_family({"model": "private-model-name"}) == "unknown" + + +def test_subscriber_contract_rejects_unknown_fields_and_dimension_values(): + event = SimpleNamespace( + kind="scope", + category="llm", + category_profile={"model_name": "gpt"}, + name="hermes.model_call", + scope_category="end", + metadata={"hermes.metrics.schema_version": "hermes.metrics.event.v1"}, + data={ + "call_role": "primary", + "locality": "remote", + "outcome": "success", + "provider_family": "direct", + }, + ) + + assert model_call_dimensions(event) == { + "call_role": "primary", + "locality": "remote", + "model_family": "gpt", + "outcome": "success", + "provider_family": "direct", + } + event.category_profile["model_name"] = "private-model-name" + assert model_call_dimensions(event) is None + event.category_profile["model_name"] = "gpt" + event.data["prompt"] = "must-not-pass" + assert model_call_dimensions(event) is None + event.data.pop("prompt") + event.metadata["prompt"] = "must-not-pass" + assert model_call_dimensions(event) is None + event.metadata.pop("prompt") + event.category_profile["private"] = "must-not-pass" + assert model_call_dimensions(event) is None + event.category_profile.pop("private") + event.category = "function" + assert model_call_dimensions(event) 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: + connection.execute( + "CREATE TABLE telemetry_state (key TEXT PRIMARY KEY, value TEXT NOT NULL)" + ) + connection.execute( + "INSERT INTO telemetry_state(key, value) VALUES ('schema_version', '999')" + ) + + with pytest.raises(RuntimeError, match="Unsupported shared-metrics store schema"): + SharedMetricsStore(database_path, tmp_path / "outbox") + + with sqlite3.connect(database_path) as connection: + [schema_version] = connection.execute( + "SELECT value FROM telemetry_state WHERE key = 'schema_version'" + ).fetchone() + assert schema_version == "999" + + +def test_pending_metrics_keep_the_version_recorded_at_event_time(tmp_path): + store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox") + store.record_model_call(_dimensions(), "version-a") + store.record_model_call(_dimensions(), "version-b") + + packages = [ + json.loads(path.read_text(encoding="utf-8")) + for path in store.create_and_export_package() + ] + + assert {package["resource"]["hermes_version"] for package in packages} == { + "version-a", + "version-b", + } + assert all(package["metrics"][0]["value"] == 1 for package in packages) + + +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") + [package_path] = store.create_and_export_package() + package = json.loads(package_path.read_text(encoding="utf-8")) + invalid_package = deepcopy(package) + invalid_package["prompt"] = "must-not-be-accepted" + + jsonschema = pytest.importorskip("jsonschema") + with pytest.raises(jsonschema.ValidationError): + _schema_validator().validate(invalid_package) + + +def test_pending_package_retry_reuses_the_same_package_and_file(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") + [package_path] = store.create_and_export_package() + original_payload = package_path.read_bytes() + + with sqlite3.connect(database_path) as connection: + connection.execute("UPDATE package_outbox SET exported_at = NULL") + + restarted = SharedMetricsStore(database_path, outbox_directory) + assert restarted.create_and_export_package() == [package_path] + assert package_path.read_bytes() == original_payload + assert list(outbox_directory.glob("*.json")) == [package_path] + + +def test_file_export_failure_retries_committed_outbox_without_duplicate_delta( + tmp_path, monkeypatch +): + database_path = tmp_path / "metrics.sqlite3" + outbox_directory = tmp_path / "outbox" + store = SharedMetricsStore(database_path, outbox_directory) + store.record_model_call(_dimensions(), "test-version") + + def fail_write(*_args, **_kwargs): + raise OSError("simulated atomic export failure") + + module_globals = SharedMetricsStore._export_pending_packages.__globals__ + original_write = module_globals["atomic_json_write"] + monkeypatch.setitem(module_globals, "atomic_json_write", fail_write) + with pytest.raises(OSError, match="simulated atomic export failure"): + store.create_and_export_package() + + with sqlite3.connect(database_path) as connection: + package_id, exported_at = connection.execute( + "SELECT package_id, exported_at FROM package_outbox" + ).fetchone() + assert exported_at is None + assert store.counter_snapshot()[0]["packaged_value"] == 1 + assert list(outbox_directory.glob("*.json")) == [] + + monkeypatch.setitem(module_globals, "atomic_json_write", original_write) + assert store.create_and_export_package() == [ + outbox_directory / f"{package_id}.json" + ] + assert len(list(outbox_directory.glob("*.json"))) == 1 + assert store.create_and_export_package() == [] + + +def test_package_export_does_not_chase_concurrent_updates(tmp_path, monkeypatch): + database_path = tmp_path / "metrics.sqlite3" + outbox_directory = tmp_path / "outbox" + store = SharedMetricsStore(database_path, outbox_directory) + store.record_model_call(_dimensions(), "test-version") + original_create = store._create_package + create_calls = 0 + + def create_and_record_another(): + nonlocal create_calls + create_calls += 1 + package = original_create() + if create_calls == 1: + store.record_model_call(_dimensions(), "test-version") + return package + + monkeypatch.setattr(store, "_create_package", create_and_record_another) + first_paths = store.create_and_export_package() + + assert create_calls == 1 + assert len(first_paths) == 1 + [counter] = store.counter_snapshot() + assert counter["metric_name"] == "hermes.model_call.count" + assert counter["dimensions"] == _dimensions() + assert counter["value"] == 2 + assert counter["packaged_value"] == 1 + + second_paths = store.create_and_export_package() + assert len(second_paths) == 1 + assert store.counter_snapshot()[0]["packaged_value"] == 2 + + +def test_concurrent_model_call_updates_are_transactional(tmp_path): + database_path = tmp_path / "metrics.sqlite3" + outbox_directory = tmp_path / "outbox" + SharedMetricsStore(database_path, outbox_directory) + + def record_calls(count: int) -> None: + store = SharedMetricsStore(database_path, outbox_directory) + for _ in range(count): + store.record_model_call(_dimensions(), "test-version") + + with ThreadPoolExecutor(max_workers=2) as executor: + futures = [executor.submit(record_calls, 10) for _ in range(2)] + for future in futures: + future.result() + + restarted = SharedMetricsStore(database_path, outbox_directory) + assert restarted.counter_snapshot()[0]["value"] == 20 + + +def test_cross_process_model_call_updates_are_transactional(tmp_path): + database_path = tmp_path / "metrics.sqlite3" + outbox_directory = tmp_path / "outbox" + context = mp.get_context("spawn") + start_barrier = context.Barrier(2) + processes = [ + context.Process( + target=_record_model_calls_in_process, + args=(str(database_path), str(outbox_directory), 10, start_barrier), + ) + for _ in range(2) + ] + + for process in processes: + process.start() + for process in processes: + process.join(timeout=15) + assert not process.is_alive() + assert process.exitcode == 0 + + restarted = SharedMetricsStore(database_path, outbox_directory) + assert restarted.counter_snapshot()[0]["value"] == 20 + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission modes are unavailable") +def test_store_and_export_are_owner_only(tmp_path): + database_path = tmp_path / "private-store" / "metrics.sqlite3" + outbox_directory = tmp_path / "private-outbox" + store = SharedMetricsStore(database_path, outbox_directory) + store.record_model_call(_dimensions(), "test-version") + [package_path] = store.create_and_export_package() + + assert stat.S_IMODE(database_path.parent.stat().st_mode) == 0o700 + assert stat.S_IMODE(outbox_directory.stat().st_mode) == 0o700 + assert stat.S_IMODE(database_path.stat().st_mode) == 0o600 + assert stat.S_IMODE(package_path.stat().st_mode) == 0o600 diff --git a/uv.lock b/uv.lock index 59f9d8e2628b..f0daf9798e5a 100644 --- a/uv.lock +++ b/uv.lock @@ -1804,7 +1804,7 @@ requires-dist = [ { name = "microsoft-teams-apps", marker = "extra == 'teams'", specifier = "==2.0.13.4" }, { name = "mistralai", marker = "extra == 'mistral'", specifier = "==2.4.8" }, { name = "modal", marker = "extra == 'modal'", specifier = "==1.3.4" }, - { name = "nemo-relay", marker = "extra == 'nemo-relay'", specifier = ">=0.5,<1.0" }, + { name = "nemo-relay", marker = "extra == 'nemo-relay'", specifier = ">=0.5.0,<0.6.0" }, { name = "numpy", marker = "extra == 'voice'", specifier = "==2.4.3" }, { name = "openai", specifier = "==2.24.0" }, { name = "packaging", specifier = "==26.0" },