From c9c13fa8f387f582479c153b2136e6b63949799e Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:00:46 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20relay=5Fshared=5Fmetr?= =?UTF-8?q?ics=20=E2=80=94=20=5Fmark()/=5Fguarded()=20collapse=20repeated?= =?UTF-8?q?=20emit=20and=20try/except-log=20blocks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../observability/relay_shared_metrics.py | 135 ++++++++---------- 1 file changed, 60 insertions(+), 75 deletions(-) diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index e5b22cec04..7c3798e3aa 100644 --- a/hermes_cli/observability/relay_shared_metrics.py +++ b/hermes_cli/observability/relay_shared_metrics.py @@ -3,6 +3,7 @@ from __future__ import annotations import atexit +import contextlib import contextvars import logging import threading @@ -171,12 +172,18 @@ class _Runtime: def _emit_client_active(self, session: _MetricsSession) -> None: with session.lock: - if session.closing: - return - self._run_in_session( - session, self.relay.scope.event, CLIENT_ACTIVE_MARK, - handle=session.relay_session.handle, data={}, metadata=self._event_metadata(), - ) + if not session.closing: + self._mark(session, None, CLIENT_ACTIVE_MARK, {}) + + def _mark( + self, session: _MetricsSession, task: _TaskRun | None, name: str, data: dict[str, str] + ) -> None: + """Emit one Relay mark under the task scope when given, else the session scope.""" + handle = task.handle if task is not None else session.relay_session.handle + self._run_scoped( + session, task, self.relay.scope.event, name, + handle=handle, data=data, metadata=self._event_metadata(), + ) def _run_in_session( self, session: _MetricsSession, callback: Callable[..., Any], *args: Any, **kwargs: Any @@ -356,10 +363,8 @@ class _Runtime: if tool_call is not None: tool_call.approval_outcome = outcome attribution = "tool_call" - self._run_in_task( - task, self.relay.scope.event, TOOL_APPROVAL_MARK, handle=task.handle, - data={"attribution": attribution, "outcome": outcome}, - metadata=self._event_metadata(), + self._mark( + session, task, TOOL_APPROVAL_MARK, {"attribution": attribution, "outcome": outcome} ) def record_tool_call(self, event: dict[str, Any]) -> None: @@ -426,10 +431,7 @@ class _Runtime: or not self._event_matches_task_turn(task, event) ): return - self._run_in_task( - task, self.relay.scope.event, mark, - handle=task.handle, data=fields, metadata=self._event_metadata(), - ) + self._mark(session, task, mark, fields) return if session_id and task_id: return @@ -477,19 +479,16 @@ class _Runtime: try: self.relay.subscribers.flush() except Exception as exc: - failure: str | None = f"subscriber flush failed: {exc}" + logger.warning( + "Hermes shared-metrics session %s closed with errors: subscriber flush failed: %s", + session.session_id, + exc, + ) else: - failure = None self._export() with self._sessions_lock: if self._sessions.get(session.session_id) is session: self._sessions.pop(session.session_id, None) - if failure: - logger.warning( - "Hermes shared-metrics session %s closed with errors: %s", - session.session_id, - failure, - ) def shutdown(self) -> None: with self._sessions_lock: @@ -578,12 +577,8 @@ class _Runtime: return None if model_call is None else (model_call_key, model_call) def _run_scoped( - self, - session: _MetricsSession, - task: _TaskRun | None, - callback: Callable[..., Any], - *args: Any, - **kwargs: Any, + self, session: _MetricsSession, task: _TaskRun | None, callback: Callable[..., Any], + *args: Any, **kwargs: Any, ) -> Any: """Run under the task context when the call belongs to a task, else the session.""" if task is not None: @@ -591,13 +586,14 @@ class _Runtime: return self._run_in_session(session, callback, *args, **kwargs) def _flush_and_export(self, failure_message: str) -> None: - try: - self.relay.subscribers.flush() - except Exception: - logger.warning(failure_message, exc_info=True) - else: + """Flush the Relay subscriber, then export; a failed flush skips the export.""" + if self._guarded(failure_message, self._flush_ok): self._export() + def _flush_ok(self) -> bool: + self.relay.subscribers.flush() + return True + def _abort_tasks(self, session: _MetricsSession, base_event: dict[str, Any]) -> None: """Close every open task of a closing session as system-aborted (caller holds the lock).""" for task_id in list(session.tasks): @@ -605,10 +601,8 @@ class _Runtime: self._end_pending_model_calls(session, base_event) def _unregister_atexit(self) -> None: - try: + with contextlib.suppress(Exception): atexit.unregister(self.shutdown) - except Exception: - pass def _task_session( self, event: dict[str, Any], *, allow_task_id_fallback: bool = False @@ -730,13 +724,11 @@ class _Runtime: approval_outcome=tool_call.approval_outcome, fallback_duration_ms=max(0, (monotonic_ns() - tool_call.started_ns) // 1_000_000), ) - try: - self._run_in_task( - task, self.relay.tools.call_end, tool_call.handle, fields, - metadata=self._event_metadata(), - ) - except Exception: - logger.warning("Hermes shared-metrics tool call close failed", exc_info=True) + self._guarded( + "Hermes shared-metrics tool call close failed", + self._run_in_task, task, self.relay.tools.call_end, tool_call.handle, fields, + metadata=self._event_metadata(), + ) def _end_pending_tool_calls( self, session: _MetricsSession, task: _TaskRun, event: dict[str, Any] @@ -753,13 +745,12 @@ class _Runtime: model_call = session.model_calls.pop(model_call_key, None) if model_call is None: return - try: - self._run_scoped( - session, session.tasks.get(model_call.task_id), self.relay.llm.call_end, - model_call.handle, model_call.fields, metadata=self._event_metadata(), - ) - except Exception: - logger.warning("Hermes shared-metrics model call close failed", exc_info=True) + self._guarded( + "Hermes shared-metrics model call close failed", + self._run_scoped, session, session.tasks.get(model_call.task_id), + self.relay.llm.call_end, model_call.handle, model_call.fields, + metadata=self._event_metadata(), + ) def _end_pending_model_calls(self, session: _MetricsSession, event: dict[str, Any]) -> None: task_id = _text(event, "task_id") @@ -804,12 +795,11 @@ class _Runtime: retry_count=task.retry_count, ) try: - self._run_in_task( - task, relay_runtime.pop_relay_scope, self.relay, task.handle, + self._guarded( + "Hermes shared-metrics task close failed", + self._run_in_task, task, relay_runtime.pop_relay_scope, self.relay, task.handle, output=fields, metadata=self._event_metadata(), ) - except Exception: - logger.warning("Hermes shared-metrics task close failed", exc_info=True) finally: session.tasks.pop(task_id, None) session.retired_turn_ids.extend(task.turn_ids) @@ -837,10 +827,10 @@ class _Runtime: Failures must never break the export hook, but are logged at warning: silently failing to close a consent window is a privacy-relevant event. """ - try: - _reconcile_store_consent(self.subscriber.store, send_enabled) - except Exception: - logger.warning("Unable to record a shared-metrics consent transition", exc_info=True) + self._guarded( + "Unable to record a shared-metrics consent transition", + _reconcile_store_consent, self.subscriber.store, send_enabled, + ) def _send_exported_packages(self) -> None: try: @@ -877,12 +867,8 @@ class _Runtime: resolved = _resolved_send_config() return resolved.send and resolved.endpoint == endpoint - try: - SharedMetricsSender( - self.subscriber.store, endpoint, consent_check=still_consented - ).send_pending() - except Exception: - logger.warning("Shared-metrics send pass failed", exc_info=True) + sender = SharedMetricsSender(self.subscriber.store, endpoint, consent_check=still_consented) + self._guarded("Shared-metrics send pass failed", sender.send_pending) def _event_metadata(self) -> dict[str, str]: return { @@ -891,13 +877,18 @@ class _Runtime: } @staticmethod - def _safe(callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + def _guarded(message: str, callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + """Run ``callback``; log-and-swallow any exception, returning None.""" try: return callback(*args, **kwargs) except Exception: - logger.warning("Hermes shared metrics operation failed", exc_info=True) + logger.warning(message, exc_info=True) return None + @classmethod + def _safe(cls, callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + return cls._guarded("Hermes shared metrics operation failed", callback, *args, **kwargs) + def _resolved_send_config(): """Resolve the opt-in send policy from the read-only config snapshot.""" @@ -1047,9 +1038,7 @@ def start_task_run( runtime._safe( runtime.start_task, { - "session_id": session_id, - "task_id": task_id, - "platform": platform, + "session_id": session_id, "task_id": task_id, "platform": platform, "parent_session_id": parent_session_id, }, ) @@ -1092,12 +1081,8 @@ def finish_task_run( runtime._safe( runtime.finish_task, { - "session_id": session_id, - "task_id": task_id, - "platform": platform, - "completed": completed, - "failed": failed, - "interrupted": interrupted, + "session_id": session_id, "task_id": task_id, "platform": platform, + "completed": completed, "failed": failed, "interrupted": interrupted, "turn_exit_reason": reason, }, )