From 1c582c4c4a073e94fe1e4858901aff8a9441349d Mon Sep 17 00:00:00 2001
From: Alex Fournier <afournier@nvidia.com>
Date: Tue, 28 Jul 2026 12:08:10 -0700
Subject: [PATCH] fix(observability): gate export on subscriber flush

Signed-off-by: Alex Fournier <afournier@nvidia.com>
---
 .../observability/relay_shared_metrics.py     | 25 ++++++--
 .../test_relay_shared_metrics_runtime.py      | 60 +++++++++++++++++++
 2 files changed, 80 insertions(+), 5 deletions(-)

diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py
index 2274f8cd4df..c35be04d697 100644
--- a/hermes_cli/observability/relay_shared_metrics.py
+++ b/hermes_cli/observability/relay_shared_metrics.py
@@ -353,8 +353,15 @@ class _Runtime:
                 return
             finished = self._finish_task(session, task_id, event)
         if finished:
-            self._safe(self.relay.subscribers.flush)
-            self._export()
+            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)
@@ -383,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)
@@ -402,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
diff --git a/tests/hermes_cli/test_relay_shared_metrics_runtime.py b/tests/hermes_cli/test_relay_shared_metrics_runtime.py
index 0005d8653e0..f1690dc3926 100644
--- a/tests/hermes_cli/test_relay_shared_metrics_runtime.py
+++ b/tests/hermes_cli/test_relay_shared_metrics_runtime.py
@@ -5,6 +5,7 @@ from __future__ import annotations
 import contextvars
 import asyncio
 import json
+import sqlite3
 import threading
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
@@ -2357,6 +2358,65 @@ def test_desktop_task_completion_exports_once_per_utc_day(
     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):
     lifecycle.invoke_hook(
         "pre_llm_call",