fix(telemetry): move shared metrics flush off turn path

hermes_cli/observability/relay_shared_metrics.py

@@ -228,6 +228,11 @@ class _Runtime:
         # Guards the opt-in send pass: at most one in flight per process.
         self._send_lock = threading.RLock()
         self._send_thread: threading.Thread | None = None
+        # Flushing is a process-wide Relay barrier. Keep it off interactive turn threads while
+        # coalescing concurrent finish hooks into one worker.
+        self._flush_lock = threading.RLock()
+        self._flush_pending = False
+        self._flush_thread: threading.Thread | None = None
         self._snapshot_checked_ns: int | None = None
         self._subscriber_name = f"{SUBSCRIBER_NAME}.{self.host.runtime_id}"
         self.subscriber = SharedMetricsSubscriber(
@@ -554,7 +559,7 @@ class _Runtime:
         if retired:
             self.close_session({"session_id": session.session_id})
         elif finished:
-            self._flush_and_export("Hermes shared-metrics task flush failed")
+            self._schedule_flush_and_export("Hermes shared-metrics task flush failed")
 
     def close_session(self, event: dict[str, Any]) -> None:
         session = self._session(event)
@@ -566,16 +571,9 @@ class _Runtime:
         ):
             return
         self._emit_session_summary(session)
-        try:
-            self.relay.subscribers.flush()
-        except Exception as exc:
-            logger.warning(
-                "Hermes shared-metrics session %s closed with errors: subscriber flush failed: %s",
-                session.session_id,
-                exc,
-            )
-        else:
-            self._export()
+        self._schedule_flush_and_export(
+            f"Hermes shared-metrics session {session.session_id} flush failed"
+        )
         with self._sessions_lock:
             _forget(self._sessions, session.session_id, session)
 
@@ -670,6 +668,27 @@ class _Runtime:
             return self._run_in_task(task, callback, *args, **kwargs)
         return self.host.run_in_session(session.relay_session, callback, *args, **kwargs)
 
+    def _schedule_flush_and_export(self, failure_message: str) -> None:
+        """Schedule the process-wide Relay flush without blocking the finishing turn."""
+        with self._flush_lock:
+            self._flush_pending = True
+            if self._flush_thread is None or not self._flush_thread.is_alive():
+                self._flush_thread = threading.Thread(
+                    target=self._run_flush_worker,
+                    args=(failure_message,),
+                    name="hermes-shared-metrics-flush",
+                    daemon=True,
+                )
+                self._flush_thread.start()
+
+    def _run_flush_worker(self, failure_message: str) -> None:
+        while True:
+            with self._flush_lock:
+                if not self._flush_pending:
+                    return
+                self._flush_pending = False
+            self._flush_and_export(failure_message)
+
     def _flush_and_export(self, failure_message: str) -> None:
         """Flush the Relay subscriber, then export; a failed flush skips the export."""
         try: