From 2aa34c54848c94afac8ba52607f2bb95b728dc87 Mon Sep 17 00:00:00 2001 From: Alex Fournier Date: Tue, 28 Jul 2026 11:06:14 -0700 Subject: [PATCH] fix(observability): export shared metrics after tasks Signed-off-by: Alex Fournier --- .../observability/relay_shared_metrics.py | 12 ++-- hermes_cli/observability/shared_metrics.py | 28 ++++++++ tests/hermes_cli/test_relay_shared_metrics.py | 64 +++++++++++++++++ .../test_relay_shared_metrics_runtime.py | 70 ++++++++++++++++--- 4 files changed, 160 insertions(+), 14 deletions(-) diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index fb6f885b2f05..2274f8cd4df7 100644 --- a/hermes_cli/observability/relay_shared_metrics.py +++ b/hermes_cli/observability/relay_shared_metrics.py @@ -351,7 +351,10 @@ class _Runtime: with session.lock: if session.closing: return - self._finish_task(session, task_id, event) + finished = self._finish_task(session, task_id, event) + if finished: + self._safe(self.relay.subscribers.flush) + self._export() def close_session(self, event: dict[str, Any]) -> None: session = self._session(event) @@ -560,10 +563,10 @@ class _Runtime: session: _MetricsSession, task_id: str, event: dict[str, Any], - ) -> None: + ) -> bool: task = session.tasks.get(task_id) if task is None: - return + return False self._end_pending_model_calls(session, {**event, "task_id": task_id}) fields = task_terminal_fields( {**task.start_fields, **event}, @@ -592,9 +595,10 @@ class _Runtime: turn_key = (session.session_id, turn_id) if self._turn_sessions.get(turn_key) is session: self._turn_sessions.pop(turn_key, None) + return True def _export(self) -> None: - self._safe(self.subscriber.store.create_and_export_package) + self._safe(self.subscriber.store.create_and_export_package_if_due) def _event_metadata(self) -> dict[str, str]: return { diff --git a/hermes_cli/observability/shared_metrics.py b/hermes_cli/observability/shared_metrics.py index 506fb4ab1314..666053e0d19d 100644 --- a/hermes_cli/observability/shared_metrics.py +++ b/hermes_cli/observability/shared_metrics.py @@ -114,6 +114,14 @@ class SharedMetricsStore: for _ in range(pending_periods): if self._create_package() is None: break + return self._export_and_prune() + + def create_and_export_package_if_due(self) -> list[Path]: + """Create pending packages at most once per UTC day, then export them.""" + self._create_pending_packages_if_due() + return self._export_and_prune() + + def _export_and_prune(self) -> list[Path]: exported = self._export_pending_packages() try: self._prune_expired_history() @@ -281,6 +289,26 @@ class SharedMetricsStore: ).fetchone() return int(row["period_count"]) if row is not None else 0 + def _create_pending_packages_if_due(self) -> None: + now = _utc_now() + with self._connection() as connection: + with write_txn(connection): + # Gate on the committed package, not its file write, so a failed + # outbox export can be retried without packaging deltas twice. + package_created_today = connection.execute( + """ + SELECT 1 + FROM package_outbox + WHERE substr(created_at, 1, 10) >= ? + LIMIT 1 + """, + (now.date().isoformat(),), + ).fetchone() + if package_created_today is not None: + return + while self._create_package_in_transaction(connection, now) is not None: + pass + def _create_package(self) -> dict[str, Any] | None: now = _utc_now() with self._connection() as connection: diff --git a/tests/hermes_cli/test_relay_shared_metrics.py b/tests/hermes_cli/test_relay_shared_metrics.py index 058e07db7bc7..573835be9845 100644 --- a/tests/hermes_cli/test_relay_shared_metrics.py +++ b/tests/hermes_cli/test_relay_shared_metrics.py @@ -18,6 +18,7 @@ from types import SimpleNamespace from typing import Any 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 ( COUNT_BUCKETS, @@ -142,6 +143,43 @@ def test_model_call_counter_survives_restart_and_exports_only_new_deltas(tmp_pat assert restarted.counter_snapshot()[0]["packaged_value"] == 3 +def test_due_export_runs_once_per_utc_day_and_catches_up_pending_deltas( + tmp_path, monkeypatch +): + current_time = datetime(2026, 7, 28, 9, tzinfo=timezone.utc) + 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") + 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") + 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", + "dimensions": _dimensions(), + "value": 2, + "packaged_value": 1, + } + + current_time = datetime(2026, 7, 29, 9, tzinfo=timezone.utc) + store.record_model_call(_dimensions(), "test-version") + 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") + assert store.create_and_export_package_if_due() == [] + assert len(list((tmp_path / "outbox").glob("*.json"))) == 3 + + def test_package_schema_matches_the_model_call_contract(): properties = _package_dimension_schema()["properties"] @@ -758,6 +796,32 @@ def test_concurrent_package_builders_commit_one_delta(tmp_path): assert store.counter_snapshot()[0]["packaged_value"] == 1 +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") + ready = threading.Barrier(8) + + def export() -> None: + worker_store = SharedMetricsStore(database_path, outbox_directory) + ready.wait(timeout=5) + worker_store.create_and_export_package_if_due() + + with ThreadPoolExecutor(max_workers=8) as executor: + futures = [executor.submit(export) for _ in range(8)] + for future in futures: + future.result() + + with sqlite3.connect(database_path) as connection: + [outbox_count] = connection.execute( + "SELECT COUNT(*) FROM package_outbox" + ).fetchone() + assert outbox_count == 1 + assert len(list(outbox_directory.glob("*.json"))) == 1 + assert store.counter_snapshot()[0]["packaged_value"] == 1 + + def test_concurrent_model_call_updates_are_transactional(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 ccda2abc1f4c..0005d8653e08 100644 --- a/tests/hermes_cli/test_relay_shared_metrics_runtime.py +++ b/tests/hermes_cli/test_relay_shared_metrics_runtime.py @@ -6,6 +6,7 @@ import contextvars import asyncio import json import threading +from datetime import datetime, timedelta, timezone from pathlib import Path from types import SimpleNamespace from typing import Any @@ -321,6 +322,7 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot( real_binding_runtime, tmp_path, + monkeypatch, ): assert real_binding_runtime._native is not None prompt_canary = "real-relay-sensitive-prompt" @@ -416,6 +418,12 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot( root = tmp_path / "hermes-home" / "telemetry" / "shared_metrics" store = SharedMetricsStore(root / "metrics.sqlite3", root / "outbox") + tomorrow = datetime.now(timezone.utc) + timedelta(days=1) + monkeypatch.setattr( + "hermes_cli.observability.shared_metrics._utc_now", + lambda: tomorrow, + ) + assert len(store.create_and_export_package_if_due()) == 1 snapshot = store.counter_snapshot() by_metric: dict[str, list[dict[str, Any]]] = {} for counter in snapshot: @@ -451,7 +459,7 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot( } package_values: dict[tuple[str, tuple[tuple[str, str], ...]], int] = {} packages = sorted((root / "outbox").glob("*.json")) - assert len(packages) == 3 + assert len(packages) == 2 package_payloads = [ json.loads(package.read_text(encoding="utf-8")) for package in packages ] @@ -2279,32 +2287,74 @@ def test_session_finalize_closes_a_pending_task_as_system_aborted(direct_runtime } -def test_sequential_tasks_in_one_session_aggregate_once_each(direct_runtime, tmp_path): +def test_desktop_task_completion_exports_once_per_utc_day( + direct_runtime, tmp_path, monkeypatch +): + current_time = datetime(2026, 7, 28, 9, tzinfo=timezone.utc) + monkeypatch.setattr( + "hermes_cli.observability.shared_metrics._utc_now", + lambda: current_time, + ) for task_id in ("t1", "t2"): lifecycle.invoke_hook( "pre_llm_call", session_id="s1", task_id=task_id, - platform="cli", + platform="desktop", ) lifecycle.invoke_hook( "on_session_end", session_id="s1", task_id=task_id, - platform="cli", + platform="desktop", completed=True, failed=False, interrupted=False, turn_exit_reason="text_response(stop)", ) - lifecycle.finalize_session(session_id="s1") outbox = tmp_path / "hermes-home" / "telemetry" / "shared_metrics" / "outbox" - [package_path] = list(outbox.glob("*.json")) - package = json.loads(package_path.read_text(encoding="utf-8")) - metrics = {metric["name"]: metric for metric in package["metrics"]} - assert metrics["hermes.task_run.started"]["value"] == 2 - assert metrics["hermes.task_run.finished"]["value"] == 2 + [first_package_path] = list(outbox.glob("*.json")) + first_package = json.loads(first_package_path.read_text(encoding="utf-8")) + first_metrics = {metric["name"]: metric for metric in first_package["metrics"]} + assert first_metrics["hermes.task_run.started"]["value"] == 1 + assert first_metrics["hermes.task_run.started"]["dimensions"] == { + "entrypoint": "interactive", + "execution_surface": "desktop", + } + assert first_metrics["hermes.task_run.finished"]["value"] == 1 + + lifecycle.finalize_session(session_id="s1") + assert list(outbox.glob("*.json")) == [first_package_path] + + current_time = datetime(2026, 7, 29, 9, tzinfo=timezone.utc) + lifecycle.invoke_hook( + "pre_llm_call", + session_id="s1", + task_id="t3", + platform="desktop", + ) + lifecycle.invoke_hook( + "on_session_end", + session_id="s1", + task_id="t3", + platform="desktop", + completed=True, + failed=False, + interrupted=False, + turn_exit_reason="text_response(stop)", + ) + + packages = [ + json.loads(package_path.read_text(encoding="utf-8")) + for package_path in outbox.glob("*.json") + ] + totals: dict[str, int] = {} + for package in packages: + for metric in package["metrics"]: + totals[metric["name"]] = totals.get(metric["name"], 0) + metric["value"] + assert totals["hermes.task_run.started"] == 3 + assert totals["hermes.task_run.finished"] == 3 def test_task_ownership_survives_session_id_rotation(direct_runtime):