diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index 72e6c2e8a4..7eb36c9a8b 100644 --- a/hermes_cli/observability/relay_shared_metrics.py +++ b/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: