diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index fb6f885b2f0..c35be04d697 100644 --- a/hermes_cli/observability/relay_shared_metrics.py +++ b/hermes_cli/observability/relay_shared_metrics.py @@ -351,7 +351,17 @@ 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: + try: + self.relay.subscribers.flush() + except Exception: + logger.warning( + "Hermes shared-metrics task flush failed", + exc_info=True, + ) + else: + self._export() def close_session(self, event: dict[str, Any]) -> None: session = self._session(event) @@ -380,7 +390,8 @@ class _Runtime: self.relay.subscribers.flush() except Exception as exc: failures.append(f"subscriber flush failed: {exc}") - self._export() + else: + self._export() with self._sessions_lock: if self._sessions.get(session.session_id) is session: self._sessions.pop(session.session_id, None) @@ -399,8 +410,15 @@ class _Runtime: self._safe(self.close_session, {"session_id": session_id}) if not self._registered: return - self._safe(self.relay.subscribers.flush) - self._export() + try: + self.relay.subscribers.flush() + except Exception: + logger.warning( + "Hermes shared-metrics shutdown flush failed", + exc_info=True, + ) + else: + self._export() self._safe(self.relay.subscribers.deregister, self._subscriber_name) self.host.release_managed_execution(self._subscriber_name) self._registered = False @@ -560,10 +578,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 +610,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 506fb4ab131..666053e0d19 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 058e07db7bc..573835be984 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 ccda2abc1f4..f1690dc3926 100644 --- a/tests/hermes_cli/test_relay_shared_metrics_runtime.py +++ b/tests/hermes_cli/test_relay_shared_metrics_runtime.py @@ -5,7 +5,9 @@ from __future__ import annotations import contextvars import asyncio import json +import sqlite3 import threading +from datetime import datetime, timedelta, timezone from pathlib import Path from types import SimpleNamespace from typing import Any @@ -321,6 +323,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 +419,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 +460,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 +2288,133 @@ 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")) + [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_failed_flush_keeps_daily_export_open_for_later_task( + direct_runtime, tmp_path, monkeypatch, caplog +): + current_time = datetime(2026, 7, 28, 9, tzinfo=timezone.utc) + monkeypatch.setattr( + "hermes_cli.observability.shared_metrics._utc_now", + lambda: current_time, + ) + original_flush = direct_runtime.subscribers.flush + flush_attempts = 0 + + def fail_first_flush() -> None: + nonlocal flush_attempts + flush_attempts += 1 + if flush_attempts == 1: + raise RuntimeError("simulated flush failure") + original_flush() + + direct_runtime.subscribers.flush = fail_first_flush + + def finish_desktop_task(task_id: str) -> None: + lifecycle.invoke_hook( + "pre_llm_call", + session_id="s1", + task_id=task_id, + platform="desktop", + ) + lifecycle.invoke_hook( + "on_session_end", + session_id="s1", + task_id=task_id, + platform="desktop", + completed=True, + failed=False, + interrupted=False, + turn_exit_reason="text_response(stop)", + ) + + finish_desktop_task("t1") + + root = tmp_path / "hermes-home" / "telemetry" / "shared_metrics" + assert list((root / "outbox").glob("*.json")) == [] + with sqlite3.connect(root / "metrics.sqlite3") as connection: + [package_count] = connection.execute( + "SELECT COUNT(*) FROM package_outbox" + ).fetchone() + assert package_count == 0 + + finish_desktop_task("t2") + + [package_path] = list((root / "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 + assert flush_attempts == 2 + assert "Hermes shared-metrics task flush failed" in caplog.text def test_task_ownership_survives_session_id_rotation(direct_runtime):