diff --git a/docs/observability/relay-shared-metrics.md b/docs/observability/relay-shared-metrics.md index 182608dc225ee..860d916e2172a 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 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 @@ -79,6 +80,15 @@ New calls use `hermes.model_route.count`. The previous `hermes.model_call.count` contract remains readable only so pending local 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 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 +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, @@ -150,6 +160,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: @@ -166,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, 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. diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index e1d8c43ca594d..756c6caa88362 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) @@ -1089,7 +1111,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.v2.schema.json b/hermes_cli/observability/schemas/hermes.shared_metrics.v2.schema.json index 1416581a2ae48..5d0bea610a7bf 100644 --- a/hermes_cli/observability/schemas/hermes.shared_metrics.v2.schema.json +++ b/hermes_cli/observability/schemas/hermes.shared_metrics.v2.schema.json @@ -71,6 +71,9 @@ "minItems": 1, "items": { "oneOf": [ + { + "$ref": "#/$defs/client_active_counter" + }, { "$ref": "#/$defs/model_call_counter" }, @@ -100,6 +103,32 @@ } }, "$defs": { + "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 + } + } + }, "uuid": { "type": "string", "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" diff --git a/hermes_cli/observability/shared_metrics.py b/hermes_cli/observability/shared_metrics.py index f488c0d8224a2..fd42b0623065e 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_ROUTE_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_ROUTE_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 2bbea8e1de017..b7637c9317c57 100644 --- a/hermes_cli/observability/shared_metrics_contract.py +++ b/hermes_cli/observability/shared_metrics_contract.py @@ -18,10 +18,12 @@ 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" +CLIENT_ACTIVE_METRIC = "hermes.client.active" LEGACY_MODEL_CALL_METRIC = "hermes.model_call.count" MODEL_ROUTE_METRIC = "hermes.model_route.count" TASK_STARTED_METRIC = "hermes.task_run.started" @@ -306,6 +308,7 @@ _LEGACY_MODEL_FAMILIES = frozenset({ }) _COUNTER_DIMENSION_VALUES: dict[str, dict[str, frozenset[str]]] = { + CLIENT_ACTIVE_METRIC: {}, # Retained only so pre-v2 pending rows remain packageable. LEGACY_MODEL_CALL_METRIC: { "call_role": frozenset({"primary"}), @@ -352,6 +355,7 @@ _COUNTER_DIMENSION_VALUES: dict[str, dict[str, frozenset[str]]] = { }, } COUNTER_METRICS: frozenset[str] = frozenset({ + CLIENT_ACTIVE_METRIC, MODEL_ROUTE_METRIC, SKILL_LIFECYCLE_METRIC, SKILL_LOAD_METRIC, @@ -400,6 +404,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 logical model-call end event.""" auxiliary = _auxiliary_model_call_dimensions(event) diff --git a/hermes_cli/observability/shared_metrics_subscriber.py b/hermes_cli/observability/shared_metrics_subscriber.py index 742edd9ca98d1..54c05e5780d1e 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_ROUTE_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_ROUTE_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_ROUTE_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 b60a6b5dc7816..ba11567532d27 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_route.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_route.count"] expected_model = { "name": "hermes.model_route.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_route.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']}" + ) [model] = metrics["hermes.model_route.count"] if model["dimensions"] != { "model": MODEL_CANARY, diff --git a/tests/hermes_cli/test_relay_shared_metrics.py b/tests/hermes_cli/test_relay_shared_metrics.py index 4918e7cffce30..e26d3fd2cafd9 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 @@ -22,6 +23,7 @@ from agent import relay_runtime 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, @@ -49,6 +51,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, @@ -155,6 +158,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" @@ -363,6 +376,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(): schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) properties = _package_dimension_schema()["properties"] @@ -419,6 +558,31 @@ def test_package_schema_matches_the_client_resource_contract(): CLIENT_INSTALL_METHODS ) + +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.v2", + }, + 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"] @@ -1298,6 +1462,55 @@ 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" + database_path.touch() + blocker = sqlite3.connect(database_path) + blocker.execute("BEGIN IMMEDIATE") + + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit( + SharedMetricsStore, + database_path, + outbox_directory, + ) + try: + time.sleep(0.4) + assert not future.done() + finally: + blocker.rollback() + blocker.close() + store = future.result(timeout=2) + + assert store.counter_snapshot() == [] @pytest.mark.skipif(os.name == "nt", reason="POSIX permission modes are unavailable") diff --git a/tests/hermes_cli/test_relay_shared_metrics_runtime.py b/tests/hermes_cli/test_relay_shared_metrics_runtime.py index dc8328a86a2a7..b38232276b35b 100644 --- a/tests/hermes_cli/test_relay_shared_metrics_runtime.py +++ b/tests/hermes_cli/test_relay_shared_metrics_runtime.py @@ -359,6 +359,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", @@ -380,12 +387,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_route.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_route.count"]["dimensions"] == { "model": "claude-sonnet", "provider": "anthropic", @@ -605,6 +619,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_route.count"]) == 1 @@ -1214,6 +1230,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")) == []