From 46bae7504292a51e3c52635b59f727b51ada1043 Mon Sep 17 00:00:00 2001 From: Alex Fournier Date: Wed, 29 Jul 2026 10:10:10 -0700 Subject: [PATCH 1/2] feat(observability): add Relay active install metrics Signed-off-by: Alex Fournier --- docs/observability/relay-shared-metrics.md | 27 ++- .../observability/relay_shared_metrics.py | 24 ++- .../hermes.shared_metrics.v1.schema.json | 29 +++ hermes_cli/observability/shared_metrics.py | 156 ++++++++++---- .../observability/shared_metrics_contract.py | 19 ++ .../shared_metrics_subscriber.py | 25 ++- scripts/smoke_nemo_relay_shared_metrics.py | 24 +++ tests/hermes_cli/test_relay_shared_metrics.py | 192 +++++++++++++++++- .../test_relay_shared_metrics_runtime.py | 33 ++- 9 files changed, 479 insertions(+), 50 deletions(-) diff --git a/docs/observability/relay-shared-metrics.md b/docs/observability/relay-shared-metrics.md index 7d68f36b299ca..33c1de6498ab2 100644 --- a/docs/observability/relay-shared-metrics.md +++ b/docs/observability/relay-shared-metrics.md @@ -55,8 +55,9 @@ dependency does not change the collection or privacy policy. ## Current Slices -The current vertical slices record logical model calls, top-level task runs, -tool and approval outcomes, and skill lifecycle and reuse: +The current vertical slices record anonymous install activity, logical model +calls, top-level task runs, tool and approval outcomes, and skill lifecycle and +reuse: ```text Hermes turn, API, tool, and approval hooks @@ -76,6 +77,15 @@ checked-in model catalog. Pricing and model-family classification belong to the metrics backend. Prompts, responses, endpoints, errors, session IDs, task IDs, and request IDs are not included in the metrics event or package. +The first consented session start emits an empty `hermes.client.active` Relay +mark. The profile-scoped subscriber creates a random UUID install identity and +uses a transactional compare-and-set to record at most one active-install +counter in any rolling 24-hour window. The metric has no dimensions; Hermes +version, OS family, architecture, and install method remain bounded package +resources. Concurrent Hermes processes share the SQLite latch, so simultaneous +starts cannot double-count one install. A later session or task can attempt the +mark again, but the subscriber suppresses it until the rolling window expires. + Each task run is a Relay `Function` scope named `hermes.task_run`, parented to the owning Hermes session. The start counter contains only bounded execution surface and entrypoint values. The terminal counter contains bounded outcome, @@ -144,6 +154,13 @@ the persistent local identifier by default. It requires a separate product and privacy decision covering consent, identity scope, rotation or keyed pseudonymization, reset behavior, retention, and deletion. +The install identity is scoped to one `HERMES_HOME`. To reset it, stop Hermes +processes and remove `$HERMES_HOME/telemetry/shared_metrics`. This deliberately +removes the old identity, aggregate database, and queued local packages +together; the next consented session creates a new identity. Disabling shared +metrics stops new collection but does not silently delete previously collected +local state. + ## Smoke Test Run a real Hermes CLI turn against the deterministic local model server: @@ -160,6 +177,6 @@ The smoke has the local model request a real `read_file` tool call before its final response, then drives create, load, reuse, patch, edit, stale, archive, restore, and install skill transitions through the installed Relay binding. It verifies model, provider, task, tool, and skill counters in SQLite, validates -all exported delta packages against the closed schema, and checks that prompt, -response, tool-call ID, tool-result, and skill-name canaries are absent from the -packages. +all exported delta packages against the closed schema, verifies the anonymous +active-install counter, and checks that prompt, response, tool-call ID, +tool-result, and skill-name canaries are absent from the packages. diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index e6bf8b21c110f..6d144a76e071c 100644 --- a/hermes_cli/observability/relay_shared_metrics.py +++ b/hermes_cli/observability/relay_shared_metrics.py @@ -16,6 +16,7 @@ from hermes_cli import __version__ from .shared_metrics import SharedMetricsStore from .shared_metrics_contract import ( + CLIENT_ACTIVE_MARK, MODEL_CALL_PROFILE_MODEL, MODEL_CALL_SCOPE, SCHEMA_KEY, @@ -166,6 +167,26 @@ class _Runtime: return None return session + def record_client_active(self, event: dict[str, Any]) -> None: + """Emit one payload-free activation attempt under the session scope.""" + session = self.ensure_session(event) + if session is None: + return + self._emit_client_active(session) + + def _emit_client_active(self, session: _MetricsSession) -> None: + with session.lock: + if session.closing: + return + self._run_in_session( + session, + self.relay.scope.event, + CLIENT_ACTIVE_MARK, + handle=session.relay_session.handle, + data={}, + metadata=self._event_metadata(), + ) + def _run_in_session( self, session: _MetricsSession, @@ -210,6 +231,7 @@ class _Runtime: or session.relay_session.context is None ): return None + self._emit_client_active(session) task_context = session.relay_session.context.copy() start_fields = task_start_fields(event) active_turn = relay_runtime.active_turn(session.session_id) @@ -1085,7 +1107,7 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None: return try: if hook_name == "on_session_start": - runtime.ensure_session(kwargs) + runtime.record_client_active(kwargs) elif hook_name == "pre_llm_call": runtime.start_task(kwargs) elif hook_name == "pre_api_request": 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 9af8101a2bc4a..d70b332c853cf 100644 --- a/hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json +++ b/hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json @@ -68,6 +68,9 @@ "minItems": 1, "items": { "oneOf": [ + { + "$ref": "#/$defs/client_active_counter" + }, { "$ref": "#/$defs/model_call_counter" }, @@ -123,6 +126,32 @@ "unknown" ] }, + "client_active_counter": { + "type": "object", + "additionalProperties": false, + "required": [ + "name", + "type", + "dimensions", + "value" + ], + "properties": { + "name": { + "const": "hermes.client.active" + }, + "type": { + "const": "counter" + }, + "dimensions": { + "type": "object", + "additionalProperties": false, + "maxProperties": 0 + }, + "value": { + "const": 1 + } + } + }, "model_call_counter": { "type": "object", "additionalProperties": false, diff --git a/hermes_cli/observability/shared_metrics.py b/hermes_cli/observability/shared_metrics.py index bf78ac2cef01d..85212f7cfc530 100644 --- a/hermes_cli/observability/shared_metrics.py +++ b/hermes_cli/observability/shared_metrics.py @@ -17,6 +17,7 @@ from hermes_constants import get_hermes_home from utils import atomic_json_write from .shared_metrics_contract import ( + CLIENT_ACTIVE_METRIC, COUNTER_METRICS, MODEL_CALL_METRIC, client_resource_is_valid, @@ -29,6 +30,8 @@ _STORE_SCHEMA_VERSION = "2" _BUSY_TIMEOUT_MS = 250 _SCHEMA_BUSY_TIMEOUT_MS = 5_000 _LOCAL_HISTORY_RETENTION_DAYS = 30 +_ACTIVE_INSTALL_STATE_KEY = "client_active_recorded_at" +_ACTIVE_INSTALL_INTERVAL = timedelta(hours=24) logger = logging.getLogger(__name__) @@ -65,6 +68,55 @@ class SharedMetricsStore: """Increment the terminal model-call counter for the current UTC day.""" self.record_counter(MODEL_CALL_METRIC, dimensions, resource) + def record_client_active(self, resource: dict[str, str]) -> bool: + """Record this install at most once in any rolling 24-hour window.""" + dimensions: dict[str, str] = {} + self._validate_counter(CLIENT_ACTIVE_METRIC, dimensions, resource) + now = _utc_now() + with self._connection() as connection: + with write_txn(connection): + row = connection.execute( + "SELECT value FROM telemetry_state WHERE key = ?", + (_ACTIVE_INSTALL_STATE_KEY,), + ).fetchone() + if row is not None: + last_recorded = self._parse_state_timestamp(row["value"]) + if last_recorded is not None and last_recorded > now: + # A wall-clock correction must not suppress activity until + # the stale future timestamp plus another full interval. + connection.execute( + """ + UPDATE telemetry_state + SET value = ? + WHERE key = ? + """, + (_isoformat(now), _ACTIVE_INSTALL_STATE_KEY), + ) + return False + if ( + last_recorded is not None + and now < last_recorded + _ACTIVE_INSTALL_INTERVAL + ): + return False + + self._install_id(connection) + self._record_counter_in_transaction( + connection, + CLIENT_ACTIVE_METRIC, + dimensions, + resource, + period_start=now.date().isoformat(), + ) + connection.execute( + """ + INSERT INTO telemetry_state(key, value) + VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value + """, + (_ACTIVE_INSTALL_STATE_KEY, _isoformat(now)), + ) + return True + def record_counter( self, metric_name: str, @@ -72,53 +124,77 @@ class SharedMetricsStore: resource: dict[str, str], ) -> None: """Increment one allowlisted counter for the current UTC day.""" + self._validate_counter(metric_name, dimensions, resource) + with self._connection() as connection: + self._record_counter_in_transaction( + connection, + metric_name, + dimensions, + resource, + period_start=_utc_now().date().isoformat(), + ) + + @staticmethod + def _validate_counter( + metric_name: str, + dimensions: dict[str, str], + resource: dict[str, str], + ) -> None: 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") + + @staticmethod + def _record_counter_in_transaction( + connection: sqlite3.Connection, + metric_name: str, + dimensions: dict[str, str], + resource: dict[str, str], + *, + period_start: str, + ) -> None: 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, - os_family, - architecture, - install_method, - dimensions_json, - value, - packaged_value - ) VALUES (?, ?, ?, ?, ?, ?, ?, 1, 0) - ON CONFLICT( - period_start, - metric_name, - hermes_version, - os_family, - architecture, - install_method, - dimensions_json - ) - DO UPDATE SET value = value + 1 - """, - ( - period_start, - metric_name, - resource["hermes_version"], - resource["os_family"], - resource["architecture"], - resource["install_method"], - dimensions_json, - ), + connection.execute( + """ + INSERT INTO counter_aggregates( + period_start, + metric_name, + hermes_version, + os_family, + architecture, + install_method, + dimensions_json, + value, + packaged_value + ) VALUES (?, ?, ?, ?, ?, ?, ?, 1, 0) + ON CONFLICT( + period_start, + metric_name, + hermes_version, + os_family, + architecture, + install_method, + dimensions_json ) + DO UPDATE SET value = value + 1 + """, + ( + period_start, + metric_name, + resource["hermes_version"], + resource["os_family"], + resource["architecture"], + resource["install_method"], + dimensions_json, + ), + ) def create_and_export_package(self) -> list[Path]: """Commit one pending delta package, then atomically export the outbox.""" @@ -349,6 +425,16 @@ class SharedMetricsStore: raise RuntimeError("Unable to create the shared-metrics install identity") return str(row["value"]) + @staticmethod + def _parse_state_timestamp(value: Any) -> datetime | None: + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except (TypeError, ValueError): + return None + if parsed.tzinfo is None: + return None + return parsed.astimezone(timezone.utc) + def _pending_period_count(self) -> int: with self._connection() as connection: row = connection.execute( diff --git a/hermes_cli/observability/shared_metrics_contract.py b/hermes_cli/observability/shared_metrics_contract.py index c70032c2aa990..9969f01f6fc76 100644 --- a/hermes_cli/observability/shared_metrics_contract.py +++ b/hermes_cli/observability/shared_metrics_contract.py @@ -13,11 +13,13 @@ MODEL_CALL_SCOPE = "hermes.model_call" MODEL_CALL_PROFILE_MODEL = "unknown" TASK_SCOPE = "hermes.task_run" TOOL_CALL_SCOPE = "hermes.tool_call" +CLIENT_ACTIVE_MARK = "hermes.client.active" TOOL_APPROVAL_MARK = "hermes.tool_approval" SKILL_LIFECYCLE_MARK = "hermes.skill.lifecycle" SKILL_LOAD_MARK = "hermes.skill.load" SUBSCRIBER_NAME = "hermes.nemo_relay.shared_metrics" MODEL_CALL_METRIC = "hermes.model_call.count" +CLIENT_ACTIVE_METRIC = "hermes.client.active" TASK_STARTED_METRIC = "hermes.task_run.started" TASK_FINISHED_METRIC = "hermes.task_run.finished" TOOL_CALL_METRIC = "hermes.tool_call.count" @@ -265,6 +267,7 @@ def client_resource_is_valid(resource: Any) -> bool: _COUNTER_DIMENSION_VALUES: dict[str, dict[str, frozenset[str]]] = { + CLIENT_ACTIVE_METRIC: {}, TASK_STARTED_METRIC: { "entrypoint": TASK_ENTRYPOINTS, "execution_surface": EXECUTION_SURFACES, @@ -345,6 +348,22 @@ def _event_metadata_is_valid(event: Any) -> bool: ) in {"OK", "ERROR"} +def client_active_counter(event: Any) -> tuple[str, dict[str, str]] | None: + """Return the active-install counter for one empty allowlisted mark.""" + if not _event_metadata_is_valid(event): + return None + if ( + str(getattr(event, "kind", "") or "") != "mark" + or str(getattr(event, "name", "") or "") != CLIENT_ACTIVE_MARK + or getattr(event, "category", None) is not None + or getattr(event, "scope_category", None) is not None + or getattr(event, "category_profile", None) is not None + or getattr(event, "data", None) != {} + ): + return None + return CLIENT_ACTIVE_METRIC, {} + + def model_call_dimensions(event: Any) -> dict[str, str] | None: """Return package dimensions for one valid primary model-call end event.""" if not _event_metadata_is_valid(event): diff --git a/hermes_cli/observability/shared_metrics_subscriber.py b/hermes_cli/observability/shared_metrics_subscriber.py index ee23b11aad0a4..36ca4ce80ff47 100644 --- a/hermes_cli/observability/shared_metrics_subscriber.py +++ b/hermes_cli/observability/shared_metrics_subscriber.py @@ -12,8 +12,10 @@ from hermes_cli.config import detect_install_method from .shared_metrics import SharedMetricsStore from .shared_metrics_contract import ( + CLIENT_ACTIVE_METRIC, MODEL_CALL_METRIC, TOOL_CALL_METRIC, + client_active_counter, client_resource, model_call_dimensions, skill_counter, @@ -59,8 +61,14 @@ class SharedMetricsSubscriber: or metadata.get(RUNTIME_INSTANCE_KEY) != self._runtime_id ): return - dimensions = model_call_dimensions(event) - metric_name = MODEL_CALL_METRIC + metric = client_active_counter(event) + dimensions = None + metric_name = CLIENT_ACTIVE_METRIC + if metric is not None: + metric_name, dimensions = metric + if dimensions is None: + dimensions = model_call_dimensions(event) + metric_name = MODEL_CALL_METRIC if dimensions is None: dimensions = tool_call_dimensions(event) metric_name = TOOL_CALL_METRIC @@ -77,11 +85,14 @@ class SharedMetricsSubscriber: if not self._active: return try: - self.store.record_counter( - metric_name, - dimensions, - self._client_resource, - ) + if metric_name == CLIENT_ACTIVE_METRIC: + self.store.record_client_active(self._client_resource) + else: + self.store.record_counter( + metric_name, + dimensions, + self._client_resource, + ) except Exception: logger.warning( "Unable to persist the Hermes shared metric: %s", diff --git a/scripts/smoke_nemo_relay_shared_metrics.py b/scripts/smoke_nemo_relay_shared_metrics.py index 30502cc6b8e4b..49d5bb77512ec 100644 --- a/scripts/smoke_nemo_relay_shared_metrics.py +++ b/scripts/smoke_nemo_relay_shared_metrics.py @@ -298,6 +298,7 @@ def _validate_store(database_path: Path) -> list[dict[str, Any]]: for counter in counters: by_name.setdefault(counter["name"], []).append(counter) if set(by_name) != { + "hermes.client.active", "hermes.model_call.count", "hermes.skill.lifecycle.count", "hermes.skill.load.count", @@ -308,6 +309,17 @@ def _validate_store(database_path: Path) -> list[dict[str, Any]]: raise AssertionError( f"Unexpected SQLite counters:\n{json.dumps(counters, indent=2)}" ) + if by_name["hermes.client.active"] != [ + { + "name": "hermes.client.active", + "dimensions": {}, + "value": 1, + "packaged_value": 1, + } + ]: + raise AssertionError( + f"Unexpected client-active counter: {by_name['hermes.client.active']}" + ) [model] = by_name["hermes.model_call.count"] expected_model = { "name": "hermes.model_call.count", @@ -455,6 +467,7 @@ def _validate_packages( for metric in package.get("metrics", []): metrics.setdefault(metric["name"], []).append(metric) if set(metrics) != { + "hermes.client.active", "hermes.model_call.count", "hermes.skill.lifecycle.count", "hermes.skill.load.count", @@ -465,6 +478,17 @@ def _validate_packages( raise AssertionError( f"Unexpected package metrics:\n{json.dumps(metrics, indent=2)}" ) + if metrics["hermes.client.active"] != [ + { + "name": "hermes.client.active", + "type": "counter", + "dimensions": {}, + "value": 1, + } + ]: + raise AssertionError( + f"Unexpected client-active metric: {metrics['hermes.client.active']}" + ) models = metrics["hermes.model_call.count"] if ( sum(metric["value"] for metric in models) != 2 diff --git a/tests/hermes_cli/test_relay_shared_metrics.py b/tests/hermes_cli/test_relay_shared_metrics.py index f47889e99029d..a05999d8e539b 100644 --- a/tests/hermes_cli/test_relay_shared_metrics.py +++ b/tests/hermes_cli/test_relay_shared_metrics.py @@ -5,6 +5,7 @@ from __future__ import annotations import json import multiprocessing as mp import os +import shutil import sqlite3 import stat import threading @@ -12,7 +13,7 @@ import time import uuid from concurrent.futures import ThreadPoolExecutor from copy import deepcopy -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path from types import SimpleNamespace from typing import Any @@ -21,6 +22,7 @@ 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_ACTIVE_METRIC, CLIENT_ARCHITECTURES, CLIENT_INSTALL_METHODS, CLIENT_OS_FAMILIES, @@ -44,6 +46,7 @@ from hermes_cli.observability.shared_metrics_contract import ( TOOL_LATENCY_BUCKETS, TOOL_OUTCOMES, TOOL_RETRY_BUCKETS, + client_active_counter, client_architecture, client_install_method, client_os_family, @@ -140,6 +143,16 @@ def _record_model_calls_in_process( store.record_model_call(_dimensions(), _resource()) +def _record_client_active_in_process( + database_path: str, + outbox_directory: str, + start_barrier: Any, +) -> None: + store = SharedMetricsStore(Path(database_path), Path(outbox_directory)) + start_barrier.wait() + store.record_client_active(_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" @@ -220,6 +233,132 @@ def test_due_export_runs_once_per_utc_day_and_catches_up_pending_deltas( assert len(list((tmp_path / "outbox").glob("*.json"))) == 3 +def test_client_active_uses_a_transactional_rolling_24_hour_latch( + tmp_path, + monkeypatch, +): + database_path = tmp_path / "metrics.sqlite3" + outbox_directory = tmp_path / "outbox" + store = SharedMetricsStore(database_path, outbox_directory) + now = datetime(2026, 7, 22, 10, 0, tzinfo=timezone.utc) + monkeypatch.setattr(shared_metrics_module, "_utc_now", lambda: now) + + assert store.record_client_active(_resource()) + assert not store.record_client_active(_resource()) + + now += timedelta(hours=23, minutes=59, seconds=59) + assert not store.record_client_active(_resource()) + + now += timedelta(seconds=1) + assert store.record_client_active(_resource()) + + active = [ + counter + for counter in store.counter_snapshot() + if counter["metric_name"] == CLIENT_ACTIVE_METRIC + ] + assert [counter["dimensions"] for counter in active] == [{}, {}] + assert [counter["period_start"] for counter in active] == [ + "2026-07-22", + "2026-07-23", + ] + assert [counter["value"] for counter in active] == [1, 1] + + +def test_client_active_recovers_from_an_invalid_latch_and_creates_identity( + tmp_path, + monkeypatch, +): + database_path = tmp_path / "metrics.sqlite3" + store = SharedMetricsStore(database_path, tmp_path / "outbox") + with sqlite3.connect(database_path) as connection: + connection.execute( + "INSERT INTO telemetry_state(key, value) VALUES (?, ?)", + ("client_active_recorded_at", "invalid-timestamp"), + ) + now = datetime(2026, 7, 22, 10, 0, tzinfo=timezone.utc) + monkeypatch.setattr(shared_metrics_module, "_utc_now", lambda: now) + + assert store.record_client_active(_resource()) + + with sqlite3.connect(database_path) as connection: + state = dict( + connection.execute( + "SELECT key, value FROM telemetry_state WHERE key != 'schema_version'" + ).fetchall() + ) + uuid.UUID(state["install_id"]) + assert state["client_active_recorded_at"] == "2026-07-22T10:00:00Z" + + +def test_client_active_rebases_a_future_latch_without_double_counting( + tmp_path, + monkeypatch, +): + database_path = tmp_path / "metrics.sqlite3" + store = SharedMetricsStore(database_path, tmp_path / "outbox") + now = datetime(2026, 7, 22, 10, 0, tzinfo=timezone.utc) + monkeypatch.setattr(shared_metrics_module, "_utc_now", lambda: now) + + assert store.record_client_active(_resource()) + with sqlite3.connect(database_path) as connection: + connection.execute( + "UPDATE telemetry_state SET value = ? WHERE key = ?", + ("2026-07-24T10:00:00Z", "client_active_recorded_at"), + ) + + assert not store.record_client_active(_resource()) + with sqlite3.connect(database_path) as connection: + latch = connection.execute( + "SELECT value FROM telemetry_state WHERE key = ?", + ("client_active_recorded_at",), + ).fetchone()[0] + + assert latch == "2026-07-22T10:00:00Z" + [counter] = store.counter_snapshot() + assert counter["metric_name"] == CLIENT_ACTIVE_METRIC + assert counter["value"] == 1 + + +def test_client_active_package_uses_empty_dimensions_and_stable_install_id(tmp_path): + store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox") + + assert store.record_client_active(_resource()) + [package_path] = store.create_and_export_package() + package = json.loads(package_path.read_text(encoding="utf-8")) + + _schema_validator().validate(package) + uuid.UUID(package["install_id"]) + assert package["metrics"] == [ + { + "name": CLIENT_ACTIVE_METRIC, + "type": "counter", + "dimensions": {}, + "value": 1, + } + ] + + +def test_deleting_local_metrics_state_resets_install_identity(tmp_path): + root = tmp_path / "shared-metrics" + database_path = root / "metrics.sqlite3" + outbox_directory = root / "outbox" + first = SharedMetricsStore(database_path, outbox_directory) + assert first.record_client_active(_resource()) + [first_package_path] = first.create_and_export_package() + first_package = json.loads(first_package_path.read_text(encoding="utf-8")) + + shutil.rmtree(root) + + reset = SharedMetricsStore(database_path, outbox_directory) + assert reset.record_client_active(_resource()) + [reset_package_path] = reset.create_and_export_package() + reset_package = json.loads(reset_package_path.read_text(encoding="utf-8")) + + assert reset_package["install_id"] != first_package["install_id"] + assert reset_package["metrics"][0]["name"] == CLIENT_ACTIVE_METRIC + + def test_package_schema_matches_the_model_call_contract(): properties = _package_dimension_schema()["properties"] @@ -272,6 +411,30 @@ def test_package_schema_matches_the_client_resource_contract(): ) +def test_client_active_mark_accepts_only_an_empty_allowlisted_payload(): + event = SimpleNamespace( + kind="mark", + category=None, + category_profile=None, + name="hermes.client.active", + scope_category=None, + metadata={ + "hermes.metrics.schema_version": "hermes.metrics.event.v1", + }, + data={}, + ) + + assert client_active_counter(event) == (CLIENT_ACTIVE_METRIC, {}) + + with_payload = deepcopy(event) + with_payload.data = {"session_id": "privacy-canary"} + assert client_active_counter(with_payload) is None + + wrong_schema = deepcopy(event) + wrong_schema.metadata["hermes.metrics.schema_version"] = "unknown" + assert client_active_counter(wrong_schema) is None + + 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"] @@ -1315,6 +1478,33 @@ def test_cross_process_model_call_updates_are_transactional(tmp_path): assert restarted.counter_snapshot()[0]["value"] == 20 +def test_cross_process_client_active_attempts_record_one_install(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_client_active_in_process, + args=(str(database_path), str(outbox_directory), 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 + + store = SharedMetricsStore(database_path, outbox_directory) + [active] = store.counter_snapshot() + assert active["metric_name"] == CLIENT_ACTIVE_METRIC + assert active["dimensions"] == {} + assert active["value"] == 1 + + def test_schema_initialization_waits_for_an_existing_writer(tmp_path): database_path = tmp_path / "metrics.sqlite3" outbox_directory = tmp_path / "outbox" diff --git a/tests/hermes_cli/test_relay_shared_metrics_runtime.py b/tests/hermes_cli/test_relay_shared_metrics_runtime.py index 7022e5ac1af6a..f2567b5e326dc 100644 --- a/tests/hermes_cli/test_relay_shared_metrics_runtime.py +++ b/tests/hermes_cli/test_relay_shared_metrics_runtime.py @@ -344,6 +344,13 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa } assert starts[0][2] == {} assert starts[0][3]["model_name"] == "unknown" + active_marks = [ + event + for event in direct_runtime.events + if event[0] == "scope.event" and event[1] == "hermes.client.active" + ] + assert len(active_marks) == 2 + assert all(mark[2]["data"] == {} for mark in active_marks) assert ends[0][2] == { "model": "claude-sonnet", "provider": "anthropic", @@ -365,12 +372,19 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa package = json.loads(packages[0].read_text(encoding="utf-8")) metrics = {metric["name"]: metric for metric in package["metrics"]} assert set(metrics) == { + "hermes.client.active", "hermes.model_call.count", "hermes.task_run.finished", "hermes.task_run.started", "hermes.tool_approval.count", "hermes.tool_call.count", } + assert metrics["hermes.client.active"] == { + "name": "hermes.client.active", + "type": "counter", + "dimensions": {}, + "value": 1, + } assert metrics["hermes.model_call.count"]["dimensions"] == { "model": "claude-sonnet", "provider": "anthropic", @@ -590,6 +604,8 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot( for counter in snapshot: by_metric.setdefault(counter["metric_name"], []).append(counter) + assert by_metric["hermes.client.active"][0]["dimensions"] == {} + assert by_metric["hermes.client.active"][0]["value"] == 1 assert len(by_metric["hermes.task_run.started"]) == 1 assert by_metric["hermes.task_run.started"][0]["value"] == 3 assert len(by_metric["hermes.model_call.count"]) == 1 @@ -1029,7 +1045,11 @@ def test_core_mark_uses_the_shared_session_handle_without_a_plugin(direct_runtim metadata={"data_schema": "hermes.skill.lifecycle.v1"}, ) - [mark] = [event for event in direct_runtime.events if event[0] == "scope.event"] + [mark] = [ + event + for event in direct_runtime.events + if event[0] == "scope.event" and event[1] == "hermes.skill.created" + ] assert mark[1] == "hermes.skill.created" assert mark[2]["handle"] == handle assert plugins.get_plugin_manager().list_plugins() == [] @@ -1509,16 +1529,26 @@ def test_shared_metrics_subscribers_isolate_two_enabled_profiles(tmp_path, monke finally: reset_hermes_home_override(token) + install_ids: set[str] = set() for profile in (profile_a, profile_b): packages = list( (profile / "telemetry" / "shared_metrics" / "outbox").glob("*.json") ) assert len(packages) == 1 package = json.loads(packages[0].read_text(encoding="utf-8")) + install_ids.add(package["install_id"]) metrics = {metric["name"]: metric for metric in package["metrics"]} + assert metrics["hermes.client.active"] == { + "name": "hermes.client.active", + "type": "counter", + "dimensions": {}, + "value": 1, + } assert metrics["hermes.task_run.started"]["value"] == 1 assert metrics["hermes.task_run.finished"]["value"] == 1 + assert len(install_ids) == 2 + relay_shared_metrics._reset_for_tests() relay_runtime._reset_for_tests() @@ -1671,6 +1701,7 @@ def test_disabling_shared_metrics_stops_collection_and_shutdown_export( root = profile / "telemetry" / "shared_metrics" store = SharedMetricsStore(root / "metrics.sqlite3", root / "outbox") assert [row["metric_name"] for row in store.counter_snapshot()] == [ + "hermes.client.active", "hermes.task_run.started" ] assert list((root / "outbox").glob("*.json")) == [] From 535a59c5c0d8703924932a7b457b46e5006c6bfb Mon Sep 17 00:00:00 2001 From: Alex Fournier Date: Tue, 4 Aug 2026 15:13:34 -0700 Subject: [PATCH 2/2] docs(observability): clarify active profile identity Signed-off-by: Alex Fournier --- docs/observability/relay-shared-metrics.md | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/docs/observability/relay-shared-metrics.md b/docs/observability/relay-shared-metrics.md index ad5711c06e2db..860d916e2172a 100644 --- a/docs/observability/relay-shared-metrics.md +++ b/docs/observability/relay-shared-metrics.md @@ -55,9 +55,9 @@ dependency does not change the collection or privacy policy. ## Current Slices -The current vertical slices record anonymous install activity, logical model -calls, top-level task runs, tool and approval outcomes, and skill lifecycle and -reuse: +The current vertical slices record pseudonymous profile activity, logical +model calls, top-level task runs, tool and approval outcomes, and skill +lifecycle and reuse: ```text Hermes turn, API, tool, and approval hooks @@ -82,7 +82,7 @@ counters created by older builds can be exported without losing data. The first consented session start emits an empty `hermes.client.active` Relay mark. The profile-scoped subscriber creates a random UUID install identity and -uses a transactional compare-and-set to record at most one active-install +uses a transactional compare-and-set to record at most one client-active counter in any rolling 24-hour window. The metric has no dimensions; Hermes version, OS family, architecture, and install method remain bounded package resources. Concurrent Hermes processes share the SQLite latch, so simultaneous @@ -183,6 +183,6 @@ The smoke has the local model request a real `read_file` tool call before its final response, then drives create, load, reuse, patch, edit, stale, archive, restore, and install skill transitions through the installed Relay binding. It verifies model, provider, task, tool, and skill counters in SQLite, validates -all exported delta packages against the closed schema, verifies the anonymous -active-install counter, and checks that prompt, response, tool-call ID, -tool-result, and skill-name canaries are absent from the packages. +all exported delta packages against the closed schema, verifies the +pseudonymous client-active counter, and checks that prompt, response, tool-call +ID, tool-result, and skill-name canaries are absent from the packages.