fix(telemetry): prune expired local metrics

Signed-off-by: Alex Fournier <afournier@nvidia.com>

cli-config.yaml.example

@@ -1410,6 +1410,8 @@ display:
 # packages under $HERMES_HOME/telemetry/shared_metrics; it does not upload them.
 # Packages include a random profile-scoped ID that stays stable until this
 # directory is deleted. It is not derived from hardware, account, or host data.
+# Successfully exported local history is retained for 30 days; pending deltas
+# are retained until they can be exported.
 telemetry:
   shared_metrics:
     enabled: false

docs/observability/relay-shared-metrics.md

@@ -72,7 +72,9 @@ $HERMES_HOME/telemetry/shared_metrics/outbox/*.json
 
 The database keeps transactional aggregate and package-outbox state. Package
 files are immutable delta documents that conform to a closed JSON schema and
-are written with atomic replacement.
+are written with atomic replacement. Fully packaged aggregate rows and
+successfully exported package rows and files are retained locally for 30 days.
+Pending package rows and counters with unexported deltas are never pruned.
 
 Each package contains an `install_id` generated as a random UUID. Despite the
 schema field name, its current scope is one `HERMES_HOME`, so it is more

hermes_cli/observability/shared_metrics.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import sqlite3
 import uuid
 from collections.abc import Iterator
@@ -26,6 +27,9 @@ _PACKAGE_SCHEMA_VERSION = "hermes.shared_metrics.v1"
 _STORE_SCHEMA_VERSION = "1"
 _BUSY_TIMEOUT_MS = 250
 _SCHEMA_BUSY_TIMEOUT_MS = 5_000
+_LOCAL_HISTORY_RETENTION_DAYS = 30
+
+logger = logging.getLogger(__name__)
 
 
 def _utc_now() -> datetime:
@@ -110,7 +114,15 @@ class SharedMetricsStore:
         for _ in range(pending_periods):
             if self._create_package() is None:
                 break
-        return self._export_pending_packages()
+        exported = self._export_pending_packages()
+        try:
+            self._prune_expired_history()
+        except Exception:
+            logger.warning(
+                "Unable to prune expired shared-metrics history",
+                exc_info=True,
+            )
+        return exported
 
     def counter_snapshot(self) -> list[dict[str, Any]]:
         """Return cumulative counters for focused tests and local inspection."""
@@ -409,3 +421,66 @@ class SharedMetricsStore:
                 )
             exported.append(path)
         return exported
+
+    def _prune_expired_history(self, *, now: datetime | None = None) -> None:
+        """Remove exported local history after the bounded retention window."""
+        cutoff = (now or _utc_now()) - timedelta(
+            days=_LOCAL_HISTORY_RETENTION_DAYS
+        )
+        cutoff_timestamp = _isoformat(cutoff)
+        cutoff_period = cutoff.date().isoformat()
+        with self._connection() as connection:
+            rows = connection.execute(
+                """
+                SELECT package_id
+                FROM package_outbox
+                WHERE exported_at IS NOT NULL
+                  AND exported_at < ?
+                ORDER BY exported_at, package_id
+                """,
+                (cutoff_timestamp,),
+            ).fetchall()
+
+        removable_package_ids: list[str] = []
+        for row in rows:
+            package_id = str(row["package_id"])
+            try:
+                (self.outbox_directory / f"{package_id}.json").unlink(
+                    missing_ok=True
+                )
+            except OSError:
+                logger.warning(
+                    "Unable to prune expired shared-metrics package %s",
+                    package_id,
+                    exc_info=True,
+                )
+                continue
+            removable_package_ids.append(package_id)
+
+        with self._connection() as connection:
+            with write_txn(connection):
+                for package_id in removable_package_ids:
+                    connection.execute(
+                        """
+                        DELETE FROM package_outbox
+                        WHERE package_id = ?
+                          AND exported_at IS NOT NULL
+                          AND exported_at < ?
+                        """,
+                        (package_id, cutoff_timestamp),
+                    )
+                connection.execute(
+                    """
+                    DELETE FROM counter_aggregates
+                    WHERE period_start < ?
+                      AND value = packaged_value
+                      AND NOT EXISTS (
+                          SELECT 1
+                          FROM package_outbox
+                          WHERE exported_at IS NULL
+                            AND substr(package_outbox.period_start, 1, 10)
+                                = counter_aggregates.period_start
+                      )
+                    """,
+                    (cutoff_period,),
+                )

tests/hermes_cli/test_relay_shared_metrics.py

@@ -12,6 +12,7 @@ import time
 import uuid
 from concurrent.futures import ThreadPoolExecutor
 from copy import deepcopy
+from datetime import datetime, timezone
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any
@@ -586,6 +587,83 @@ def test_pending_package_retry_reuses_the_same_package_and_file(tmp_path):
     assert list(outbox_directory.glob("*.json")) == [package_path]
 
 
+def test_retention_prunes_only_expired_exported_history(tmp_path):
+    database_path = tmp_path / "metrics.sqlite3"
+    outbox_directory = tmp_path / "outbox"
+    store = SharedMetricsStore(database_path, outbox_directory)
+
+    store.record_model_call(_dimensions(), "expired-version")
+    [expired_path] = store.create_and_export_package()
+    store.record_model_call(_dimensions(), "current-version")
+    [current_path] = store.create_and_export_package()
+    store.record_model_call(_dimensions(), "pending-version")
+    pending_package = store._create_package()
+    assert pending_package is not None
+
+    with sqlite3.connect(database_path) as connection:
+        connection.execute(
+            """
+            UPDATE counter_aggregates
+            SET period_start = '2026-05-01'
+            WHERE hermes_version = 'expired-version'
+            """
+        )
+        connection.execute(
+            """
+            UPDATE package_outbox
+            SET period_start = '2026-05-01T00:00:00Z',
+                period_end = '2026-05-02T00:00:00Z',
+                exported_at = '2026-05-02T00:00:00Z'
+            WHERE package_id = ?
+            """,
+            (expired_path.stem,),
+        )
+
+    store._prune_expired_history(
+        now=datetime(2026, 7, 23, tzinfo=timezone.utc)
+    )
+
+    assert not expired_path.exists()
+    assert current_path.exists()
+    assert not (outbox_directory / f"{pending_package['package_id']}.json").exists()
+    with sqlite3.connect(database_path) as connection:
+        outbox_rows = connection.execute(
+            "SELECT package_id, exported_at FROM package_outbox ORDER BY package_id"
+        ).fetchall()
+        aggregate_versions = {
+            row[0]
+            for row in connection.execute(
+                "SELECT hermes_version FROM counter_aggregates"
+            ).fetchall()
+        }
+    assert {row[0] for row in outbox_rows} == {
+        current_path.stem,
+        pending_package["package_id"],
+    }
+    assert next(
+        row[1] for row in outbox_rows if row[0] == pending_package["package_id"]
+    ) is None
+    assert aggregate_versions == {"current-version", "pending-version"}
+
+
+def test_retention_failure_does_not_fail_a_committed_export(tmp_path, monkeypatch):
+    store = SharedMetricsStore(
+        tmp_path / "metrics.sqlite3",
+        tmp_path / "outbox",
+    )
+    store.record_model_call(_dimensions(), "test-version")
+
+    def fail_pruning():
+        raise OSError("retention unavailable")
+
+    monkeypatch.setattr(store, "_prune_expired_history", fail_pruning)
+
+    [package_path] = store.create_and_export_package()
+
+    assert package_path.exists()
+    assert store.counter_snapshot()[0]["packaged_value"] == 1
+
+
 def test_file_export_failure_retries_committed_outbox_without_duplicate_delta(
     tmp_path, monkeypatch
 ):