c13ea774e6
Runtime identity resolved through hermes_cli.__version__ (a static 0.0.0 on source installs, rewritten by release stamping) leaked v0.0.0 into About, /api/health, User-Agents, and plugin compat, and source updates showed "couldn't reach update server" because identity and channel authority disagreed with the checkout. Now: get_version_info() resolves install stamp -> live git -> unknown, never pyproject metadata, never a package constant. Source checkouts derive identity from their reachable release tag; the completion tail of every successful install/update/historical takeover atomically rewrites install-stamp.json with that identity; a stale source stamp whose commit no longer matches HEAD defers to live git. ACP/TUI use derived_version for display and base_version for protocol fields; all ~44 runtime __version__ consumers migrated; hermes_cli.__version__ and generated _version.py are gone; release stamping only touches the native manifests external builders consume (nix/tauri/cargo) and passes release identity straight into write_install_stamp.py; pyproject.toml stays inert 0.0.0. Desktop no longer synthesizes a competing install-stamp.json: the checkout owns its stamp, and desktop-bootstrap classification keys on the bootstrap-complete marker. verify-bootstrap-version-stamp.py now cross-checks the checkout's stamp (baseVersion + commit == HEAD). Validation: 31-file focused suite green (version identity, stamping, adoption, providers, gateway, acp/tui runtime identity, api server via extras env, release graph); desktop tsc + 25 vitest green; real-repo probe: base=unknown derived=git.0635606.dirty source=git on this checkout; clean-env imports resolve entirely from this tree; windows footgun + compat-pointer scans clean.
1094 lines
46 KiB
Python
1094 lines
46 KiB
Python
"""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.version_info import get_version_info
|
|
|
|
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(), get_version_info().base_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 ----
|