diff --git a/docs/observability/relay-shared-metrics.md b/docs/observability/relay-shared-metrics.md index f6c0b960b742e..7d68f36b299ca 100644 --- a/docs/observability/relay-shared-metrics.md +++ b/docs/observability/relay-shared-metrics.md @@ -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 diff --git a/hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json b/hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json index bafaadf7ab6c7..9af8101a2bc4a 100644 --- a/hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json +++ b/hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json @@ -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"] } } }, diff --git a/hermes_cli/observability/shared_metrics.py b/hermes_cli/observability/shared_metrics.py index 666053e0d19d1..bf78ac2cef01d 100644 --- a/hermes_cli/observability/shared_metrics.py +++ b/hermes_cli/observability/shared_metrics.py @@ -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"], ), ) diff --git a/hermes_cli/observability/shared_metrics_contract.py b/hermes_cli/observability/shared_metrics_contract.py index bf6a3c9468d83..c70032c2aa990 100644 --- a/hermes_cli/observability/shared_metrics_contract.py +++ b/hermes_cli/observability/shared_metrics_contract.py @@ -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: { diff --git a/hermes_cli/observability/shared_metrics_subscriber.py b/hermes_cli/observability/shared_metrics_subscriber.py index 8ca708044649d..ee23b11aad0a4 100644 --- a/hermes_cli/observability/shared_metrics_subscriber.py +++ b/hermes_cli/observability/shared_metrics_subscriber.py @@ -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( diff --git a/scripts/smoke_nemo_relay_shared_metrics.py b/scripts/smoke_nemo_relay_shared_metrics.py index ae3b1ee72c0f6..30502cc6b8e4b 100644 --- a/scripts/smoke_nemo_relay_shared_metrics.py +++ b/scripts/smoke_nemo_relay_shared_metrics.py @@ -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 ( diff --git a/tests/hermes_cli/test_relay_shared_metrics.py b/tests/hermes_cli/test_relay_shared_metrics.py index e2c5c9f66fe35..f47889e99029d 100644 --- a/tests/hermes_cli/test_relay_shared_metrics.py +++ b/tests/hermes_cli/test_relay_shared_metrics.py @@ -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