498 lines
16 KiB
Python
498 lines
16 KiB
Python
"""Bounded product contract for the first Hermes shared-metrics slice."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from agent.relay_runtime import (
|
|
LOGICAL_LLM_SCOPE,
|
|
RUNTIME_INSTANCE_KEY,
|
|
RUNTIME_SCHEMA_KEY,
|
|
RUNTIME_SCHEMA_VERSION,
|
|
)
|
|
|
|
SCHEMA_KEY = "hermes.metrics.schema_version"
|
|
SCHEMA_VERSION = "hermes.metrics.event.v2"
|
|
MODEL_CALL_SCOPE = "hermes.model_call"
|
|
MODEL_CALL_PROFILE_MODEL = "unknown"
|
|
TASK_SCOPE = "hermes.task_run"
|
|
SUBSCRIBER_NAME = "hermes.nemo_relay.shared_metrics"
|
|
LEGACY_MODEL_CALL_METRIC = "hermes.model_call.count"
|
|
MODEL_ROUTE_METRIC = "hermes.model_route.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",
|
|
"batch",
|
|
"cli",
|
|
"desktop",
|
|
"gateway",
|
|
"python",
|
|
"scheduled_task",
|
|
"tui",
|
|
"other",
|
|
"unknown",
|
|
})
|
|
TASK_OUTCOMES: frozenset[str] = frozenset({
|
|
"cancelled",
|
|
"failed",
|
|
"success",
|
|
"timed_out",
|
|
"unknown",
|
|
})
|
|
TASK_END_REASONS: frozenset[str] = frozenset({
|
|
"approval_denied",
|
|
"completed",
|
|
"failed",
|
|
"guardrail_blocked",
|
|
"iteration_limit",
|
|
"system_aborted",
|
|
"timed_out",
|
|
"unknown",
|
|
"user_cancelled",
|
|
})
|
|
TASK_TERMINATIONS: frozenset[str] = frozenset({
|
|
"none",
|
|
"system_aborted",
|
|
"timed_out",
|
|
"unknown",
|
|
"user_cancelled",
|
|
})
|
|
TASK_ENTRYPOINTS: frozenset[str] = frozenset({
|
|
"api",
|
|
"background",
|
|
"batch",
|
|
"delegated",
|
|
"gateway_message",
|
|
"interactive",
|
|
"other",
|
|
"python",
|
|
"scheduled_task",
|
|
"unknown",
|
|
})
|
|
DURATION_BUCKETS: frozenset[str] = frozenset({
|
|
"1s_to_5s",
|
|
"2m_to_10m",
|
|
"30s_to_2m",
|
|
"5s_to_30s",
|
|
"gte_10m",
|
|
"lt_1s",
|
|
})
|
|
COUNT_BUCKETS: frozenset[str] = frozenset({
|
|
"0",
|
|
"1",
|
|
"2",
|
|
"3_to_5",
|
|
"6_to_10",
|
|
"gte_11",
|
|
})
|
|
|
|
_LEGACY_PROVIDER_FAMILIES = frozenset({
|
|
"aggregator",
|
|
"custom",
|
|
"direct",
|
|
"local",
|
|
"unknown",
|
|
})
|
|
_LEGACY_MODEL_LOCALITIES = frozenset({"local", "remote", "unknown"})
|
|
_LEGACY_MODEL_OUTCOMES = frozenset({"cancelled", "failed", "success"})
|
|
_LEGACY_MODEL_FAMILIES = frozenset({
|
|
"claude",
|
|
"deepseek",
|
|
"gemini",
|
|
"gemma",
|
|
"glm",
|
|
"gpt",
|
|
"grok",
|
|
"kimi",
|
|
"llama",
|
|
"minimax",
|
|
"mimo",
|
|
"mistral",
|
|
"nemotron",
|
|
"nova",
|
|
"o1",
|
|
"o3",
|
|
"o4",
|
|
"qwen",
|
|
"step",
|
|
"trinity",
|
|
"unknown",
|
|
})
|
|
|
|
_COUNTER_DIMENSION_VALUES: dict[str, dict[str, frozenset[str]]] = {
|
|
# Retained only so pre-v2 pending rows remain packageable.
|
|
LEGACY_MODEL_CALL_METRIC: {
|
|
"call_role": frozenset({"primary"}),
|
|
"locality": _LEGACY_MODEL_LOCALITIES,
|
|
"model_family": _LEGACY_MODEL_FAMILIES,
|
|
"outcome": _LEGACY_MODEL_OUTCOMES,
|
|
"provider_family": _LEGACY_PROVIDER_FAMILIES,
|
|
},
|
|
TASK_STARTED_METRIC: {
|
|
"entrypoint": TASK_ENTRYPOINTS,
|
|
"execution_surface": EXECUTION_SURFACES,
|
|
},
|
|
TASK_FINISHED_METRIC: {
|
|
"duration_bucket": DURATION_BUCKETS,
|
|
"end_reason": TASK_END_REASONS,
|
|
"entrypoint": TASK_ENTRYPOINTS,
|
|
"execution_surface": EXECUTION_SURFACES,
|
|
"model_call_count_bucket": COUNT_BUCKETS,
|
|
"outcome": TASK_OUTCOMES,
|
|
"retry_count_bucket": COUNT_BUCKETS,
|
|
"termination": TASK_TERMINATIONS,
|
|
"tool_call_count_bucket": COUNT_BUCKETS,
|
|
},
|
|
}
|
|
COUNTER_METRICS: frozenset[str] = frozenset({
|
|
MODEL_ROUTE_METRIC,
|
|
TASK_FINISHED_METRIC,
|
|
TASK_STARTED_METRIC,
|
|
})
|
|
|
|
|
|
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_ROUTE_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
|
|
return all(
|
|
isinstance(dimensions[field], str)
|
|
and dimensions[field] in allowed_values
|
|
for field, allowed_values in contract.items()
|
|
)
|
|
|
|
|
|
def model_call_dimensions(event: Any) -> dict[str, str] | None:
|
|
"""Return package dimensions for one valid logical model-call end event."""
|
|
auxiliary = _auxiliary_model_call_dimensions(event)
|
|
if auxiliary is not None:
|
|
return auxiliary
|
|
|
|
metadata = getattr(event, "metadata", None)
|
|
if not isinstance(metadata, dict) or metadata.get(SCHEMA_KEY) != SCHEMA_VERSION:
|
|
return None
|
|
relay_metadata = set(metadata) - {SCHEMA_KEY, RUNTIME_INSTANCE_KEY}
|
|
if relay_metadata - {"otel.status_code"} or metadata.get(
|
|
"otel.status_code", "OK"
|
|
) not in {"OK", "ERROR"}:
|
|
return None
|
|
if (
|
|
str(getattr(event, "kind", "") or "") != "scope"
|
|
or str(getattr(event, "category", "") or "") != "llm"
|
|
or str(getattr(event, "name", "") or "") != MODEL_CALL_SCOPE
|
|
or str(getattr(event, "scope_category", "") or "") != "end"
|
|
):
|
|
return None
|
|
category_profile = getattr(event, "category_profile", None)
|
|
if not isinstance(category_profile, dict) or set(category_profile) != {
|
|
"model_name"
|
|
}:
|
|
return None
|
|
# 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 = {"model", "provider"}
|
|
if not isinstance(data, dict) or set(data) != expected_fields:
|
|
return None
|
|
dimensions = {field: data.get(field) for field in sorted(expected_fields)}
|
|
if not counter_dimensions_are_valid(MODEL_ROUTE_METRIC, dimensions):
|
|
return None
|
|
return dimensions
|
|
|
|
|
|
def _auxiliary_model_call_dimensions(event: Any) -> dict[str, str] | None:
|
|
"""Project a terminal auxiliary route from its Hermes logical scope."""
|
|
metadata = getattr(event, "metadata", None)
|
|
if (
|
|
not isinstance(metadata, dict)
|
|
or metadata.get(RUNTIME_SCHEMA_KEY) != RUNTIME_SCHEMA_VERSION
|
|
):
|
|
return None
|
|
relay_metadata = set(metadata) - {
|
|
RUNTIME_INSTANCE_KEY,
|
|
RUNTIME_SCHEMA_KEY,
|
|
"hermes.call_role",
|
|
}
|
|
if relay_metadata - {"otel.status_code"} or metadata.get(
|
|
"otel.status_code", "OK"
|
|
) not in {"OK", "ERROR"}:
|
|
return None
|
|
call_role = metadata.get("hermes.call_role")
|
|
if not isinstance(call_role, str) or not call_role.startswith("auxiliary:"):
|
|
return None
|
|
if (
|
|
str(getattr(event, "kind", "") or "") != "scope"
|
|
or str(getattr(event, "category", "") or "") != "function"
|
|
or str(getattr(event, "name", "") or "") != LOGICAL_LLM_SCOPE
|
|
or str(getattr(event, "scope_category", "") or "") != "end"
|
|
or getattr(event, "category_profile", None) is not None
|
|
):
|
|
return None
|
|
data = getattr(event, "data", None)
|
|
if (
|
|
not isinstance(data, dict)
|
|
or set(data)
|
|
not in (
|
|
{"model", "outcome", "provider"},
|
|
{"model", "outcome", "provider", "response_model"},
|
|
)
|
|
or data.get("outcome") not in {"cancelled", "failed", "success"}
|
|
):
|
|
return None
|
|
dimensions = model_call_fields(data)
|
|
if not counter_dimensions_are_valid(MODEL_ROUTE_METRIC, dimensions):
|
|
return None
|
|
return dimensions
|
|
|
|
|
|
def task_counter(event: Any) -> tuple[str, dict[str, str]] | None:
|
|
"""Return one validated task counter from a task scope event."""
|
|
metadata = getattr(event, "metadata", None)
|
|
if not isinstance(metadata, dict) or metadata.get(SCHEMA_KEY) != SCHEMA_VERSION:
|
|
return None
|
|
relay_metadata = set(metadata) - {SCHEMA_KEY, RUNTIME_INSTANCE_KEY}
|
|
if relay_metadata - {"otel.status_code"} or metadata.get(
|
|
"otel.status_code", "OK"
|
|
) not in {"OK", "ERROR"}:
|
|
return None
|
|
if (
|
|
str(getattr(event, "kind", "") or "") != "scope"
|
|
or str(getattr(event, "category", "") or "") != "function"
|
|
or str(getattr(event, "name", "") or "") != TASK_SCOPE
|
|
):
|
|
return None
|
|
if getattr(event, "category_profile", None) is not None:
|
|
return None
|
|
|
|
scope_category = str(getattr(event, "scope_category", "") or "")
|
|
data = getattr(event, "data", None)
|
|
if scope_category == "start":
|
|
expected_fields = {"entrypoint", "execution_surface"}
|
|
if not isinstance(data, dict) or set(data) != expected_fields:
|
|
return None
|
|
dimensions = {
|
|
"entrypoint": data.get("entrypoint"),
|
|
"execution_surface": data.get("execution_surface"),
|
|
}
|
|
if not counter_dimensions_are_valid(TASK_STARTED_METRIC, dimensions):
|
|
return None
|
|
return TASK_STARTED_METRIC, dimensions
|
|
|
|
expected_fields = {
|
|
"duration_bucket",
|
|
"end_reason",
|
|
"entrypoint",
|
|
"execution_surface",
|
|
"model_call_count_bucket",
|
|
"outcome",
|
|
"retry_count_bucket",
|
|
"termination",
|
|
"tool_call_count_bucket",
|
|
}
|
|
if (
|
|
scope_category != "end"
|
|
or not isinstance(data, dict)
|
|
or set(data) != expected_fields
|
|
):
|
|
return None
|
|
dimensions = {field: data.get(field) for field in sorted(expected_fields)}
|
|
if not counter_dimensions_are_valid(TASK_FINISHED_METRIC, dimensions):
|
|
return None
|
|
return TASK_FINISHED_METRIC, dimensions
|
|
|
|
|
|
def execution_surface(kwargs: dict[str, Any]) -> str:
|
|
"""Normalize the safe session surface carried by the parent Relay scope."""
|
|
value = (
|
|
str(kwargs.get("execution_surface") or kwargs.get("platform") or "unknown")
|
|
.strip()
|
|
.lower()
|
|
)
|
|
if value in EXECUTION_SURFACES:
|
|
return value
|
|
if value == "api_server":
|
|
return "api"
|
|
if value in {"cron", "scheduler", "scheduled"}:
|
|
return "scheduled_task"
|
|
try:
|
|
from hermes_cli.platforms import get_all_platforms
|
|
|
|
if value in get_all_platforms():
|
|
return "gateway"
|
|
except Exception:
|
|
pass
|
|
if value in {"discord", "email", "slack", "telegram", "teams", "whatsapp"}:
|
|
return "gateway"
|
|
return "unknown" if value == "unknown" else "other"
|
|
|
|
|
|
def task_start_fields(kwargs: dict[str, Any]) -> dict[str, str]:
|
|
"""Build the bounded fields recorded on a task scope start event."""
|
|
surface = execution_surface(kwargs)
|
|
return {
|
|
"entrypoint": task_entrypoint(kwargs, surface),
|
|
"execution_surface": surface,
|
|
}
|
|
|
|
|
|
def task_entrypoint(kwargs: dict[str, Any], surface: str | None = None) -> str:
|
|
"""Normalize the task dispatch owner without exporting source strings."""
|
|
declared = str(kwargs.get("entrypoint") or "").strip().lower()
|
|
if declared in TASK_ENTRYPOINTS:
|
|
return declared
|
|
resolved_surface = surface or execution_surface(kwargs)
|
|
if kwargs.get("parent_task_id") or kwargs.get("parent_session_id"):
|
|
return "delegated"
|
|
return {
|
|
"api": "api",
|
|
"batch": "batch",
|
|
"cli": "interactive",
|
|
"desktop": "interactive",
|
|
"gateway": "gateway_message",
|
|
"python": "python",
|
|
"scheduled_task": "scheduled_task",
|
|
"tui": "interactive",
|
|
"unknown": "unknown",
|
|
}.get(resolved_surface, "other")
|
|
|
|
|
|
def task_terminal_fields(
|
|
kwargs: dict[str, Any],
|
|
*,
|
|
duration_ms: int,
|
|
model_call_count: int,
|
|
tool_call_count: int,
|
|
retry_count: int,
|
|
) -> dict[str, str]:
|
|
"""Build the bounded terminal payload for one task scope."""
|
|
start_fields = task_start_fields(kwargs)
|
|
outcome, end_reason, termination = task_terminal_state(kwargs)
|
|
return {
|
|
**start_fields,
|
|
"duration_bucket": duration_bucket(duration_ms),
|
|
"end_reason": end_reason,
|
|
"model_call_count_bucket": count_bucket(model_call_count),
|
|
"outcome": outcome,
|
|
"retry_count_bucket": count_bucket(retry_count),
|
|
"termination": termination,
|
|
"tool_call_count_bucket": count_bucket(tool_call_count),
|
|
}
|
|
|
|
|
|
def task_terminal_state(kwargs: dict[str, Any]) -> tuple[str, str, str]:
|
|
"""Map Hermes terminal state to bounded task outcome dimensions."""
|
|
reason = str(kwargs.get("turn_exit_reason") or "").strip().lower()
|
|
if kwargs.get("interrupted") or "interrupt" in reason or "cancel" in reason:
|
|
return "cancelled", "user_cancelled", "user_cancelled"
|
|
if "timeout" in reason or "timed_out" in reason:
|
|
return "timed_out", "timed_out", "timed_out"
|
|
if "max_iterations" in reason or "budget_exhausted" in reason:
|
|
return "failed", "iteration_limit", "system_aborted"
|
|
if "approval" in reason and ("denied" in reason or "rejected" in reason):
|
|
return "failed", "approval_denied", "none"
|
|
if "guardrail" in reason:
|
|
return "failed", "guardrail_blocked", "system_aborted"
|
|
if reason == "system_aborted":
|
|
return "failed", "system_aborted", "system_aborted"
|
|
if kwargs.get("completed") is True:
|
|
return "success", "completed", "none"
|
|
if kwargs.get("failed") is True or (reason and reason != "unknown"):
|
|
return "failed", "failed", "none"
|
|
return "unknown", "unknown", "unknown"
|
|
|
|
|
|
def duration_bucket(duration_ms: int) -> str:
|
|
"""Bucket a non-negative task duration into a fixed low-cardinality range."""
|
|
value = max(0, int(duration_ms))
|
|
if value < 1_000:
|
|
return "lt_1s"
|
|
if value < 5_000:
|
|
return "1s_to_5s"
|
|
if value < 30_000:
|
|
return "5s_to_30s"
|
|
if value < 120_000:
|
|
return "30s_to_2m"
|
|
if value < 600_000:
|
|
return "2m_to_10m"
|
|
return "gte_10m"
|
|
|
|
|
|
def count_bucket(count: int) -> str:
|
|
"""Bucket a non-negative per-task count into a fixed range."""
|
|
value = max(0, int(count))
|
|
if value <= 2:
|
|
return str(value)
|
|
if value <= 5:
|
|
return "3_to_5"
|
|
if value <= 10:
|
|
return "6_to_10"
|
|
return "gte_11"
|
|
|
|
|
|
def model_call_fields(kwargs: dict[str, Any]) -> dict[str, str]:
|
|
"""Return the terminal model identity and provider route known to Hermes."""
|
|
model = _metric_identifier(
|
|
kwargs.get("response_model"),
|
|
max_length=MODEL_IDENTIFIER_MAX_LENGTH,
|
|
)
|
|
if model == "unknown":
|
|
model = _metric_identifier(
|
|
kwargs.get("model"),
|
|
max_length=MODEL_IDENTIFIER_MAX_LENGTH,
|
|
)
|
|
return {
|
|
"model": model,
|
|
"provider": _metric_identifier(
|
|
kwargs.get("provider"),
|
|
max_length=PROVIDER_IDENTIFIER_MAX_LENGTH,
|
|
),
|
|
}
|
|
|
|
|
|
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
|