diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py index fb6e6bd5e2..98cdf547f1 100644 --- a/hermes_cli/observability/relay_shared_metrics.py +++ b/hermes_cli/observability/relay_shared_metrics.py @@ -51,11 +51,10 @@ def _session_pair(event: dict[str, Any], key: str) -> tuple[str, str] | None: return (session_id, value) if session_id and value else None -def _retry_ordinal(event: dict[str, Any]) -> int | None: +def _retry_ordinal(event: dict[str, Any]) -> int: + """Hermes's provider-local retry ordinal; 0 when absent or malformed.""" value = event.get("retry_count") - if isinstance(value, int) and not isinstance(value, bool) and value >= 0: - return value - return None + return value if isinstance(value, int) and not isinstance(value, bool) and value > 0 else 0 def _forget(index: dict[Any, _MetricsSession], key: Any, owner: _MetricsSession) -> None: @@ -193,9 +192,7 @@ class _Runtime: session = _MetricsSession(session_id=session_id, relay_session=relay_session) self._sessions[session_id] = session with session.lock: - if session.closing: - return None - return session + return None if session.closing else session def record_client_active(self, event: dict[str, Any]) -> None: """Emit one payload-free activation attempt under the session scope.""" @@ -291,7 +288,6 @@ class _Runtime: return model_call_key = (task_id, request_id) fields = contract.model_call_fields(event) - retry_ordinal = _retry_ordinal(event) with session.lock: if session.closing: return @@ -308,7 +304,7 @@ class _Runtime: return if task is not None: task.model_call_ids.add(request_id) - if retry_ordinal is not None and retry_ordinal > 0: + if _retry_ordinal(event) > 0: # A real Hermes retry can advance api_request_id while carrying the # retry ordinal. Count that physical attempt. task.retry_count += 1 @@ -572,13 +568,13 @@ class _Runtime: def _flush_and_export(self, failure_message: str) -> None: """Flush the Relay subscriber, then export; a failed flush skips the export.""" - if self._guarded(failure_message, self._flush_ok): + try: + self.relay.subscribers.flush() + except Exception: + logger.warning(failure_message, exc_info=True) + else: self._export() - def _flush_ok(self) -> bool: - self.relay.subscribers.flush() - return True - def _abort_session(self, session: _MetricsSession, base_event: dict[str, Any]) -> bool: """Mark the session closing and system-abort its open tasks; False if already closing.""" with session.lock: @@ -593,33 +589,28 @@ class _Runtime: def _task_session( self, event: dict[str, Any], *, allow_task_id_fallback: bool = False ) -> _MetricsSession | None: + """Owner session by (session, turn), then (session, task), then unique task_id.""" session_id, task_id = _text(event, "session_id"), _text(event, "task_id") if not task_id: return None - turn_key = _session_pair(event, "turn_id") with self._task_sessions_lock: - owner = self._turn_sessions.get(turn_key) if turn_key is not None else None + owner = self._turn_sessions.get(_session_pair(event, "turn_id")) if owner is None and session_id: owner = self._task_sessions.get((session_id, task_id)) - if owner is not None: + if owner is not None or not allow_task_id_fallback: return owner - if not allow_task_id_fallback: - return None return _sole( - session - for (_, candidate_task_id), session in self._task_sessions.items() - if candidate_task_id == task_id + session for (_, tid), session in self._task_sessions.items() if tid == task_id ) def _remember_turn( self, session: _MetricsSession, task: _TaskRun, event: dict[str, Any] ) -> None: turn_id = _text(event, "turn_id") - if not turn_id: - return - task.turn_ids.add(turn_id) - with self._task_sessions_lock: - self._turn_sessions[(session.session_id, turn_id)] = session + if turn_id: + task.turn_ids.add(turn_id) + with self._task_sessions_lock: + self._turn_sessions[(session.session_id, turn_id)] = session @staticmethod def _tool_call_identity(event: dict[str, Any]) -> tuple[str, str, str]: @@ -632,9 +623,9 @@ class _Runtime: turn_id = _text(event, "turn_id") if not turn_id: return True - if turn_id in task.retired_turn_ids: - return False - return not task.turn_ids or turn_id in task.turn_ids + return turn_id not in task.retired_turn_ids and ( + not task.turn_ids or turn_id in task.turn_ids + ) def _admits( self, @@ -649,9 +640,11 @@ class _Runtime: Rejects closing sessions and stale turns; with ``current`` also requires ``task`` to still be the session's live run for its ID. Admitted events have their turn remembered. """ - if session.closing or not self._event_matches_task_turn(task, event): - return False - if current and session.tasks.get(task.task_id) is not task: + if ( + session.closing + or not self._event_matches_task_turn(task, event) + or (current and session.tasks.get(task.task_id) is not task) + ): return False self._remember_turn(session, task, event) return True