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

feat(observability): add Relay client resource metrics 

Signed-off-by: Alex Fournier <afournier@nvidia.com>
This commit is contained in:
Alex Fournier 2026-07-29 10:08:40 -07:00
parent f1fd678e44
commit 5607d09e03
7 changed files with 532 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -123,9 +123,13 @@ $HERMES_HOME/telemetry/shared_metrics/outbox/*.json
The database keeps transactional aggregate and package-outbox state. Package
files are immutable delta documents that conform to a closed JSON schema and
are written with atomic replacement. Fully packaged aggregate rows and
successfully exported package rows and files are retained locally for 30 days.
Pending package rows and counters with unexported deltas are never pruned.
are written with atomic replacement. Each package records the Hermes version,
OS family, architecture, and install method as bounded client resources.
Unrecognized platform or installation values are exported as `unknown`; raw
platform strings, hostnames, and paths are never included. Fully packaged
aggregate rows and successfully exported package rows and files are retained
locally for 30 days. Pending package rows and counters with unexported deltas
are never pruned.
Each package contains an `install_id` generated as a random UUID. Despite the
schema field name, its current scope is one `HERMES_HOME`, so it is more

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

@ -48,6 +48,18 @@
"type": "string",
"minLength": 1,
"maxLength": 64
},
"os_family": {
"type": "string",
"enum": ["linux", "macos", "unknown", "windows"]
},
"architecture": {
"type": "string",
"enum": ["arm", "arm64", "unknown", "x86", "x86_64"]
},
"install_method": {
"type": "string",
"enum": ["docker", "git", "homebrew", "nixos", "pip", "unknown"]
}
}
},

190
hermes_cli/observability/shared_metrics.py
View File

@ -19,12 +19,13 @@ from utils import atomic_json_write
from .shared_metrics_contract import (
COUNTER_METRICS,
MODEL_CALL_METRIC,
client_resource_is_valid,
counter_dimensions_are_valid,
)
_PACKAGE_SCHEMA_VERSION = "hermes.shared_metrics.v1"
_STORE_SCHEMA_VERSION = "1"
_STORE_SCHEMA_VERSION = "2"
_BUSY_TIMEOUT_MS = 250
_SCHEMA_BUSY_TIMEOUT_MS = 5_000
_LOCAL_HISTORY_RETENTION_DAYS = 30
@ -59,22 +60,24 @@ class SharedMetricsStore:
def record_model_call(
self,
dimensions: dict[str, str],
hermes_version: str,
resource: dict[str, str],
) -> None:
"""Increment the terminal model-call counter for the current UTC day."""
self.record_counter(MODEL_CALL_METRIC, dimensions, hermes_version)
self.record_counter(MODEL_CALL_METRIC, dimensions, resource)
def record_counter(
self,
metric_name: str,
dimensions: dict[str, str],
hermes_version: str,
resource: dict[str, str],
) -> None:
"""Increment one allowlisted counter for the current UTC day."""
if metric_name not in COUNTER_METRICS:
raise ValueError(f"Unsupported shared metric: {metric_name}")
if not counter_dimensions_are_valid(metric_name, dimensions):
raise ValueError(f"Unsupported dimensions for shared metric: {metric_name}")
if not client_resource_is_valid(resource):
raise ValueError("Unsupported shared-metrics client resource")
dimensions_json = json.dumps(
dimensions,
sort_keys=True,
@ -88,14 +91,20 @@ class SharedMetricsStore:
period_start,
metric_name,
hermes_version,
os_family,
architecture,
install_method,
dimensions_json,
value,
packaged_value
) VALUES (?, ?, ?, ?, 1, 0)
) VALUES (?, ?, ?, ?, ?, ?, ?, 1, 0)
ON CONFLICT(
period_start,
metric_name,
hermes_version,
os_family,
architecture,
install_method,
dimensions_json
)
DO UPDATE SET value = value + 1
@ -103,7 +112,10 @@ class SharedMetricsStore:
(
period_start,
metric_name,
hermes_version or "unknown",
resource["hermes_version"],
resource["os_family"],
resource["architecture"],
resource["install_method"],
dimensions_json,
),
)
@ -141,18 +153,33 @@ class SharedMetricsStore:
period_start,
metric_name,
hermes_version,
os_family,
architecture,
install_method,
dimensions_json,
value,
packaged_value
FROM counter_aggregates
ORDER BY period_start, hermes_version, metric_name, dimensions_json
ORDER BY
period_start,
hermes_version,
os_family,
architecture,
install_method,
metric_name,
dimensions_json
"""
).fetchall()
return [
{
"period_start": row["period_start"],
"metric_name": row["metric_name"],
"hermes_version": row["hermes_version"],
"resource": {
"hermes_version": row["hermes_version"],
"os_family": row["os_family"],
"architecture": row["architecture"],
"install_method": row["install_method"],
},
"dimensions": json.loads(row["dimensions_json"]),
"value": row["value"],
"packaged_value": row["packaged_value"],
@ -213,29 +240,15 @@ class SharedMetricsStore:
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:
schema_version = str(schema_row["value"]) if schema_row is not None else None
if schema_version == "1":
SharedMetricsStore._migrate_v1_counter_aggregates(connection)
schema_version = _STORE_SCHEMA_VERSION
if schema_version is not None and schema_version != _STORE_SCHEMA_VERSION:
raise RuntimeError(
"Unsupported shared-metrics store schema version: "
f"{schema_row['value']}"
f"Unsupported shared-metrics store schema version: {schema_version}"
)
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
)
)
"""
)
SharedMetricsStore._create_counter_aggregates_table(connection)
connection.execute(
"""
CREATE TABLE IF NOT EXISTS package_outbox (
@ -250,12 +263,74 @@ class SharedMetricsStore:
)
connection.execute(
"""
INSERT OR IGNORE INTO telemetry_state(key, value)
INSERT INTO telemetry_state(key, value)
VALUES ('schema_version', ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value
""",
(_STORE_SCHEMA_VERSION,),
)
@staticmethod
def _create_counter_aggregates_table(connection: sqlite3.Connection) -> None:
connection.execute(
"""
CREATE TABLE IF NOT EXISTS counter_aggregates (
period_start TEXT NOT NULL,
metric_name TEXT NOT NULL,
hermes_version TEXT NOT NULL,
os_family TEXT NOT NULL,
architecture TEXT NOT NULL,
install_method 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,
os_family,
architecture,
install_method,
dimensions_json
)
)
"""
)
@staticmethod
def _migrate_v1_counter_aggregates(connection: sqlite3.Connection) -> None:
connection.execute(
"ALTER TABLE counter_aggregates RENAME TO counter_aggregates_v1"
)
SharedMetricsStore._create_counter_aggregates_table(connection)
connection.execute(
"""
INSERT INTO counter_aggregates(
period_start,
metric_name,
hermes_version,
os_family,
architecture,
install_method,
dimensions_json,
value,
packaged_value
)
SELECT
period_start,
metric_name,
hermes_version,
'unknown',
'unknown',
'unknown',
dimensions_json,
value,
packaged_value
FROM counter_aggregates_v1
"""
)
connection.execute("DROP TABLE counter_aggregates_v1")
def _install_id(self, connection: sqlite3.Connection) -> str:
row = connection.execute(
"SELECT value FROM telemetry_state WHERE key = 'install_id'"
@ -280,10 +355,20 @@ class SharedMetricsStore:
"""
SELECT COUNT(*) AS period_count
FROM (
SELECT period_start, hermes_version
SELECT
period_start,
hermes_version,
os_family,
architecture,
install_method
FROM counter_aggregates
WHERE value > packaged_value
GROUP BY period_start, hermes_version
GROUP BY
period_start,
hermes_version,
os_family,
architecture,
install_method
)
"""
).fetchone()
@ -322,10 +407,20 @@ class SharedMetricsStore:
) -> dict[str, Any] | None:
period_row = connection.execute(
"""
SELECT period_start, hermes_version
SELECT
period_start,
hermes_version,
os_family,
architecture,
install_method
FROM counter_aggregates
WHERE value > packaged_value
ORDER BY period_start, hermes_version
ORDER BY
period_start,
hermes_version,
os_family,
architecture,
install_method
LIMIT 1
"""
).fetchone()
@ -339,16 +434,33 @@ class SharedMetricsStore:
FROM counter_aggregates
WHERE period_start = ?
AND hermes_version = ?
AND os_family = ?
AND architecture = ?
AND install_method = ?
AND value > packaged_value
ORDER BY metric_name, dimensions_json
""",
(period_value, period_row["hermes_version"]),
(
period_value,
period_row["hermes_version"],
period_row["os_family"],
period_row["architecture"],
period_row["install_method"],
),
).fetchall()
period_start = datetime.fromisoformat(str(period_value)).replace(
tzinfo=timezone.utc
)
period_end = period_start + timedelta(days=1)
package_id = str(uuid.uuid4())
resource = {
"hermes_version": period_row["hermes_version"],
"os_family": period_row["os_family"],
"architecture": period_row["architecture"],
"install_method": period_row["install_method"],
}
if not client_resource_is_valid(resource):
raise ValueError("Unsupported shared-metrics client resource")
payload = {
"schema_version": _PACKAGE_SCHEMA_VERSION,
"package_id": package_id,
@ -356,7 +468,7 @@ class SharedMetricsStore:
"period_start": _isoformat(period_start),
"period_end": _isoformat(period_end),
"generated_at": _isoformat(now),
"resource": {"hermes_version": period_row["hermes_version"]},
"resource": resource,
"metrics": [self._package_metric(row) for row in rows],
}
payload_json = json.dumps(
@ -390,12 +502,18 @@ class SharedMetricsStore:
WHERE period_start = ?
AND metric_name = ?
AND hermes_version = ?
AND os_family = ?
AND architecture = ?
AND install_method = ?
AND dimensions_json = ?
""",
(
period_value,
row["metric_name"],
period_row["hermes_version"],
period_row["os_family"],
period_row["architecture"],
period_row["install_method"],
row["dimensions_json"],
),
)

91
hermes_cli/observability/shared_metrics_contract.py
View File

@ -172,6 +172,97 @@ SKILL_POST_PATCH_STATES: frozenset[str] = frozenset({
"not_applicable",
"reused_after_patch",
})
CLIENT_OS_FAMILIES: frozenset[str] = frozenset({
"linux",
"macos",
"unknown",
"windows",
})
CLIENT_ARCHITECTURES: frozenset[str] = frozenset({
"arm",
"arm64",
"unknown",
"x86",
"x86_64",
})
CLIENT_INSTALL_METHODS: frozenset[str] = frozenset({
"docker",
"git",
"homebrew",
"nixos",
"pip",
"unknown",
})
CLIENT_RESOURCE_KEYS: frozenset[str] = frozenset({
"architecture",
"hermes_version",
"install_method",
"os_family",
})
def client_os_family(value: Any) -> str:
"""Map a platform system name to the shared-metrics OS taxonomy."""
normalized = str(value or "").strip().lower()
return {
"darwin": "macos",
"linux": "linux",
"macos": "macos",
"windows": "windows",
}.get(normalized, "unknown")
def client_architecture(value: Any) -> str:
"""Map a machine architecture to the shared-metrics taxonomy."""
normalized = str(value or "").strip().lower().replace("-", "_")
if normalized in {"amd64", "x64", "x86_64"}:
return "x86_64"
if normalized in {"aarch64", "arm64"}:
return "arm64"
if normalized in {"i386", "i486", "i586", "i686", "x86"}:
return "x86"
if normalized.startswith("armv"):
return "arm"
return "unknown"
def client_install_method(value: Any) -> str:
"""Return an allowlisted Hermes installation method."""
normalized = str(value or "").strip().lower()
return normalized if normalized in CLIENT_INSTALL_METHODS else "unknown"
def client_resource(
hermes_version: Any,
*,
os_name: Any,
architecture: Any,
install_method: Any,
) -> dict[str, str]:
"""Build the bounded client resource attached to aggregate packages."""
normalized_version = str(hermes_version or "").strip()
if not normalized_version or len(normalized_version) > 64:
normalized_version = "unknown"
return {
"architecture": client_architecture(architecture),
"hermes_version": normalized_version,
"install_method": client_install_method(install_method),
"os_family": client_os_family(os_name),
}
def client_resource_is_valid(resource: Any) -> bool:
"""Return whether a package resource exactly matches the bounded contract."""
if not isinstance(resource, dict) or set(resource) != CLIENT_RESOURCE_KEYS:
return False
version = resource.get("hermes_version")
return (
isinstance(version, str)
and 0 < len(version) <= 64
and resource.get("os_family") in CLIENT_OS_FAMILIES
and resource.get("architecture") in CLIENT_ARCHITECTURES
and resource.get("install_method") in CLIENT_INSTALL_METHODS
)
_COUNTER_DIMENSION_VALUES: dict[str, dict[str, frozenset[str]]] = {
TASK_STARTED_METRIC: {

12
hermes_cli/observability/shared_metrics_subscriber.py
View File

@ -3,15 +3,18 @@
from __future__ import annotations
import logging
import platform
import threading
from typing import Any
from agent.relay_runtime import RUNTIME_INSTANCE_KEY
from hermes_cli.config import detect_install_method
from .shared_metrics import SharedMetricsStore
from .shared_metrics_contract import (
MODEL_CALL_METRIC,
TOOL_CALL_METRIC,
client_resource,
model_call_dimensions,
skill_counter,
task_counter,
@ -33,7 +36,12 @@ class SharedMetricsSubscriber:
runtime_id: str | None = None,
) -> None:
self.store = store
self._hermes_version = hermes_version or "unknown"
self._client_resource = client_resource(
hermes_version,
os_name=platform.system(),
architecture=platform.machine(),
install_method=detect_install_method(),
)
self._runtime_id = runtime_id
self._active = True
self._lock = threading.RLock()
@ -72,7 +80,7 @@ class SharedMetricsSubscriber:
self.store.record_counter(
metric_name,
dimensions,
self._hermes_version,
self._client_resource,
)
except Exception:
logger.warning(

7
scripts/smoke_nemo_relay_shared_metrics.py
View File

@ -429,6 +429,13 @@ def _validate_packages(
]
for package in packages:
jsonschema.validate(package, schema)
if set(package["resource"]) != {
"architecture",
"hermes_version",
"install_method",
"os_family",
}:
raise AssertionError(f"Unexpected client resource: {package['resource']}")
serialized = json.dumps(packages)
for prohibited in (

284
tests/hermes_cli/test_relay_shared_metrics.py
View File

@ -21,6 +21,9 @@ import pytest
from hermes_cli.observability import shared_metrics as shared_metrics_module
from hermes_cli.observability.shared_metrics import SharedMetricsStore
from hermes_cli.observability.shared_metrics_contract import (
CLIENT_ARCHITECTURES,
CLIENT_INSTALL_METHODS,
CLIENT_OS_FAMILIES,
COUNT_BUCKETS,
DURATION_BUCKETS,
EXECUTION_SURFACES,
@ -41,6 +44,10 @@ from hermes_cli.observability.shared_metrics_contract import (
TOOL_LATENCY_BUCKETS,
TOOL_OUTCOMES,
TOOL_RETRY_BUCKETS,
client_architecture,
client_install_method,
client_os_family,
client_resource,
count_bucket,
duration_bucket,
execution_surface,
@ -105,6 +112,21 @@ def _dimensions() -> dict[str, str]:
}
def _resource(
hermes_version: str = "test-version",
*,
os_family: str = "linux",
architecture: str = "x86_64",
install_method: str = "git",
) -> dict[str, str]:
return {
"architecture": architecture,
"hermes_version": hermes_version,
"install_method": install_method,
"os_family": os_family,
}
def _record_model_calls_in_process(
database_path: str,
outbox_directory: str,
@ -115,15 +137,15 @@ def _record_model_calls_in_process(
start_barrier.wait()
store = SharedMetricsStore(Path(database_path), Path(outbox_directory))
for _ in range(count):
store.record_model_call(_dimensions(), "test-version")
store.record_model_call(_dimensions(), _resource())
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")
store.record_model_call(_dimensions(), _resource())
store.record_model_call(_dimensions(), _resource())
first_paths = store.create_and_export_package()
@ -133,7 +155,7 @@ def test_model_call_counter_survives_restart_and_exports_only_new_deltas(tmp_pat
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["resource"] == _resource()
assert first_package["metrics"] == [
{
"name": "hermes.model_call.count",
@ -149,7 +171,7 @@ def test_model_call_counter_survives_restart_and_exports_only_new_deltas(tmp_pat
assert restarted.create_and_export_package() == []
assert len(list(outbox_directory.glob("*.json"))) == 1
restarted.record_model_call(_dimensions(), "test-version")
restarted.record_model_call(_dimensions(), _resource())
second_paths = restarted.create_and_export_package()
assert len(second_paths) == 1
@ -168,32 +190,32 @@ def test_due_export_runs_once_per_utc_day_and_catches_up_pending_deltas(
monkeypatch.setattr(shared_metrics_module, "_utc_now", lambda: current_time)
store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
store.record_model_call(_dimensions(), "test-version")
store.record_model_call(_dimensions(), _resource())
assert len(store.create_and_export_package_if_due()) == 1
current_time = datetime(2026, 7, 28, 18, tzinfo=timezone.utc)
store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
store.record_model_call(_dimensions(), "test-version")
store.record_model_call(_dimensions(), _resource())
assert store.create_and_export_package_if_due() == []
assert len(list((tmp_path / "outbox").glob("*.json"))) == 1
assert store.counter_snapshot()[0] == {
"period_start": "2026-07-28",
"metric_name": "hermes.model_call.count",
"hermes_version": "test-version",
"resource": _resource(),
"dimensions": _dimensions(),
"value": 2,
"packaged_value": 1,
}
current_time = datetime(2026, 7, 29, 9, tzinfo=timezone.utc)
store.record_model_call(_dimensions(), "test-version")
store.record_model_call(_dimensions(), _resource())
assert len(store.create_and_export_package_if_due()) == 2
assert len(list((tmp_path / "outbox").glob("*.json"))) == 3
assert all(
row["value"] == row["packaged_value"] for row in store.counter_snapshot()
)
store.record_model_call(_dimensions(), "test-version")
store.record_model_call(_dimensions(), _resource())
assert store.create_and_export_package_if_due() == []
assert len(list((tmp_path / "outbox").glob("*.json"))) == 3
@ -208,6 +230,48 @@ def test_package_schema_matches_the_model_call_contract():
assert "enum" not in properties["provider"]
def test_client_resource_classification_is_bounded():
assert client_os_family("Darwin") == "macos"
assert client_os_family("Windows") == "windows"
assert client_architecture("AMD64") == "x86_64"
assert client_architecture("aarch64") == "arm64"
assert client_architecture("armv7l") == "arm"
assert client_install_method("Homebrew") == "homebrew"
assert client_resource(
"",
os_name="privacy-os-canary",
architecture="privacy-arch-canary",
install_method="privacy-install-canary",
) == _resource(
"unknown",
os_family="unknown",
architecture="unknown",
install_method="unknown",
)
def test_package_schema_matches_the_client_resource_contract():
schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
resource = schema["properties"]["resource"]
# Existing v1 outbox entries predate the bounded client dimensions. New
# packages always populate every property, while the schema remains able
# to validate those immutable queued payloads.
assert set(resource["required"]) == {"hermes_version"}
assert set(resource["properties"]) == {
"architecture",
"hermes_version",
"install_method",
"os_family",
}
assert set(resource["properties"]["os_family"]["enum"]) == CLIENT_OS_FAMILIES
assert set(resource["properties"]["architecture"]["enum"]) == (CLIENT_ARCHITECTURES)
assert set(resource["properties"]["install_method"]["enum"]) == (
CLIENT_INSTALL_METHODS
)
def test_package_schema_matches_the_task_contract():
schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
start = _task_dimension_schema("task_started_counter")["properties"]
@ -750,29 +814,148 @@ def test_store_rejects_an_unsupported_schema_version(tmp_path):
assert schema_version == "999"
def test_pending_metrics_keep_the_version_recorded_at_event_time(tmp_path):
def test_store_migrates_v1_counters_with_unknown_client_dimensions(tmp_path):
database_path = tmp_path / "metrics.sqlite3"
outbox_directory = tmp_path / "outbox"
install_id = str(uuid.uuid4())
with sqlite3.connect(database_path) as connection:
connection.execute(
"CREATE TABLE telemetry_state (key TEXT PRIMARY KEY, value TEXT NOT NULL)"
)
connection.executemany(
"INSERT INTO telemetry_state(key, value) VALUES (?, ?)",
[("schema_version", "1"), ("install_id", install_id)],
)
connection.execute(
"""
CREATE TABLE 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(
"""
INSERT INTO counter_aggregates(
period_start,
metric_name,
hermes_version,
dimensions_json,
value,
packaged_value
) VALUES (?, ?, ?, ?, ?, ?)
""",
(
"2026-07-21",
"hermes.model_call.count",
"old-version",
json.dumps(_dimensions(), sort_keys=True, separators=(",", ":")),
3,
1,
),
)
store = SharedMetricsStore(database_path, outbox_directory)
[counter] = store.counter_snapshot()
assert counter["resource"] == _resource(
"old-version",
os_family="unknown",
architecture="unknown",
install_method="unknown",
)
assert counter["value"] == 3
assert counter["packaged_value"] == 1
[package_path] = store.create_and_export_package()
package = json.loads(package_path.read_text(encoding="utf-8"))
_schema_validator().validate(package)
assert package["install_id"] == install_id
assert package["metrics"][0]["value"] == 2
def test_pending_metrics_keep_the_client_resource_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")
resource_a = _resource("version-a", architecture="arm64", install_method="pip")
resource_b = _resource("version-a", os_family="macos")
store.record_model_call(_dimensions(), resource_a)
store.record_model_call(_dimensions(), resource_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 {tuple(sorted(package["resource"].items())) for package in packages} == {
tuple(sorted(resource_a.items())),
tuple(sorted(resource_b.items())),
}
assert all(package["metrics"][0]["value"] == 1 for package in packages)
def test_legacy_v1_outbox_package_remains_exportable_and_schema_valid(tmp_path):
store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
package_id = str(uuid.uuid4())
payload = {
"schema_version": "hermes.shared_metrics.v1",
"package_id": package_id,
"install_id": str(uuid.uuid4()),
"period_start": "2026-07-21T00:00:00Z",
"period_end": "2026-07-22T00:00:00Z",
"generated_at": "2026-07-22T00:00:00Z",
"resource": {"hermes_version": "old-version"},
"metrics": [
{
"name": "hermes.model_call.count",
"type": "counter",
"dimensions": _dimensions(),
"value": 1,
}
],
}
with sqlite3.connect(store.database_path) as connection:
connection.execute(
"""
INSERT INTO package_outbox(
package_id,
period_start,
period_end,
payload_json,
created_at,
exported_at
) VALUES (?, ?, ?, ?, ?, NULL)
""",
(
package_id,
payload["period_start"],
payload["period_end"],
json.dumps(payload),
payload["generated_at"],
),
)
[package_path] = store.create_and_export_package()
exported = json.loads(package_path.read_text(encoding="utf-8"))
assert exported == payload
_schema_validator().validate(exported)
def test_store_exports_task_started_and_terminal_counters(tmp_path):
store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
store.record_counter(
"hermes.task_run.started",
{"entrypoint": "interactive", "execution_surface": "cli"},
"test-version",
_resource(),
)
terminal = task_terminal_fields(
{
@ -785,7 +968,7 @@ def test_store_exports_task_started_and_terminal_counters(tmp_path):
tool_call_count=2,
retry_count=0,
)
store.record_counter("hermes.task_run.finished", terminal, "test-version")
store.record_counter("hermes.task_run.finished", terminal, _resource())
[package_path] = store.create_and_export_package()
package = json.loads(package_path.read_text(encoding="utf-8"))
@ -799,7 +982,7 @@ def test_store_exports_task_started_and_terminal_counters(tmp_path):
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")
store.record_model_call(_dimensions(), _resource())
[package_path] = store.create_and_export_package()
package = json.loads(package_path.read_text(encoding="utf-8"))
invalid_package = deepcopy(package)
@ -817,7 +1000,22 @@ def test_store_rejects_dimensions_outside_the_metric_contract(tmp_path):
store.record_counter(
"hermes.model_call.count",
{"prompt": "must-not-be-persisted"},
"test-version",
_resource(),
)
assert store.counter_snapshot() == []
def test_store_rejects_client_resources_outside_the_contract(tmp_path):
store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
with pytest.raises(ValueError, match="Unsupported shared-metrics client resource"):
store.record_model_call(
_dimensions(),
{
**_resource(),
"architecture": "privacy-architecture-canary",
},
)
assert store.counter_snapshot() == []
@ -827,7 +1025,7 @@ def test_package_builder_rejects_tampered_dimensions(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(), _resource())
with sqlite3.connect(database_path) as connection:
connection.execute(
"UPDATE counter_aggregates SET dimensions_json = ?",
@ -840,11 +1038,31 @@ def test_package_builder_rejects_tampered_dimensions(tmp_path):
assert list(outbox_directory.glob("*.json")) == []
def test_package_builder_rejects_tampered_client_resources(tmp_path):
database_path = tmp_path / "metrics.sqlite3"
outbox_directory = tmp_path / "outbox"
store = SharedMetricsStore(database_path, outbox_directory)
store.record_model_call(_dimensions(), _resource())
with sqlite3.connect(database_path) as connection:
connection.execute(
"UPDATE counter_aggregates SET os_family = ?",
("privacy-os-canary",),
)
with pytest.raises(
ValueError,
match="Unsupported shared-metrics client resource",
):
store.create_and_export_package()
assert list(outbox_directory.glob("*.json")) == []
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")
store.record_model_call(_dimensions(), _resource())
[package_path] = store.create_and_export_package()
original_payload = package_path.read_bytes()
@ -862,11 +1080,11 @@ def test_retention_prunes_only_expired_exported_history(tmp_path):
outbox_directory = tmp_path / "outbox"
store = SharedMetricsStore(database_path, outbox_directory)
store.record_model_call(_dimensions(), "expired-version")
store.record_model_call(_dimensions(), _resource("expired-version"))
[expired_path] = store.create_and_export_package()
store.record_model_call(_dimensions(), "current-version")
store.record_model_call(_dimensions(), _resource("current-version"))
[current_path] = store.create_and_export_package()
store.record_model_call(_dimensions(), "pending-version")
store.record_model_call(_dimensions(), _resource("pending-version"))
pending_package = store._create_package()
assert pending_package is not None
@ -921,7 +1139,7 @@ def test_retention_failure_does_not_fail_a_committed_export(tmp_path, monkeypatc
tmp_path / "metrics.sqlite3",
tmp_path / "outbox",
)
store.record_model_call(_dimensions(), "test-version")
store.record_model_call(_dimensions(), _resource())
def fail_pruning():
raise OSError("retention unavailable")
@ -940,7 +1158,7 @@ def test_file_export_failure_retries_committed_outbox_without_duplicate_delta(
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(), _resource())
def fail_write(*_args, **_kwargs):
raise OSError("simulated atomic export failure")
@ -971,7 +1189,7 @@ 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")
store.record_model_call(_dimensions(), _resource())
original_create = store._create_package
create_calls = 0
@ -980,7 +1198,7 @@ def test_package_export_does_not_chase_concurrent_updates(tmp_path, monkeypatch)
create_calls += 1
package = original_create()
if create_calls == 1:
store.record_model_call(_dimensions(), "test-version")
store.record_model_call(_dimensions(), _resource())
return package
monkeypatch.setattr(store, "_create_package", create_and_record_another)
@ -1003,7 +1221,7 @@ def test_concurrent_package_builders_commit_one_delta(tmp_path):
database_path = tmp_path / "metrics.sqlite3"
outbox_directory = tmp_path / "outbox"
store = SharedMetricsStore(database_path, outbox_directory)
store.record_model_call(_dimensions(), "test-version")
store.record_model_call(_dimensions(), _resource())
ready = threading.Barrier(2)
def export() -> list[Path]:
@ -1032,7 +1250,7 @@ def test_concurrent_due_exports_create_one_daily_package(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(), _resource())
ready = threading.Barrier(8)
def export() -> None:
@ -1062,7 +1280,7 @@ def test_concurrent_model_call_updates_are_transactional(tmp_path):
def record_calls(count: int) -> None:
store = SharedMetricsStore(database_path, outbox_directory)
for _ in range(count):
store.record_model_call(_dimensions(), "test-version")
store.record_model_call(_dimensions(), _resource())
with ThreadPoolExecutor(max_workers=2) as executor:
futures = [executor.submit(record_calls, 10) for _ in range(2)]
@ -1126,7 +1344,7 @@ 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")
store.record_model_call(_dimensions(), _resource())
[package_path] = store.create_and_export_package()
assert stat.S_IMODE(database_path.parent.stat().st_mode) == 0o700