RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(observability): report model and provider usage

This commit is contained in:
Alex Fournier 2026-07-29 08:52:46 -07:00
parent c5d37bb95c
commit dc4714b1e0
8 changed files with 473 additions and 419 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
docs/observability/relay-shared-metrics.md
View File

@ -67,10 +67,13 @@ Hermes turn, API, and tool hooks
Hermes sends an empty `LLMRequest` into the metrics-owned lifecycle. This does
not describe the separate managed-execution call through the native runtime
documented above. The terminal metrics event contains only bounded model
family, provider family, locality, call role, and outcome values. Prompts,
responses, exact model IDs, endpoints, errors, session IDs, task IDs, and
request IDs are not included in the metrics event or package.
documented above. The terminal metrics event contains the model identifier and
provider route that Hermes used for the logical call, such as
`nvidia/nemotron-3-ultra` through `openrouter`. These identifiers are
lowercased and structurally bounded, but they are not normalized through a
checked-in model catalog. Pricing and model-family classification belong to
the metrics backend. Prompts, responses, endpoints, errors, session IDs, task
IDs, and request IDs are not included in the metrics event or package.
Each task run is a Relay `Function` scope named `hermes.task_run`, parented to
the owning Hermes session. The start counter contains only bounded execution
@ -123,5 +126,5 @@ The script uses the installed `nemo-relay` dependency by default. Pass
binding.
The smoke verifies the model request reached the local server, model and task
counters were stored, one package was exported, and prompt, response, and
exact-model canaries are absent from the package.
counters were stored with the expected model and provider, one package was
exported, and prompt and response canaries are absent from the package.

109
hermes_cli/observability/relay_shared_metrics.py
View File

@ -15,13 +15,13 @@ from hermes_cli import __version__
from .shared_metrics import SharedMetricsStore
from .shared_metrics_contract import (
MODEL_CALL_PROFILE_MODEL,
MODEL_CALL_SCOPE,
SCHEMA_KEY,
SCHEMA_VERSION,
SUBSCRIBER_NAME,
TASK_SCOPE,
model_call_fields,
model_call_outcome,
task_start_fields,
task_terminal_fields,
)
@ -81,7 +81,7 @@ class _MetricsSession:
relay_session: relay_runtime.RelaySession
lock: threading.RLock = field(default_factory=threading.RLock, repr=False)
closing: bool = False
model_calls: dict[str, _ModelCall] = field(default_factory=dict)
model_calls: dict[tuple[str, str], _ModelCall] = field(default_factory=dict)
tasks: dict[str, _TaskRun] = field(default_factory=dict)
@ -230,25 +230,26 @@ class _Runtime:
session = self.ensure_session(event)
if session is None:
return
request_id = str(event.get("api_request_id") or "")
if not request_id:
model_call_key = self._new_model_call_key(event)
if model_call_key is None:
return
_, request_id = model_call_key
fields = model_call_fields(event)
retry_ordinal = _retry_ordinal(event)
model_family = fields["model_family"]
with session.lock:
if session.closing:
return
if task is not None:
self._remember_turn(session, task, event)
existing = session.model_calls.get(request_id)
existing = session.model_calls.get(model_call_key)
if existing is not None:
existing.fields = fields
if task is not None:
if retry_ordinal is None or existing.retry_ordinal is None:
task.retry_count += 1
elif retry_ordinal > existing.retry_ordinal:
task.retry_count += retry_ordinal - existing.retry_ordinal
# 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 retry counter.
task.retry_count += 1
if retry_ordinal is not None:
existing.retry_ordinal = max(
existing.retry_ordinal or 0,
@ -268,7 +269,7 @@ class _Runtime:
self.relay.LLMRequest({}, {}),
handle=task.handle,
metadata=self._event_metadata(),
model_name=model_family,
model_name=MODEL_CALL_PROFILE_MODEL,
)
else:
handle = self._run_in_session(
@ -278,15 +279,33 @@ class _Runtime:
self.relay.LLMRequest({}, {}),
handle=session.relay_session.handle,
metadata=self._event_metadata(),
model_name=model_family,
model_name=MODEL_CALL_PROFILE_MODEL,
)
session.model_calls[request_id] = _ModelCall(
session.model_calls[model_call_key] = _ModelCall(
handle=handle,
task_id=str(event.get("task_id") or ""),
fields=fields,
retry_ordinal=retry_ordinal,
)
def record_model_call_error(self, event: dict[str, Any]) -> None:
"""Retain the latest attempt error without closing the logical call."""
session = self._task_session(event, allow_task_id_fallback=True)
if session is None:
session = self._session(event)
if session is None:
return
with session.lock:
if session.closing:
return
model_call_key = self._existing_model_call_key(session, event)
if model_call_key is None:
return
model_call = session.model_calls.get(model_call_key)
if model_call is None:
return
model_call.fields = model_call_fields(event)
def record_tool_call(self, event: dict[str, Any]) -> None:
"""Count one unique tool invocation under its owning task."""
task_id = str(event.get("task_id") or "")
@ -307,25 +326,26 @@ class _Runtime:
else:
task.unidentified_tool_calls += 1
def end_model_call(self, event: dict[str, Any], outcome: str | None = None) -> None:
def end_model_call(self, event: dict[str, Any]) -> None:
session = self._task_session(event, allow_task_id_fallback=True)
if session is None:
session = self._session(event)
if session is None:
return
request_id = str(event.get("api_request_id") or "")
with session.lock:
if session.closing:
return
model_call = session.model_calls.get(request_id)
model_call_key = self._existing_model_call_key(session, event)
if model_call_key is None:
return
model_call = session.model_calls.get(model_call_key)
if model_call is None:
return
fields = model_call_fields(event)
model_call.fields = fields
self._finish_model_call(
session,
request_id,
outcome or model_call_outcome(event),
model_call_key,
)
def end_pending_model_calls(self, event: dict[str, Any]) -> None:
@ -529,10 +549,9 @@ class _Runtime:
def _finish_model_call(
self,
session: _MetricsSession,
request_id: str,
outcome: str,
model_call_key: tuple[str, str],
) -> None:
model_call = session.model_calls.pop(request_id, None)
model_call = session.model_calls.pop(model_call_key, None)
if model_call is None:
return
try:
@ -542,7 +561,7 @@ class _Runtime:
task,
self.relay.llm.call_end,
model_call.handle,
{**model_call.fields, "outcome": outcome},
model_call.fields,
metadata=self._event_metadata(),
)
else:
@ -550,7 +569,7 @@ class _Runtime:
session,
self.relay.llm.call_end,
model_call.handle,
{**model_call.fields, "outcome": outcome},
model_call.fields,
metadata=self._event_metadata(),
)
except Exception:
@ -564,14 +583,41 @@ class _Runtime:
event: dict[str, Any],
) -> None:
task_id = str(event.get("task_id") or "")
request_ids = [
request_id
for request_id, model_call in session.model_calls.items()
model_call_keys = [
model_call_key
for model_call_key, model_call in session.model_calls.items()
if not task_id or model_call.task_id == task_id
]
outcome = "cancelled" if event.get("interrupted") else "failed"
for request_id in request_ids:
self._finish_model_call(session, request_id, outcome)
for model_call_key in model_call_keys:
self._finish_model_call(
session,
model_call_key,
)
@staticmethod
def _new_model_call_key(event: dict[str, Any]) -> tuple[str, str] | None:
request_id = str(event.get("api_request_id") or "")
if not request_id:
return None
return str(event.get("task_id") or ""), request_id
@classmethod
def _existing_model_call_key(
cls,
session: _MetricsSession,
event: dict[str, Any],
) -> tuple[str, str] | None:
key = cls._new_model_call_key(event)
if key is None:
return None
if key in session.model_calls:
return key
if key[0]:
return None
candidates = [
candidate for candidate in session.model_calls if candidate[1] == key[1]
]
return candidates[0] if len(candidates) == 1 else None
def _finish_task(
self,
@ -682,10 +728,9 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None:
elif hook_name == "post_tool_call":
runtime.record_tool_call(kwargs)
elif hook_name == "post_api_request":
runtime.end_model_call(kwargs, "success")
runtime.end_model_call(kwargs)
elif hook_name == "api_request_error":
if kwargs.get("retryable") is False:
runtime.end_model_call(kwargs, "failed")
runtime.record_model_call_error(kwargs)
elif hook_name == "on_session_end":
runtime.finish_task(kwargs)
elif hook_name == "subagent_stop":

66
hermes_cli/observability/schemas/hermes.shared_metrics.v1.schema.json
View File

@ -94,63 +94,21 @@
"type": "object",
"additionalProperties": false,
"required": [
"call_role",
"locality",
"model_family",
"outcome",
"provider_family"
"model",
"provider"
],
"properties": {
"call_role": {
"const": "primary"
"model": {
"type": "string",
"minLength": 1,
"maxLength": 256,
"pattern": "^[a-z0-9][a-z0-9._:/@+\\-]*$"
},
"locality": {
"enum": [
"local",
"remote",
"unknown"
]
},
"model_family": {
"enum": [
"claude",
"deepseek",
"gemini",
"gemma",
"glm",
"gpt",
"grok",
"kimi",
"llama",
"minimax",
"mimo",
"mistral",
"nemotron",
"nova",
"o1",
"o3",
"o4",
"qwen",
"step",
"trinity",
"unknown"
]
},
"outcome": {
"enum": [
"cancelled",
"failed",
"success"
]
},
"provider_family": {
"enum": [
"aggregator",
"custom",
"direct",
"local",
"unknown"
]
"provider": {
"type": "string",
"minLength": 1,
"maxLength": 64,
"pattern": "^[a-z0-9][a-z0-9._:/@+\\-]*$"
}
}
},

239
hermes_cli/observability/shared_metrics_contract.py
View File

@ -2,8 +2,6 @@
from __future__ import annotations
import re
from functools import lru_cache
from typing import Any
from agent.relay_runtime import RUNTIME_INSTANCE_KEY
@ -11,12 +9,20 @@ from agent.relay_runtime import RUNTIME_INSTANCE_KEY
SCHEMA_KEY = "hermes.metrics.schema_version"
SCHEMA_VERSION = "hermes.metrics.event.v1"
MODEL_CALL_SCOPE = "hermes.model_call"
MODEL_CALL_PROFILE_MODEL = "unknown"
TASK_SCOPE = "hermes.task_run"
SUBSCRIBER_NAME = "hermes.nemo_relay.shared_metrics"
PRIMARY_MODEL_CALL_ROLE = "primary"
MODEL_CALL_METRIC = "hermes.model_call.count"
TASK_STARTED_METRIC = "hermes.task_run.started"
TASK_FINISHED_METRIC = "hermes.task_run.finished"
MODEL_IDENTIFIER_MAX_LENGTH = 256
PROVIDER_IDENTIFIER_MAX_LENGTH = 64
_METRIC_IDENTIFIER_CHARACTERS = frozenset(
"abcdefghijklmnopqrstuvwxyz0123456789._:/@+-"
)
_METRIC_IDENTIFIER_START_CHARACTERS = frozenset(
"abcdefghijklmnopqrstuvwxyz0123456789"
)
EXECUTION_SURFACES: frozenset[str] = frozenset({
"api",
@ -30,15 +36,6 @@ EXECUTION_SURFACES: frozenset[str] = frozenset({
"other",
"unknown",
})
PROVIDER_FAMILIES: frozenset[str] = frozenset({
"aggregator",
"custom",
"direct",
"local",
"unknown",
})
MODEL_LOCALITIES: frozenset[str] = frozenset({"local", "remote", "unknown"})
MODEL_OUTCOMES: frozenset[str] = frozenset({"cancelled", "failed", "success"})
TASK_OUTCOMES: frozenset[str] = frozenset({
"cancelled",
"failed",
@ -93,41 +90,7 @@ COUNT_BUCKETS: frozenset[str] = frozenset({
"gte_11",
})
# Shared metrics use an explicit family allowlist rather than raw model IDs or
# dynamically sourced catalog values. The latter would make the exported schema
# drift independently of this contract.
MODEL_FAMILIES: frozenset[str] = frozenset({
"claude",
"deepseek",
"gemini",
"gemma",
"glm",
"gpt",
"grok",
"kimi",
"llama",
"minimax",
"mimo",
"mistral",
"nemotron",
"nova",
"qwen",
"step",
"trinity",
"o1",
"o3",
"o4",
"unknown",
})
_COUNTER_DIMENSION_VALUES: dict[str, dict[str, frozenset[str]]] = {
MODEL_CALL_METRIC: {
"call_role": frozenset({PRIMARY_MODEL_CALL_ROLE}),
"locality": MODEL_LOCALITIES,
"model_family": MODEL_FAMILIES,
"outcome": MODEL_OUTCOMES,
"provider_family": PROVIDER_FAMILIES,
},
TASK_STARTED_METRIC: {
"entrypoint": TASK_ENTRYPOINTS,
"execution_surface": EXECUTION_SURFACES,
@ -144,41 +107,30 @@ _COUNTER_DIMENSION_VALUES: dict[str, dict[str, frozenset[str]]] = {
"tool_call_count_bucket": COUNT_BUCKETS,
},
}
COUNTER_METRICS: frozenset[str] = frozenset(_COUNTER_DIMENSION_VALUES)
_MODEL_FAMILY_PATTERN = re.compile(
r"(?:^|[/_.:-])("
+ "|".join(
re.escape(family)
for family in sorted(
MODEL_FAMILIES - {"unknown"},
key=lambda value: len(value),
reverse=True,
)
)
+ r")(?=$|[/_.:-]|\d)"
COUNTER_METRICS: frozenset[str] = frozenset(
{*_COUNTER_DIMENSION_VALUES, MODEL_CALL_METRIC}
)
# These providers route across model families but are not marked as aggregators
# in Hermes's execution metadata because that flag has narrower routing/catalog
# semantics there.
_TELEMETRY_AGGREGATOR_OVERRIDES = frozenset({
"copilot-acp",
"github-copilot",
"moa",
"nous",
})
# Hermes intentionally resolves these local runtimes through the generic custom
# provider path, so canonical provider metadata cannot distinguish them alone.
_LOCAL_CUSTOM_PROVIDER_ALIASES = frozenset({"mlx", "ollama"})
def counter_dimensions_are_valid(
metric_name: str,
dimensions: dict[str, Any],
) -> bool:
"""Return whether dimensions match one closed shared-metric contract."""
if metric_name == MODEL_CALL_METRIC:
return (
set(dimensions) == {"model", "provider"}
and dimensions["model"]
== _metric_identifier(
dimensions["model"],
max_length=MODEL_IDENTIFIER_MAX_LENGTH,
)
and dimensions["provider"]
== _metric_identifier(
dimensions["provider"],
max_length=PROVIDER_IDENTIFIER_MAX_LENGTH,
)
)
contract = _COUNTER_DIMENSION_VALUES.get(metric_name)
if contract is None or set(dimensions) != set(contract):
return False
@ -211,26 +163,15 @@ def model_call_dimensions(event: Any) -> dict[str, str] | None:
"model_name"
}:
return None
event_model_family = category_profile.get("model_name")
if event_model_family not in MODEL_FAMILIES:
# The synthetic scope can span provider fallback. The accepted terminal
# route is carried in the validated payload rather than this start profile.
if category_profile.get("model_name") != MODEL_CALL_PROFILE_MODEL:
return None
data = getattr(event, "data", None)
expected_fields = {
"call_role",
"locality",
"model_family",
"outcome",
"provider_family",
}
expected_fields = {"model", "provider"}
if not isinstance(data, dict) or set(data) != expected_fields:
return None
dimensions = {
"call_role": data.get("call_role"),
"locality": data.get("locality"),
"model_family": data.get("model_family"),
"outcome": data.get("outcome"),
"provider_family": data.get("provider_family"),
}
dimensions = {field: data.get(field) for field in sorted(expected_fields)}
if not counter_dimensions_are_valid(MODEL_CALL_METRIC, dimensions):
return None
return dimensions
@ -420,97 +361,39 @@ def count_bucket(count: int) -> str:
return "gte_11"
def provider_family(kwargs: dict[str, Any]) -> str:
"""Map a Hermes provider to a bounded product category."""
raw_provider = str(kwargs.get("provider") or "").strip().lower().replace("_", "-")
if not raw_provider:
return "unknown"
if raw_provider in _LOCAL_CUSTOM_PROVIDER_ALIASES:
return "local"
if raw_provider == "custom" or raw_provider.startswith(("custom-", "custom:")):
return "custom"
provider, is_aggregator, is_known = _provider_metadata(raw_provider)
if provider in {"lmstudio", "local"}:
return "local"
if is_aggregator or provider in _TELEMETRY_AGGREGATOR_OVERRIDES:
return "aggregator"
if provider == "custom":
return "custom"
return "direct" if is_known else "unknown"
def _provider_metadata(provider: str) -> tuple[str, bool, bool]:
"""Resolve provider identity without refreshing remote provider metadata."""
try:
from hermes_cli.models import normalize_provider as normalize_model_provider
from hermes_cli.providers import HERMES_OVERLAYS, normalize_provider
canonical = normalize_provider(normalize_model_provider(provider))
overlay = HERMES_OVERLAYS.get(canonical)
return (
canonical,
bool(overlay and overlay.is_aggregator),
canonical in _known_provider_ids(),
)
except Exception:
return provider, False, False
@lru_cache(maxsize=1)
def _known_provider_ids() -> frozenset[str]:
"""Cache Hermes's static provider catalog for the process lifetime."""
try:
from hermes_cli.provider_catalog import provider_catalog_by_slug
return frozenset(provider_catalog_by_slug())
except Exception:
return frozenset()
def model_locality(kwargs: dict[str, Any]) -> str:
"""Classify local endpoints without exporting their URL."""
return _model_locality(kwargs, provider_family(kwargs))
def _model_locality(kwargs: dict[str, Any], provider_category: str) -> str:
base_url = kwargs.get("base_url")
if isinstance(base_url, str) and base_url:
try:
from agent.model_metadata import is_local_endpoint
if is_local_endpoint(base_url):
return "local"
except Exception:
pass
if provider_category == "local":
return "local"
if provider_category in {"aggregator", "direct"}:
return "remote"
return "unknown"
def model_call_fields(kwargs: dict[str, Any]) -> dict[str, str]:
"""Build the bounded producer fields for one logical model call."""
provider_category = provider_family(kwargs)
"""Return the terminal model identity and provider route known to Hermes."""
response_model = kwargs.get("response_model")
model = (
response_model
if isinstance(response_model, str) and response_model.strip()
else kwargs.get("model")
)
return {
"call_role": PRIMARY_MODEL_CALL_ROLE,
"locality": _model_locality(kwargs, provider_category),
"model_family": model_family(kwargs),
"provider_family": provider_category,
"model": _metric_identifier(
model,
max_length=MODEL_IDENTIFIER_MAX_LENGTH,
),
"provider": _metric_identifier(
kwargs.get("provider"),
max_length=PROVIDER_IDENTIFIER_MAX_LENGTH,
),
}
def model_family(kwargs: dict[str, Any]) -> str:
"""Map a raw model identifier to an allowlisted family."""
declared_family = str(kwargs.get("model_family") or "").strip().lower()
if declared_family in MODEL_FAMILIES - {"unknown"}:
return declared_family
model = str(kwargs.get("response_model") or kwargs.get("model") or "").lower()
match = _MODEL_FAMILY_PATTERN.search(model)
return match.group(1) if match is not None else "unknown"
def model_call_outcome(kwargs: dict[str, Any]) -> str:
"""Fail closed when a terminal model-call outcome is not recognized."""
value = str(kwargs.get("outcome") or "").lower()
return value if value in MODEL_OUTCOMES else "failed"
def _metric_identifier(value: Any, *, max_length: int) -> str:
"""Normalize one structurally safe identifier without a product catalog."""
if not isinstance(value, str):
return "unknown"
identifier = value.strip().lower()
if (
not identifier
or len(identifier) > max_length
or identifier[0] not in _METRIC_IDENTIFIER_START_CHARACTERS
or any(
character not in _METRIC_IDENTIFIER_CHARACTERS
for character in identifier
)
):
return "unknown"
return identifier

19
scripts/smoke_nemo_relay_shared_metrics.py
View File

@ -232,11 +232,8 @@ def _validate_store(database_path: Path) -> list[dict[str, Any]]:
expected_model = {
"name": "hermes.model_call.count",
"dimensions": {
"call_role": "primary",
"locality": "local",
"model_family": "gpt",
"outcome": "success",
"provider_family": "custom",
"model": MODEL_CANARY,
"provider": "custom",
},
"value": 1,
"packaged_value": 1,
@ -295,7 +292,7 @@ def _validate_package(outbox: Path, schema_path: Path) -> tuple[Path, dict[str,
jsonschema.validate(package, schema)
serialized = json.dumps(package)
for prohibited in (PROMPT_CANARY, MODEL_CANARY, RESPONSE_CANARY):
for prohibited in (PROMPT_CANARY, RESPONSE_CANARY):
if prohibited in serialized:
raise AssertionError(
f"Exported package leaked prohibited value: {prohibited!r}"
@ -309,12 +306,10 @@ def _validate_package(outbox: Path, schema_path: Path) -> tuple[Path, dict[str,
raise AssertionError(
f"Unexpected package metrics:\n{json.dumps(package.get('metrics'), indent=2)}"
)
if metrics["hermes.model_call.count"]["dimensions"] != {
"call_role": "primary",
"locality": "local",
"model_family": "gpt",
"outcome": "success",
"provider_family": "custom",
model_dimensions = metrics["hermes.model_call.count"]["dimensions"]
if model_dimensions != {
"model": MODEL_CANARY,
"provider": "custom",
}:
raise AssertionError(
f"Unexpected model metric: {metrics['hermes.model_call.count']}"

69
tests/agent/test_auxiliary_relay.py
View File

@ -96,6 +96,75 @@ def test_auxiliary_retries_share_logical_relay_identity(monkeypatch):
]
def test_auxiliary_provider_fallback_closes_one_real_logical_call(
relay_turn,
monkeypatch,
):
relay, turn = relay_turn
consumer = "test.auxiliary-provider-fallback"
turn.lease.host.retain_managed_execution(consumer)
outcomes = []
original_pop = relay.scope.pop
def record_pop(*args, **kwargs):
outcomes.append((kwargs.get("output") or {}).get("outcome"))
return original_pop(*args, **kwargs)
monkeypatch.setattr(relay.scope, "pop", record_pop)
responses = iter([
SimpleNamespace(choices=[]),
SimpleNamespace(
choices=[SimpleNamespace(message=SimpleNamespace(content="recovered"))]
),
])
client = SimpleNamespace(
chat=SimpleNamespace(
completions=SimpleNamespace(
create=lambda **_kwargs: next(responses),
)
)
)
@auxiliary_client._relay_auxiliary_call
def run(task):
auxiliary_client._set_relay_auxiliary_route(
"nvidia",
"nvidia/test-model",
"chat_completions",
)
with pytest.raises(RuntimeError, match="invalid response"):
auxiliary_client._validate_llm_response(
auxiliary_client._relay_sync_completion(
client,
{"model": "nvidia/test-model", "messages": []},
),
task,
)
assert len(turn.logical_llm_calls) == 1
auxiliary_client._set_relay_auxiliary_route(
"openrouter",
"openrouter/test-model",
"chat_completions",
)
return auxiliary_client._validate_llm_response(
auxiliary_client._relay_sync_completion(
client,
{"model": "openrouter/test-model", "messages": []},
),
task,
)
try:
result = run("compression")
finally:
turn.lease.host.release_managed_execution(consumer)
assert result.choices[0].message.content == "recovered"
assert turn.logical_llm_calls == {}
assert outcomes == ["success"]
@pytest.mark.asyncio
async def test_async_auxiliary_attempt_uses_inherited_relay_adapter(monkeypatch):
captured = {}

169
tests/hermes_cli/test_relay_shared_metrics.py
View File

@ -24,11 +24,9 @@ from hermes_cli.observability.shared_metrics_contract import (
COUNT_BUCKETS,
DURATION_BUCKETS,
EXECUTION_SURFACES,
MODEL_FAMILIES,
MODEL_LOCALITIES,
MODEL_OUTCOMES,
PRIMARY_MODEL_CALL_ROLE,
PROVIDER_FAMILIES,
MODEL_CALL_PROFILE_MODEL,
MODEL_IDENTIFIER_MAX_LENGTH,
PROVIDER_IDENTIFIER_MAX_LENGTH,
TASK_END_REASONS,
TASK_ENTRYPOINTS,
TASK_OUTCOMES,
@ -36,11 +34,8 @@ from hermes_cli.observability.shared_metrics_contract import (
count_bucket,
duration_bucket,
execution_surface,
model_call_outcome,
model_call_dimensions,
model_family,
model_locality,
provider_family,
model_call_fields,
task_counter,
task_start_fields,
task_terminal_fields,
@ -79,11 +74,8 @@ def _task_dimension_schema(kind: str) -> dict[str, object]:
def _dimensions() -> dict[str, str]:
return {
"call_role": PRIMARY_MODEL_CALL_ROLE,
"locality": "remote",
"model_family": "claude",
"outcome": "success",
"provider_family": "direct",
"model": "anthropic/claude-sonnet-4.6",
"provider": "openrouter",
}
@ -183,11 +175,11 @@ def test_due_export_runs_once_per_utc_day_and_catches_up_pending_deltas(
def test_package_schema_matches_the_model_call_contract():
properties = _package_dimension_schema()["properties"]
assert properties["call_role"] == {"const": PRIMARY_MODEL_CALL_ROLE}
assert set(properties["locality"]["enum"]) == MODEL_LOCALITIES
assert set(properties["model_family"]["enum"]) == MODEL_FAMILIES
assert set(properties["outcome"]["enum"]) == MODEL_OUTCOMES
assert set(properties["provider_family"]["enum"]) == PROVIDER_FAMILIES
assert set(properties) == {"model", "provider"}
assert properties["model"]["maxLength"] == MODEL_IDENTIFIER_MAX_LENGTH
assert properties["provider"]["maxLength"] == PROVIDER_IDENTIFIER_MAX_LENGTH
assert "enum" not in properties["model"]
assert "enum" not in properties["provider"]
def test_package_schema_matches_the_task_contract():
@ -205,89 +197,45 @@ def test_package_schema_matches_the_task_contract():
assert set(terminal["termination"]["enum"]) == TASK_TERMINATIONS
@pytest.mark.parametrize(
("provider", "expected"),
[
("", "unknown"),
("not-a-hermes-provider", "unknown"),
("custom", "custom"),
("custom-local", "custom"),
("custom:private-endpoint", "custom"),
("lmstudio", "local"),
("lm_studio", "local"),
("ollama", "local"),
("nous", "aggregator"),
("openrouter", "aggregator"),
("kilo", "aggregator"),
("copilot-acp", "aggregator"),
("huggingface", "aggregator"),
("novita", "aggregator"),
("anthropic", "direct"),
("google", "direct"),
("openai-api", "direct"),
],
)
def test_provider_family_uses_bounded_product_categories(provider, expected):
assert provider_family({"provider": provider}) == expected
def test_provider_family_does_not_resolve_live_provider_metadata(monkeypatch):
def fail_live_lookup(_provider):
raise AssertionError("telemetry must not refresh provider metadata")
monkeypatch.setattr("hermes_cli.providers.get_provider", fail_live_lookup)
assert provider_family({"provider": "anthropic"}) == "direct"
def test_locality_uses_the_endpoint_only_for_local_classification():
kwargs = {
"provider": "custom",
"base_url": "http://127.0.0.1:11434/v1",
def test_model_call_fields_report_terminal_model_and_provider_without_a_catalog():
assert model_call_fields({
"model": "fallback/model",
"response_model": "NVIDIA/Nemotron-3-Ultra",
"provider": "OpenRouter",
"base_url": "https://private-endpoint.example/v1",
}) == {
"model": "nvidia/nemotron-3-ultra",
"provider": "openrouter",
}
assert model_call_fields({
"model": "ZAI/GLM-5.2",
"provider": "Brev",
}) == {
"model": "zai/glm-5.2",
"provider": "brev",
}
assert provider_family(kwargs) == "custom"
assert model_locality(kwargs) == "local"
@pytest.mark.parametrize(
("model", "expected"),
("field", "value"),
[
("google/gemma-3", "gemma"),
("x-ai/grok-4", "grok"),
("minimax/minimax-m2.5", "minimax"),
("xiaomi/mimo-v2", "mimo"),
("amazon/nova-pro", "nova"),
("stepfun/step-3.5", "step"),
("arcee-ai/trinity-large", "trinity"),
("model", ""),
("model", "contains a space"),
("model", "contains\ncontrol"),
("model", "_" + "private"),
("model", "x" * (MODEL_IDENTIFIER_MAX_LENGTH + 1)),
("model", object()),
("provider", ""),
("provider", "private provider"),
("provider", "x" * (PROVIDER_IDENTIFIER_MAX_LENGTH + 1)),
("provider", object()),
],
)
def test_model_family_covers_families_evidenced_by_the_hermes_catalog(model, expected):
assert model_family({"model": model}) == expected
def test_model_call_fields_collapse_malformed_identifiers(field, value):
event = {"model": "nvidia/nemotron-3-ultra", "provider": "openrouter"}
event[field] = value
@pytest.mark.parametrize(
"model",
[
"private-gptish-model",
"innovation-private",
"mimosa-private",
"stepstone-private",
"supernova-private",
],
)
def test_model_family_requires_identifier_boundaries(model):
assert model_family({"model": model}) == "unknown"
def test_model_family_accepts_only_allowlisted_declared_metadata():
assert model_family({"model": "private", "model_family": "qwen"}) == "qwen"
assert model_family({"model": "private", "model_family": "private"}) == "unknown"
def test_model_family_prefers_the_provider_reported_terminal_model():
assert (
model_family({"model": "gpt-5", "response_model": "claude-sonnet"}) == "claude"
)
assert model_call_fields(event)[field] == "unknown"
@pytest.mark.parametrize(
@ -404,41 +352,26 @@ def test_task_terminal_state_is_bounded(event, expected):
assert task_terminal_state(event) == expected
def test_model_outcome_fails_closed_to_a_bounded_value():
assert model_call_outcome({"outcome": "private"}) == "failed"
def test_unlisted_model_collapses_to_a_bounded_value():
assert model_family({"model": "private-model-name"}) == "unknown"
def test_subscriber_contract_rejects_unknown_fields_and_dimension_values():
event = SimpleNamespace(
kind="scope",
category="llm",
category_profile={"model_name": "gpt"},
category_profile={"model_name": MODEL_CALL_PROFILE_MODEL},
name="hermes.model_call",
scope_category="end",
metadata={"hermes.metrics.schema_version": "hermes.metrics.event.v1"},
data={
"call_role": "primary",
"locality": "remote",
"model_family": "gpt",
"outcome": "success",
"provider_family": "direct",
},
data=_dimensions(),
)
assert model_call_dimensions(event) == {
"call_role": "primary",
"locality": "remote",
"model_family": "gpt",
"outcome": "success",
"provider_family": "direct",
}
assert model_call_dimensions(event) == _dimensions()
event.category_profile["model_name"] = "gpt"
assert model_call_dimensions(event) is None
event.category_profile["model_name"] = "private-model-name"
assert model_call_dimensions(event) is None
event.category_profile["model_name"] = "gpt"
event.category_profile["model_name"] = MODEL_CALL_PROFILE_MODEL
event.data["model"] = "contains a space"
assert model_call_dimensions(event) is None
event.data["model"] = _dimensions()["model"]
event.data["prompt"] = "must-not-pass"
assert model_call_dimensions(event) is None
event.data.pop("prompt")

206
tests/hermes_cli/test_relay_shared_metrics_runtime.py
View File

@ -258,13 +258,10 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa
assert len(starts) == 1
assert len(ends) == 1
assert starts[0][2] == {}
assert starts[0][3]["model_name"] == "gpt"
assert starts[0][3]["model_name"] == "unknown"
assert ends[0][2] == {
"call_role": "primary",
"locality": "remote",
"model_family": "claude",
"outcome": "success",
"provider_family": "direct",
"model": "claude-sonnet",
"provider": "anthropic",
}
serialized_events = json.dumps(direct_runtime.events)
assert "sensitive-prompt" not in serialized_events
@ -286,7 +283,10 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa
"hermes.task_run.finished",
"hermes.task_run.started",
}
assert metrics["hermes.model_call.count"]["dimensions"]["model_family"] == "claude"
assert metrics["hermes.model_call.count"]["dimensions"] == {
"model": "claude-sonnet",
"provider": "anthropic",
}
assert metrics["hermes.model_call.count"]["value"] == 1
assert metrics["hermes.task_run.started"] == {
"name": "hermes.task_run.started",
@ -432,10 +432,13 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot(
assert len(by_metric["hermes.task_run.started"]) == 1
assert by_metric["hermes.task_run.started"][0]["value"] == 3
assert {
counter["dimensions"]["outcome"]
for counter in by_metric["hermes.model_call.count"]
} == {"success", "failed", "cancelled"}
assert len(by_metric["hermes.model_call.count"]) == 1
model_counter = by_metric["hermes.model_call.count"][0]
assert model_counter["dimensions"] == {
"model": model_canary,
"provider": "custom",
}
assert model_counter["value"] == 3
terminal_by_outcome = {
counter["dimensions"]["outcome"]: counter
for counter in by_metric["hermes.task_run.finished"]
@ -475,10 +478,10 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot(
"snapshot": snapshot,
"packages": package_payloads,
})
assert model_canary in serialized_analytics
for canary in (
prompt_canary,
response_canary,
model_canary,
tool_canary,
"sensitive-session",
"sensitive-task",
@ -1230,8 +1233,8 @@ def test_shared_metrics_keys_turn_ownership_by_session(direct_runtime):
assert task_b is not None
assert session_a is not None
assert session_b is not None
assert "request-a" in session_a.model_calls
assert "request-a" not in session_b.model_calls
assert ("task-a", "request-a") in session_a.model_calls
assert ("task-a", "request-a") not in session_b.model_calls
[model_start] = [
event for event in direct_runtime.events if event[0] == "llm.call"
]
@ -2006,7 +2009,7 @@ def test_core_runtime_ignores_self_parenting_subagent_event(direct_runtime):
assert session.parent_session_id == ""
def test_terminal_model_error_is_counted_as_failed(direct_runtime):
def test_terminal_model_error_retains_the_failed_route(direct_runtime):
base = {
"session_id": "s1",
"task_id": "t1",
@ -2016,11 +2019,110 @@ def test_terminal_model_error_is_counted_as_failed(direct_runtime):
}
lifecycle.invoke_hook("pre_api_request", **base)
lifecycle.invoke_hook("api_request_error", **base, retryable=False)
lifecycle.invoke_hook(
"api_request_error",
**base,
retryable=False,
error={"message": "sensitive-error"},
)
assert not [event for event in direct_runtime.events if event[0] == "llm.call_end"]
runtime = relay_shared_metrics._get_runtime()
session = runtime._session(base)
assert session is not None
[model_call] = session.model_calls.values()
assert model_call.fields == {
"model": "claude-sonnet",
"provider": "anthropic",
}
lifecycle.finalize_session(session_id="s1")
[end] = [event for event in direct_runtime.events if event[0] == "llm.call_end"]
assert end[2]["outcome"] == "failed"
assert end[2] == {
"model": "claude-sonnet",
"provider": "anthropic",
}
def test_nonretryable_provider_error_can_recover_within_one_logical_call(
direct_runtime,
):
base = {
"session_id": "s1",
"task_id": "t1",
"api_request_id": "r1",
"provider": "anthropic",
"model": "claude-sonnet",
}
lifecycle.invoke_hook("pre_api_request", **base, retry_count=0)
lifecycle.invoke_hook(
"api_request_error",
**base,
retry_count=0,
retryable=False,
)
fallback = {
**base,
"provider": "openai-api",
"model": "gpt-5",
}
lifecycle.invoke_hook("pre_api_request", **fallback, retry_count=0)
lifecycle.invoke_hook(
"post_api_request",
**fallback,
retry_count=0,
)
lifecycle.finalize_session(session_id="s1")
[end] = [event for event in direct_runtime.events if event[0] == "llm.call_end"]
[start] = [event for event in direct_runtime.events if event[0] == "llm.call"]
assert start[3]["model_name"] == "unknown"
assert end[2] == {
"model": "gpt-5",
"provider": "openai-api",
}
def test_same_request_id_is_isolated_between_tasks(direct_runtime):
common = {
"session_id": "s1",
"api_request_id": "shared-request",
"platform": "cli",
"provider": "anthropic",
"model": "claude-sonnet",
}
for task_id in ("t1", "t2"):
lifecycle.invoke_hook("pre_llm_call", **common, task_id=task_id)
lifecycle.invoke_hook("pre_api_request", **common, task_id=task_id)
lifecycle.invoke_hook("post_api_request", **common)
assert not [event for event in direct_runtime.events if event[0] == "llm.call_end"]
for task_id in ("t2", "t1"):
lifecycle.invoke_hook("post_api_request", **common, task_id=task_id)
lifecycle.invoke_hook(
"on_session_end",
**common,
task_id=task_id,
completed=True,
failed=False,
interrupted=False,
turn_exit_reason="text_response(stop)",
)
lifecycle.finalize_session(session_id="s1")
model_ends = [
event for event in direct_runtime.events if event[0] == "llm.call_end"
]
assert len(model_ends) == 2
task_ends = [
event[2]["output"]
for event in direct_runtime.events
if event[0] == "scope.pop" and event[1][1] == "hermes.task_run"
]
assert len(task_ends) == 2
assert all(fields["model_call_count_bucket"] == "1" for fields in task_ends)
assert all(fields["retry_count_bucket"] == "0" for fields in task_ends)
def test_task_terminal_counts_logical_calls_retries_and_unique_tools(direct_runtime):
@ -2065,7 +2167,10 @@ def test_task_terminal_counts_logical_calls_retries_and_unique_tools(direct_runt
]
assert len(model_starts) == 1
assert len(model_ends) == 1
assert model_ends[0][2]["outcome"] == "failed"
assert model_ends[0][2] == {
"model": "nvidia/nemotron-3-super-120b-a12b",
"provider": "nvidia",
}
[task_end] = [
event
for event in direct_runtime.events
@ -2136,6 +2241,66 @@ def test_task_terminal_counts_explicit_retry_with_new_request_id(direct_runtime)
assert task_end[2]["output"]["retry_count_bucket"] == "1"
def test_task_retry_count_survives_provider_fallback_ordinal_reset(direct_runtime):
base = {
"session_id": "s1",
"task_id": "t1",
"api_request_id": "r1",
"platform": "cli",
"provider": "nvidia",
"model": "nvidia/nemotron-3-super-120b-a12b",
}
lifecycle.invoke_hook("pre_llm_call", **base)
lifecycle.invoke_hook("pre_api_request", **base, retry_count=0)
lifecycle.invoke_hook(
"api_request_error",
**base,
retry_count=0,
retryable=True,
)
lifecycle.invoke_hook("pre_api_request", **base, retry_count=1)
lifecycle.invoke_hook(
"api_request_error",
**base,
retry_count=1,
retryable=True,
)
lifecycle.invoke_hook(
"pre_api_request",
**{**base, "provider": "openai", "model": "gpt-5"},
retry_count=0,
)
lifecycle.invoke_hook(
"post_api_request",
**{**base, "provider": "openai", "model": "gpt-5"},
retry_count=0,
)
lifecycle.invoke_hook(
"on_session_end",
**base,
completed=True,
failed=False,
interrupted=False,
turn_exit_reason="text_response(stop)",
)
lifecycle.finalize_session(session_id="s1")
[model_end] = [
event for event in direct_runtime.events if event[0] == "llm.call_end"
]
assert model_end[2] == {
"model": "gpt-5",
"provider": "openai",
}
[task_end] = [
event
for event in direct_runtime.events
if event[0] == "scope.pop" and event[1][1] == "hermes.task_run"
]
assert task_end[2]["output"]["retry_count_bucket"] == "2"
def test_outer_agent_boundary_closes_early_returns_and_exceptions(
direct_runtime,
monkeypatch,
@ -2470,7 +2635,10 @@ def test_task_ownership_survives_session_id_rotation(direct_runtime):
assert len(task_starts) == 1
assert len(task_ends) == 1
assert len(model_ends) == 1
assert model_ends[0][2]["outcome"] == "success"
assert model_ends[0][2] == {
"model": "nvidia/nemotron-3-super-120b-a12b",
"provider": "nvidia",
}
assert task_ends[0][2]["output"]["model_call_count_bucket"] == "1"
assert task_ends[0][2]["output"]["outcome"] == "success"