fix(observability): export shared metrics after task completion

hermes_cli/observability/relay_shared_metrics.py

@@ -351,7 +351,17 @@ def finish_task(self, event: dict[str, Any]) -> None:
         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 @@ def close_session(self, event: dict[str, Any]) -> None:
             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 @@ def shutdown(self) -> None:
             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 @@ def _finish_task(
         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 @@ def _finish_task(
                     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 {

hermes_cli/observability/shared_metrics.py

@@ -114,6 +114,14 @@ def create_and_export_package(self) -> list[Path]:
         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 @@ def _pending_period_count(self) -> int:
             ).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:

tests/hermes_cli/test_relay_shared_metrics.py

@@ -18,6 +18,7 @@
 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 export() -> list[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"

tests/hermes_cli/test_relay_shared_metrics_runtime.py

@@ -5,7 +5,9 @@
 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 base(index: int) -> dict[str, Any]:
 
     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 base(index: int) -> dict[str, Any]:
     }
     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):