mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-29 18:46:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

feat(observability): add Relay shared metrics pipeline

Browse source 
Signed-off-by: Alex Fournier <afournier@nvidia.com>
This commit is contained in:
Alex Fournier 2026-07-19 08:54:27 -04:00
parent 36f2a966c7
commit 3bd338d2a9
11 changed files with 2824 additions and 205 deletions
Download patch file Download diff file Expand all files Collapse all files

74
plugins/observability/nemo_relay/README.md
View file

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

689
plugins/observability/nemo_relay/__init__.py
View file

File diff suppressed because it is too large Load diff

153
plugins/observability/nemo_relay/schemas/hermes.shared_metrics.v1.schema.json Normal file
View file

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

376
plugins/observability/nemo_relay/shared_metrics.py Normal file
View file

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

251
plugins/observability/nemo_relay/shared_metrics_contract.py Normal file
View file

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

31
plugins/observability/nemo_relay/shared_metrics_subscriber.py Normal file
View file

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

3
pyproject.toml
View file

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

379
scripts/smoke_nemo_relay_shared_metrics.py Normal file
View file

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

609
tests/plugins/test_nemo_relay_plugin.py
View file

File diff suppressed because it is too large Load diff

462
tests/plugins/test_nemo_relay_shared_metrics.py Normal file
View file

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

2
uv.lock generated
View file

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