"""Direct NeMo Relay integration for Hermes shared client metrics.""" from __future__ import annotations import atexit import contextlib import contextvars import logging import threading from collections import deque from dataclasses import dataclass, field from time import monotonic_ns from typing import Any, Callable from agent import relay_runtime from hermes_cli import __version__ from .shared_metrics import SharedMetricsStore from . import shared_metrics_contract as contract from .shared_metrics_contract import MODEL_CALL_SCOPE, SUBSCRIBER_NAME, TASK_SCOPE from .shared_metrics_subscriber import SharedMetricsSubscriber logger = logging.getLogger(__name__) _RUNTIME_FAILED = object() _RUNTIMES: dict[str, _Runtime | object] = {} _RUNTIME_LOCK = threading.RLock() _ABORTED = {"failed": True, "turn_exit_reason": "system_aborted"} def _text(event: dict[str, Any], key: str) -> str: return str(event.get(key) or "") def _session_pair(event: dict[str, Any], key: str) -> tuple[str, str] | None: """(session_id, event[key]) when both are non-empty.""" session_id, value = _text(event, "session_id"), _text(event, key) return (session_id, value) if session_id and value else 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") 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: """Drop ``key`` from ``index`` only while it still points at ``owner``.""" if index.get(key) is owner: index.pop(key, None) def _task_parent_handle(session: _MetricsSession, task_id: str) -> Any: """The active turn's handle when it owns this exact task, else the session handle.""" active_turn = relay_runtime.active_turn(session.session_id) if ( active_turn is not None and active_turn.lease.session_id == session.session_id and active_turn.task_id == task_id and active_turn.handle is not None ): return active_turn.handle return session.relay_session.handle def _elapsed_ms(started_ns: int) -> int: return max(0, (monotonic_ns() - started_ns) // 1_000_000) def _scope_handle(session: _MetricsSession, task: _TaskRun | None) -> Any: return task.handle if task is not None else session.relay_session.handle def _sole(items: Any) -> Any: """The single distinct element of ``items`` (identity-deduplicated), else None.""" unique = {id(item): item for item in items} return next(iter(unique.values())) if len(unique) == 1 else None def _identities_compatible(candidate: tuple[str, str, str], observed: tuple[str, str, str]) -> bool: """Match partial hook context without crossing known call boundaries.""" if not observed[2] or candidate[2] != observed[2]: return False return all( not left or not right or left == right for left, right in zip(candidate[:2], observed[:2], strict=True) ) def _compatible_tool_call_keys( session: _MetricsSession, task_id: str, identity: tuple[str, str, str] ) -> list[tuple[str, str, str, str]]: return [ key for key in session.tool_calls if key[0] == task_id and _identities_compatible(key[1:], identity) ] @dataclass class _ModelCall: handle: Any task_id: str fields: dict[str, str] @dataclass class _ToolCall: handle: Any category: str started_ns: int approval_outcome: str = "not_required" @dataclass class _TaskRun: task_id: str handle: Any context: contextvars.Context started_ns: int start_fields: dict[str, str] model_call_ids: set[str] = field(default_factory=set) tool_call_ids: set[tuple[str, str, str]] = field(default_factory=set) turn_ids: set[str] = field(default_factory=set) retired_turn_ids: frozenset[str] = field(default_factory=frozenset) completed_tool_call_ids: set[tuple[str, str, str]] = field(default_factory=set) unidentified_tool_calls: int = 0 retry_count: int = 0 @dataclass class _MetricsSession: session_id: str relay_session: relay_runtime.RelaySession lock: threading.RLock = field(default_factory=threading.RLock, repr=False) closing: bool = False model_calls: dict[tuple[str, str], _ModelCall] = field(default_factory=dict) tasks: dict[str, _TaskRun] = field(default_factory=dict) tool_calls: dict[tuple[str, str, str, str], _ToolCall] = field(default_factory=dict) retired_turn_ids: deque[str] = field(default_factory=lambda: deque(maxlen=256)) class _Runtime: """Own shared-metrics state layered on the Hermes core Relay host.""" def __init__(self, host: relay_runtime.RelayRuntime | None = None) -> None: resolved_host = host or relay_runtime.get_runtime() if resolved_host is None: raise RuntimeError("Hermes core Relay runtime is unavailable") self.host: relay_runtime.RelayRuntime = resolved_host self.relay = self.host.relay self._active = True self._sessions: dict[str, _MetricsSession] = {} self._task_sessions: dict[tuple[str, str], _MetricsSession] = {} self._turn_sessions: dict[tuple[str, str], _MetricsSession] = {} self._sessions_lock = threading.RLock() self._task_creation_lock = threading.RLock() self._task_sessions_lock = threading.RLock() # 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 self._subscriber_name = f"{SUBSCRIBER_NAME}.{self.host.runtime_id}" self.subscriber = SharedMetricsSubscriber( SharedMetricsStore(), __version__, runtime_id=self.host.runtime_id ) self.relay.subscribers.register(self._subscriber_name, self.subscriber) self.host.retain_managed_execution(self._subscriber_name) self._registered = True atexit.register(self.shutdown) def ensure_session(self, event: dict[str, Any]) -> _MetricsSession | None: session_id = _text(event, "session_id") if not session_id: return None with self._sessions_lock: if not self._active: return None relay_session = self.host.ensure_session(event) if relay_session is None: return None session = self._sessions.get(session_id) if session is None: session = _MetricsSession(session_id=session_id, relay_session=relay_session) self._sessions[session_id] = session with session.lock: 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.""" session = self.ensure_session(event) if session is not None: self._emit_client_active(session) def _emit_client_active(self, session: _MetricsSession) -> None: with session.lock: if not session.closing: self._mark(session, None, contract.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.""" self._run_scoped( session, task, self.relay.scope.event, name, handle=_scope_handle(session, task), data=data, metadata=self._event_metadata(), ) def start_task(self, event: dict[str, Any]) -> _TaskRun | None: """Open one Relay function scope for a Hermes task run.""" task_key = _session_pair(event, "task_id") if task_key is None: return None _, task_id = task_key with self._task_creation_lock: owner = self._task_session(event) if owner is not None: with owner.lock: if owner.closing: return None task = owner.tasks.get(task_id) if task is not None and not self._admits(owner, task, event): return None return task session = self.ensure_session(event) if session is None: return None with session.lock: turn_id = _text(event, "turn_id") if ( session.closing or (turn_id and turn_id in session.retired_turn_ids) or session.relay_session.context is None ): return None self._emit_client_active(session) task_context = session.relay_session.context.copy() start_fields = contract.task_start_fields(event) handle = task_context.run( self._with_scope_stack, self.relay.scope.push, TASK_SCOPE, self.relay.ScopeType.Function, handle=_task_parent_handle(session, task_id), input=start_fields, metadata=self._event_metadata(), ) task = _TaskRun( task_id=task_id, handle=handle, context=task_context, started_ns=monotonic_ns(), start_fields=start_fields, retired_turn_ids=frozenset(session.retired_turn_ids), ) session.tasks[task_id] = task with self._task_sessions_lock: self._task_sessions[task_key] = session self._remember_turn(session, task, event) return task def _run_in_task( self, task: _TaskRun, callback: Callable[..., Any], *args: Any, **kwargs: Any ) -> Any: return task.context.copy().run(self._with_scope_stack, callback, *args, **kwargs) def _with_scope_stack(self, callback: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: self.relay.get_scope_stack() return callback(*args, **kwargs) def start_model_call(self, event: dict[str, Any]) -> None: task_id = _text(event, "task_id") session, task = self._task_pair(event, start=True, allow_task_id_fallback=True) if task_id and task is None: return session = session or self.ensure_session(event) if session is None: return request_id = _text(event, "api_request_id") if not request_id: return model_call_key = (task_id, request_id) fields = contract.model_call_fields(event) with session.lock: if session.closing: return if task is not None and not self._admits(session, task, event, current=True): return existing = session.model_calls.get(model_call_key) if existing is not None: existing.fields = fields if task is not None: # Every repeated start for one logical request is another physical # attempt. Provider fallback resets Hermes's provider-local retry # ordinal, so ordinal deltas are not a reliable task-level counter. task.retry_count += 1 return if task is not None: task.model_call_ids.add(request_id) 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 handle = self._run_scoped( session, task, self.relay.llm.call, MODEL_CALL_SCOPE, self.relay.LLMRequest({}, {}), handle=_scope_handle(session, task), metadata=self._event_metadata(), model_name=contract.MODEL_CALL_PROFILE_MODEL, ) session.model_calls[model_call_key] = _ModelCall(handle, task_id, fields) def update_model_call(self, event: dict[str, Any], *, finish: bool) -> None: """Refresh the located model call's fields from ``event``; ``finish`` closes it. ``api_request_error`` retains the latest attempt error without closing the logical call; ``post_api_request`` closes it. """ session = self._any_session(event) if session is None: return with session.lock: if session.closing: return model_call_key = self._existing_model_call_key(session, event) model_call = session.model_calls.get(model_call_key) if model_call_key else None if model_call is None: return model_call.fields = contract.model_call_fields(event) if finish: self._finish_model_call(session, model_call_key) def start_tool_call(self, event: dict[str, Any]) -> None: """Open one privacy-safe Relay tool lifecycle under its task.""" task_id = _text(event, "task_id") session, task = self._task_pair(event, start=True, allow_task_id_fallback=True) if session is None or task is None or not _text(event, "tool_call_id"): return identity = self._tool_call_identity(event) with session.lock: if not self._admits(session, task, event): return key = (task_id, *identity) if identity in task.completed_tool_call_ids or key in session.tool_calls: return task.tool_call_ids.add(identity) session.tool_calls[key] = self._open_tool_call(task, event) def record_approval(self, event: dict[str, Any]) -> None: """Record one bounded approval result without approval text or commands.""" session, task = self._approval_task(event) if session is None or task is None: return outcome = contract.tool_approval_outcome(event) attribution = "unattributed" with session.lock: if session.closing or not self._event_matches_task_turn(task, event): return if _text(event, "tool_call_id"): identity = self._tool_call_identity(event) tool_call = session.tool_calls.get((task.task_id, *identity)) if tool_call is None: key = _sole(_compatible_tool_call_keys(session, task.task_id, identity)) tool_call = session.tool_calls[key] if key is not None else None if tool_call is not None: tool_call.approval_outcome = outcome attribution = "tool_call" self._mark( session, task, contract.TOOL_APPROVAL_MARK, {"attribution": attribution, "outcome": outcome}, ) def record_tool_call(self, event: dict[str, Any]) -> None: """Close and count one unique privacy-safe tool lifecycle.""" task_id = _text(event, "task_id") session, task = self._task_pair(event, allow_task_id_fallback=True) if session is None or task is None: return with session.lock: if not self._admits(session, task, event): return tool_call = None if _text(event, "tool_call_id"): observed_identity = self._tool_call_identity(event) if observed_identity in task.completed_tool_call_ids: return identity = observed_identity tool_call = session.tool_calls.pop((task_id, *identity), None) if tool_call is None: if any( _identities_compatible(completed, observed_identity) for completed in task.completed_tool_call_ids ): return matching_keys = _compatible_tool_call_keys(session, task_id, observed_identity) if len(matching_keys) > 1: # Partial context cannot safely choose between concurrent calls # that reused the provider-local ID. return if matching_keys: identity = matching_keys[0][1:] tool_call = session.tool_calls.pop(matching_keys[0]) task.completed_tool_call_ids.update({identity, observed_identity}) task.tool_call_ids.add(identity) else: task.unidentified_tool_calls += 1 if tool_call is None: tool_call = self._open_tool_call(task, event) self._finish_tool_call(task, tool_call, event) def record_skill_lifecycle(self, event: dict[str, Any]) -> None: """Emit one allowlisted skill fact without its local identity.""" if _text(event, "action").strip().lower() == "loaded": mark, fields = contract.SKILL_LOAD_MARK, contract.skill_load_fields(event) else: mark, fields = contract.SKILL_LIFECYCLE_MARK, contract.skill_lifecycle_fields(event) if fields is None: return session_id, task_id = _text(event, "session_id"), _text(event, "task_id") session, task = self._task_pair(event, allow_task_id_fallback=not session_id) if session is None: if not (session_id and task_id): # No owning task: a bare process-level mark. self._with_scope_stack( self.relay.scope.event, mark, data=fields, metadata=self._event_metadata() ) return if task is None: return with session.lock: if ( not session.closing and session.tasks.get(task.task_id) is task and self._event_matches_task_turn(task, event) ): self._mark(session, task, mark, fields) def finish_task(self, event: dict[str, Any]) -> None: """Close one task scope exactly once with bounded terminal fields.""" session = self._any_session(event) if session is None: return with session.lock: finished = not session.closing and self._finish_task( session, _text(event, "task_id"), event ) if finished: self._flush_and_export("Hermes shared-metrics task flush failed") def close_session(self, event: dict[str, Any]) -> None: session = self._session(event) if session is None: return if not self._abort_session( session, {**event, **_ABORTED, "completed": False, "interrupted": False} ): return 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() with self._sessions_lock: _forget(self._sessions, session.session_id, session) def shutdown(self) -> None: with self._sessions_lock: self._active = False session_ids = list(self._sessions) for session_id in session_ids: self._safe(self.close_session, {"session_id": session_id}) if not self._registered: return self._flush_and_export("Hermes shared-metrics shutdown flush failed") self._deregister() self._release() def _deregister(self) -> None: self._safe(self.relay.subscribers.deregister, self._subscriber_name) self.host.release_managed_execution(self._subscriber_name) self._registered = False def deactivate(self) -> None: """Stop collection without exporting locally aggregated metrics.""" with self._sessions_lock: self._active = False self.subscriber.deactivate() if self._registered: self._deregister() with self._sessions_lock: sessions = list(self._sessions.values()) for session in sessions: self._abort_session(session, {"session_id": session.session_id, **_ABORTED}) with self._sessions_lock: self._sessions.clear() with self._task_sessions_lock: self._task_sessions.clear() self._turn_sessions.clear() self._release() def _release(self) -> None: """Let an in-flight send finish briefly, then drop the atexit hook. A short-lived CLI process would otherwise exit and kill the daemon send thread mid-request — the common case for this feature's one cadence. """ self._join_send_thread() with contextlib.suppress(Exception): atexit.unregister(self.shutdown) def _join_send_thread(self, timeout: float = 2.0) -> None: """Bounded on purpose: pending packages stay in SQLite and go out next run, so blocking on a slow network is the wrong trade; the daemon thread dies with the process.""" with self._send_lock: thread = self._send_thread if thread is not None and thread.is_alive(): try: thread.join(timeout) except Exception: logger.debug("Shared-metrics send thread join failed", exc_info=True) def _session(self, event: dict[str, Any]) -> _MetricsSession | None: with self._sessions_lock: return self._sessions.get(_text(event, "session_id")) def _any_session(self, event: dict[str, Any]) -> _MetricsSession | None: """Owner session by task/turn correlation, else by session_id.""" return self._task_session(event, allow_task_id_fallback=True) or self._session(event) def _task_pair( self, event: dict[str, Any], *, start: bool = False, **lookup: Any ) -> tuple[_MetricsSession | None, _TaskRun | None]: """Resolve (session, task) for a task-scoped hook; ``start`` opens a missing task.""" session = self._task_session(event, **lookup) task = session.tasks.get(_text(event, "task_id")) if session is not None else None if task is None and start: task = self.start_task(event) session = self._task_session(event) if task is not None else None return session, task def _run_scoped( 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: return self._run_in_task(task, callback, *args, **kwargs) return self.host.run_in_session(session.relay_session, callback, *args, **kwargs) def _flush_and_export(self, failure_message: str) -> None: """Flush the Relay subscriber, then export; a failed flush skips the export.""" try: self.relay.subscribers.flush() except Exception: logger.warning(failure_message, exc_info=True) else: self._export() 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: if session.closing: return False session.closing = True for task_id in list(session.tasks): self._finish_task(session, task_id, {**base_event, "task_id": task_id}) self._end_pending_model_calls(session, base_event) return True 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 with self._task_sessions_lock: 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 or not allow_task_id_fallback: return owner return _sole( 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 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]: """Identify one provider-local tool call without exporting its IDs.""" return _text(event, "api_request_id"), _text(event, "turn_id"), _text(event, "tool_call_id") @staticmethod def _event_matches_task_turn(task: _TaskRun, event: dict[str, Any]) -> bool: """Reject delayed hooks from a prior run that reused the task ID.""" turn_id = _text(event, "turn_id") if not turn_id: return True return turn_id not in task.retired_turn_ids and ( not task.turn_ids or turn_id in task.turn_ids ) def _admits( self, session: _MetricsSession, task: _TaskRun, event: dict[str, Any], *, current: bool = False, ) -> bool: """Whether ``event`` may act on ``task`` (caller holds ``session.lock``). 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) or (current and session.tasks.get(task.task_id) is not task) ): return False self._remember_turn(session, task, event) return True def _approval_task( self, event: dict[str, Any] ) -> tuple[_MetricsSession | None, _TaskRun | None]: """Resolve approval correlation without guessing across ambiguous turns.""" active = relay_runtime.active_turn() if active is not None: session, task = self._task_pair( {**event, "session_id": active.lease.session_id, "task_id": active.task_id} ) if task is not None: return session, task session, task = self._task_pair(event) if task is not None: return session, task turn_id = _text(event, "turn_id") session = None if turn_id: with self._task_sessions_lock: session = _sole( candidate for (owner_id, candidate_turn_id), candidate in self._turn_sessions.items() if candidate_turn_id == turn_id and self._sessions.get(owner_id) is candidate ) if session is None: return None, None task = _sole(task for task in session.tasks.values() if turn_id in task.turn_ids) return (None, None) if task is None else (session, task) def _open_tool_call(self, task: _TaskRun, event: dict[str, Any]) -> _ToolCall: handle = self._run_in_task( task, self.relay.tools.call, contract.TOOL_CALL_SCOPE, {}, handle=task.handle, metadata=self._event_metadata(), ) return _ToolCall(handle, contract.tool_category(event), monotonic_ns()) def _finish_tool_call( self, task: _TaskRun, tool_call: _ToolCall, event: dict[str, Any] ) -> None: fields = contract.tool_terminal_fields( event, category=tool_call.category, approval_outcome=tool_call.approval_outcome, fallback_duration_ms=_elapsed_ms(tool_call.started_ns), ) self._guarded( "Hermes shared-metrics tool call close failed", lambda: self._run_in_task( task, self.relay.tools.call_end, tool_call.handle, self.relay.ToolExecutionResult(fields), metadata=self._event_metadata(), ), ) def _end_pending_tool_calls( self, session: _MetricsSession, task: _TaskRun, event: dict[str, Any] ) -> None: task_outcome, _, _ = contract.task_terminal_state(event) status = {"cancelled": "cancelled", "timed_out": "timeout"}.get(task_outcome, "error") for key in [key for key in session.tool_calls if key[0] == task.task_id]: self._finish_tool_call(task, session.tool_calls.pop(key), {**event, "status": status}) def _finish_model_call(self, session: _MetricsSession, model_call_key: tuple[str, str]) -> None: model_call = session.model_calls.pop(model_call_key, None) if model_call is None: return 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") pending = [k for k, c in session.model_calls.items() if not task_id or c.task_id == task_id] for key in pending: self._finish_model_call(session, key) @staticmethod def _existing_model_call_key( session: _MetricsSession, event: dict[str, Any] ) -> tuple[str, str] | None: """(task_id, request_id) of an open call; a task-less event may match by request alone.""" request_id = _text(event, "api_request_id") if not request_id: return None key = (_text(event, "task_id"), request_id) if key in session.model_calls or key[0]: return key if key in session.model_calls else None candidates = [candidate for candidate in session.model_calls if candidate[1] == request_id] return candidates[0] if len(candidates) == 1 else None def _finish_task(self, session: _MetricsSession, task_id: str, event: dict[str, Any]) -> bool: task = session.tasks.get(task_id) if task is None: return False self._end_pending_tool_calls(session, task, event) self._end_pending_model_calls(session, {**event, "task_id": task_id}) fields = contract.task_terminal_fields( {**task.start_fields, **event}, duration_ms=_elapsed_ms(task.started_ns), model_call_count=len(task.model_call_ids), tool_call_count=len(task.tool_call_ids) + task.unidentified_tool_calls, retry_count=task.retry_count, ) try: popped = self._guarded( "Hermes shared-metrics task close failed", self._run_in_task, task, relay_runtime.pop_relay_scope_if_top, self.relay, task.handle, output=fields, metadata=self._event_metadata(), ) if popped is False: logger.debug("Left shared-metrics task scope %s under a concurrent turn's scope; session close drains it", task_id) finally: session.tasks.pop(task_id, None) session.retired_turn_ids.extend(task.turn_ids) with self._task_sessions_lock: _forget(self._task_sessions, (session.session_id, task_id), session) for turn_id in task.turn_ids: _forget(self._turn_sessions, (session.session_id, turn_id), session) return True def _export(self) -> None: exported = self._safe(self.subscriber.store.create_and_export_package_if_due) # Sending must never delay the caller: _export runs on finish_task, the user's # interactive path. The thread is about latency, not correctness. if exported is not None: self._safe(self._send_exported_packages) def _send_exported_packages(self) -> None: try: resolved = _resolved_send_config() except Exception: logger.debug("Unable to read shared-metrics send policy", exc_info=True) return # Observe the consent EDGE before deciding whether to send: the dominant revocation # case is "sending turned off while no pass is running", invisible to the send loop. # Failures never break the export hook but log at warning (privacy-relevant). self._guarded( "Unable to record a shared-metrics consent transition", _reconcile_store_consent, self.subscriber.store, resolved.send, ) if not resolved.send: return with self._send_lock: # One in-flight pass per process; the next hook fire picks up what is pending. if self._send_thread is None or not self._send_thread.is_alive(): self._send_thread = threading.Thread( target=self._run_send_pass, args=(resolved.endpoint,), name="hermes-shared-metrics-send", daemon=True, ) self._send_thread.start() def _run_send_pass(self, endpoint: str) -> None: from hermes_cli.observability.shared_metrics_sender import SharedMetricsSender def still_consented() -> bool: """Re-read consent so revoking `send` stops an in-flight pass.""" resolved = _resolved_send_config() return resolved.send and resolved.endpoint == endpoint 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 { contract.SCHEMA_KEY: contract.SCHEMA_VERSION, relay_runtime.RUNTIME_INSTANCE_KEY: self.host.runtime_id, } @staticmethod 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(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 _raw_config() -> dict[str, Any]: """Read-only config snapshot (lazy import: tests patch ``hermes_cli.config``). Collection consent is profile-owned: managed overlays cannot opt a profile in or out. The read-only path matters because this gate runs 2-3x per agent turn and the mutable read_raw_config() paid a full config deepcopy on every call. """ from hermes_cli.config import read_raw_config_readonly return read_raw_config_readonly() or {} def _resolved_send_config(): """Resolve the opt-in send policy from the read-only config snapshot.""" from hermes_cli.observability.shared_metrics_send_config import resolve_send_config return resolve_send_config(_raw_config()) def _reconcile_store_consent(store: SharedMetricsStore, send_enabled: bool) -> None: from hermes_cli.observability.shared_metrics_sender import reconcile_send_consent from hermes_cli.sqlite_util import write_txn with store._connection() as connection: with write_txn(connection): reconcile_send_consent(connection, send_enabled) def enabled() -> bool: """Return the shared-metrics policy for the active Hermes profile.""" profile_key = relay_runtime.current_profile_key() try: config: Any = _raw_config() except Exception: logger.debug("Unable to read Hermes shared-metrics policy", exc_info=True) config = None for key in ("telemetry", "shared_metrics"): config = config.get(key) if isinstance(config, dict) else None if isinstance(config, dict) and config.get("enabled") is True: return True with _RUNTIME_LOCK: runtime = _RUNTIMES.pop(profile_key, None) if isinstance(runtime, _Runtime): runtime.deactivate() return False def handles_hook(hook_name: str) -> bool: return hook_name in HANDLED_HOOKS and enabled() _consent_reconcile_done = False def _reconcile_send_consent_once() -> None: """Reconcile consent windows with config, once per process. Runs BEFORE and INDEPENDENT of the collection gate, so a user with ``enabled: false`` still gets send-consent windows reconciled. Skipped only when there is no store on disk AND consent is off: nothing to protect, and creating ``~/.hermes/telemetry`` for every fully-disabled user would be the wrong behaviour change. """ global _consent_reconcile_done if _consent_reconcile_done: return _consent_reconcile_done = True try: # Lazy: tests patch ``shared_metrics.SharedMetricsStore`` at its origin. from hermes_cli.observability.shared_metrics import SharedMetricsStore from hermes_constants import get_hermes_home resolved = _resolved_send_config() # Probe WITHOUT constructing a store: the constructor creates the directory and # schema as a side effect, which would make the skip below dead code. default_path = get_hermes_home() / "telemetry" / "shared_metrics" / "metrics.sqlite3" if not resolved.send and not default_path.exists(): return _reconcile_store_consent(SharedMetricsStore(), resolved.send) except Exception: logger.warning("Unable to reconcile shared-metrics send consent", exc_info=True) def observe_lifecycle(hook_name: str, **kwargs: Any) -> None: """Project one Hermes lifecycle event into the core Relay integration.""" _reconcile_send_consent_once() if not handles_hook(hook_name) or not relay_runtime.relay_instrumentation_enabled(): return runtime = _get_runtime() if runtime is None: return try: _HOOK_HANDLERS[hook_name](runtime, kwargs) except Exception: logger.warning("Hermes shared metrics hook failed: %s", hook_name, exc_info=True) def _with_runtime_toolset(event: dict[str, Any]) -> dict[str, Any]: """Attach the toolset already declared by Hermes's runtime registry.""" tool_name = _text(event, "tool_name") if event.get("toolset") or not tool_name: return event try: from model_tools import get_toolset_for_tool toolset = get_toolset_for_tool(tool_name) except Exception: toolset = None return {**event, "toolset": toolset or "other"} def _close_child_session(runtime: _Runtime, kwargs: dict[str, Any]) -> None: child_session_id = _text(kwargs, "child_session_id") if child_session_id: runtime.close_session({"session_id": child_session_id}) _HOOK_HANDLERS: dict[str, Callable[[_Runtime, dict[str, Any]], Any]] = { "on_session_start": lambda rt, kw: rt.record_client_active(kw), "pre_llm_call": lambda rt, kw: rt.start_task(kw), "pre_api_request": lambda rt, kw: rt.start_model_call(kw), "pre_tool_call": lambda rt, kw: rt.start_tool_call(_with_runtime_toolset(kw)), "post_tool_call": lambda rt, kw: rt.record_tool_call(_with_runtime_toolset(kw)), "post_approval_response": lambda rt, kw: rt.record_approval(kw), "on_skill_lifecycle": lambda rt, kw: rt.record_skill_lifecycle(kw), "post_api_request": lambda rt, kw: rt.update_model_call(kw, finish=True), "api_request_error": lambda rt, kw: rt.update_model_call(kw, finish=False), "on_session_end": lambda rt, kw: rt.finish_task(kw), "subagent_stop": _close_child_session, "on_session_finalize": lambda rt, kw: rt.close_session(kw), "on_session_reset": lambda rt, kw: rt.close_session(kw), } HANDLED_HOOKS = frozenset(_HOOK_HANDLERS) def _prepare_core_session(host: relay_runtime.RelayRuntime, context: dict[str, Any]) -> None: """Prepare the profile subscriber before the coordinator opens a scope.""" del context if host.profile_key == relay_runtime.current_profile_key() and enabled(): _get_runtime(retry_failed=True, host=host) def start_task_run( *, session_id: str, task_id: str, platform: str, parent_session_id: str = "" ) -> None: """Start task metrics at the outer Hermes execution boundary.""" _run_task_hook( "start_task", retry_failed=True, session_id=session_id, task_id=task_id, platform=platform, parent_session_id=parent_session_id, ) def finish_task_run( *, session_id: str, task_id: str, platform: str, result: dict[str, Any] | None = None, error: BaseException | None = None, ) -> None: """Finish task metrics for every return or exception path.""" _run_task_hook( "finish_task", session_id=session_id, task_id=task_id, platform=platform, **_terminal_flags(result, error), ) def _run_task_hook(method: str, *, retry_failed: bool = False, **event: Any) -> None: if not enabled(): return runtime = _get_runtime(retry_failed=retry_failed) if runtime is not None: runtime._safe(getattr(runtime, method), event) def _terminal_flags(result: dict[str, Any] | None, error: BaseException | None) -> dict[str, Any]: """Bounded completed/failed/interrupted/turn_exit_reason for a task's return or raise.""" if error is not None: interrupted = ( isinstance(error, (KeyboardInterrupt, InterruptedError)) or type(error).__name__ == "CancelledError" ) if interrupted: reason = "interrupted_by_user" else: reason = "timed_out" if isinstance(error, TimeoutError) else "system_aborted" return { "completed": False, "failed": not interrupted, "interrupted": interrupted, "turn_exit_reason": reason, } terminal = result if isinstance(result, dict) else {} failed = terminal.get("failed") is True reason = str(terminal.get("turn_exit_reason") or terminal.get("failure_reason") or "") return { "completed": terminal.get("completed") is True, "failed": failed, "interrupted": terminal.get("interrupted") is True, "turn_exit_reason": reason or ("failed" if failed else "unknown"), } def _get_runtime( *, retry_failed: bool = False, host: relay_runtime.RelayRuntime | None = None ) -> _Runtime | None: profile_key = relay_runtime.current_profile_key() with _RUNTIME_LOCK: runtime = _RUNTIMES.get(profile_key) if isinstance(runtime, _Runtime): if host is None or runtime.host is host: return runtime runtime.deactivate() elif runtime is _RUNTIME_FAILED and not retry_failed: return None try: _RUNTIMES[profile_key] = runtime = _Runtime(host=host) except Exception: logger.warning("Hermes shared metrics initialization failed", exc_info=True) _RUNTIMES[profile_key] = _RUNTIME_FAILED return None return runtime relay_runtime.SESSION_COORDINATOR.register_session_initializer( SUBSCRIBER_NAME, _prepare_core_session ) def _reset_for_tests() -> None: """Reset all profile-scoped shared-metrics state for isolated tests.""" with _RUNTIME_LOCK: runtimes = list(_RUNTIMES.values()) _RUNTIMES.clear() for runtime in runtimes: if isinstance(runtime, _Runtime): runtime.shutdown() # ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ---- # Names external plugins imported from this module before the Sep 2026 decomposition. # Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does). # The whole block is removed by reverting the commit that added it. def prepare_session_start() -> None: """Register the subscriber before any producer opens the session scope.""" if enabled(): _get_runtime(retry_failed=True) _PLUGIN_COMPAT_LAZY = { 'CLIENT_ACTIVE_MARK': ('hermes_cli.observability.shared_metrics_contract', 'CLIENT_ACTIVE_MARK'), 'MODEL_CALL_PROFILE_MODEL': ('hermes_cli.observability.shared_metrics_contract', 'MODEL_CALL_PROFILE_MODEL'), 'SCHEMA_KEY': ('hermes_cli.observability.shared_metrics_contract', 'SCHEMA_KEY'), 'SCHEMA_VERSION': ('hermes_cli.observability.shared_metrics_contract', 'SCHEMA_VERSION'), 'SKILL_LIFECYCLE_MARK': ('hermes_cli.observability.shared_metrics_contract', 'SKILL_LIFECYCLE_MARK'), 'SKILL_LOAD_MARK': ('hermes_cli.observability.shared_metrics_contract', 'SKILL_LOAD_MARK'), 'TOOL_APPROVAL_MARK': ('hermes_cli.observability.shared_metrics_contract', 'TOOL_APPROVAL_MARK'), 'TOOL_CALL_SCOPE': ('hermes_cli.observability.shared_metrics_contract', 'TOOL_CALL_SCOPE'), 'model_call_fields': ('hermes_cli.observability.shared_metrics_contract', 'model_call_fields'), 'skill_lifecycle_fields': ('hermes_cli.observability.shared_metrics_contract', 'skill_lifecycle_fields'), 'skill_load_fields': ('hermes_cli.observability.shared_metrics_contract', 'skill_load_fields'), 'task_start_fields': ('hermes_cli.observability.shared_metrics_contract', 'task_start_fields'), 'task_terminal_fields': ('hermes_cli.observability.shared_metrics_contract', 'task_terminal_fields'), 'task_terminal_state': ('hermes_cli.observability.shared_metrics_contract', 'task_terminal_state'), 'tool_approval_outcome': ('hermes_cli.observability.shared_metrics_contract', 'tool_approval_outcome'), 'tool_category': ('hermes_cli.observability.shared_metrics_contract', 'tool_category'), 'tool_terminal_fields': ('hermes_cli.observability.shared_metrics_contract', 'tool_terminal_fields'), } def __getattr__(name): # PEP 562 — lazy so no import cycles target = _PLUGIN_COMPAT_LAZY.get(name) if target is None: raise AttributeError(f"module {__name__!r} has no attribute {name!r}") import importlib from hermes_cli.plugin_compat import warn_once warn_once(__name__, name, *target) return getattr(importlib.import_module(target[0]), target[1]) # ---- END PLUGIN-COMPAT ----