mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-31 19:16:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

feat(telemetry): local-first telemetry & observability

Browse source 
Add a built-in telemetry system that records what the agent does — workflows,
model calls, tool calls, errors — to the local machine, powers `/insights`, and
can export to an operator-chosen destination. Default-on locally; nothing leaves
the machine unless the user exports it or opts into the aggregate plane.

Three planes with a hard wall between them:
  - local: full-fidelity observability (real model/provider/tool names), on by
    default, never leaves the machine.
  - aggregate: opt-in metadata, default off. No uploader ships — consent is
    recorded via telemetry.consent_state, and `preview` shows what would be
    produced, computed locally.
  - trajectories: full message content, opt-in, exported only to the operator's
    own destination.

Mechanism:
  - Bundled `telemetry` plugin registers observational lifecycle hooks
    (on_session_start / post_api_request / post_tool_call / on_session_finalize).
    No core call sites are edited; hooks already carry the data.
  - Fire-and-forget emitter: emit() returns in microseconds, never blocks or
    raises into a model/tool call. A daemon thread writes events to an
    append-only JSONL log and the tel_* tables in state.db (its own sqlite
    connection, separate from SessionDB).
  - tel_runs / tel_model_calls / tel_tool_calls live in the declarative
    SCHEMA_SQL and are reconciled automatically; SCHEMA_VERSION 16 -> 17.
  - metrics derives rollups for /usage and /insights; rollup builds per-run
    summaries for `hermes telemetry preview`.

Consent is config, not a parallel command surface. The config file is the root
of trust: set telemetry.consent_state with `hermes config set`, or pin any
telemetry.* key (including allow_aggregate) via managed scope, which overrides
the user's value per key. `hermes telemetry` exposes only what config cannot:
status (report), preview (query), and export.

Export:
  - exporter_bulk writes telemetry (and, when the trajectories plane is enabled,
    session content) to ndjson/json.
  - otlp_exporter streams spans to a configured OpenTelemetry Collector over
    OTLP/HTTP. The SDK is an optional extra (hermes-agent[otlp]), lazily
    installed via tools.lazy_deps on first use.
  - Secrets are always redacted on every export path
    (redact_sensitive_text(force=True)); content export is gated by the
    trajectories plane, and PII scrubbing follows telemetry.content_redaction.
    OTLP auth headers reference environment variable names, never inline values.

No outbound emission to Nous. The aggregate uploader is intentionally not built.
This commit is contained in:
emozilla 2026-06-24 02:14:02 -04:00
parent 64131bf975
commit ccfa079252
37 changed files with 3920 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

319
plugins/telemetry/__init__.py Normal file
View file

@ -0,0 +1,319 @@
"""Telemetry plugin — wires Hermes lifecycle hooks to the local-plane emitter.
This is the *only* instrumentation seam. It registers observational hooks (which core
already invokes fail-open) and translates each into a typed local-plane telemetry
event handed to ``agent.telemetry.emitter``. There are zero edits to core call sites:
the hooks already carry model/provider/usage/duration/tool data.
Everything here is best-effort and fail-open — a raised exception in a hook callback is
swallowed by core, and we additionally guard each callback so a telemetry bug can never
disturb a session. No content, no network: local plane only.
Hooks consumed:
on_session_start -> begin a run context (trace_id/run_id), buffer a run row
post_api_request -> one model_call event (tokens, latency, raw provider/model)
api_request_error -> one error event
post_tool_call -> one tool_call event (raw tool name, duration, result class)
on_session_finalize -> finalize the run row (end_reason, counts, cost)
subagent_start/stop -> (reserved) lineage markers
"""
from __future__ import annotations
import logging
import threading
import time
from typing import Any, Dict, Optional
logger = logging.getLogger(__name__)
# Per-run accumulators keyed by run_id, so on_session_finalize can roll up counts.
_runs: Dict[str, Dict[str, Any]] = {}
_runs_lock = threading.Lock()
def _safe(fn):
"""Decorator: never let a telemetry hook raise into core."""
def wrapper(*args, **kwargs):
try:
return fn(*args, **kwargs)
except Exception:
logger.debug("telemetry hook %s failed", getattr(fn, "__name__", "?"), exc_info=True)
return None
wrapper.__name__ = getattr(fn, "__name__", "wrapper")
return wrapper
def _run_key(session_id: Optional[str], task_id: Optional[str]) -> str:
return session_id or task_id or "default"
# ── on_session_start ────────────────────────────────────────────────────────
@_safe
def _on_session_start(**kw: Any) -> None:
from agent.telemetry import spans
session_id = kw.get("session_id") or ""
platform = kw.get("platform") or kw.get("source") or ""
ctx = spans.start_run()
key = _run_key(session_id, kw.get("task_id"))
with _runs_lock:
_runs[key] = {
"run_id": ctx.run_id,
"trace_id": ctx.trace_id,
"session_id": session_id or None,
"entrypoint": _entrypoint_for(platform, kw.get("source")),
"platform": platform or None,
"start_ns": time.time_ns(),
"model_call_count": 0,
"tool_call_count": 0,
"error_count": 0,
}
def _ensure_run(session_id: Optional[str], task_id: Optional[str], platform: str = "") -> Dict[str, Any]:
"""Return the run accumulator, lazily creating one if session_start was missed."""
from agent.telemetry import spans
key = _run_key(session_id, task_id)
with _runs_lock:
run = _runs.get(key)
if run is None:
rid = spans.current_run_id() or spans.new_id()
tid = spans.current_trace_id() or spans.new_id()
run = {
"run_id": rid,
"trace_id": tid,
"session_id": session_id or None,
"entrypoint": _entrypoint_for(platform),
"platform": platform or None,
"start_ns": time.time_ns(),
"model_call_count": 0,
"tool_call_count": 0,
"error_count": 0,
}
_runs[key] = run
return run
def _entrypoint_for(platform: Optional[str], source: Optional[str] = None) -> str:
"""Coarse entrypoint label (cli / gateway / tui / api / cron …).
This is a workflow *surface* label, not model/tool anonymization — it answers
"where did this run come from", which is genuinely categorical.
"""
s = (source or platform or "").lower()
if s in ("", "chat", "interactive", "cli", "desktop"):
return "cli"
if s in ("telegram", "discord", "slack", "whatsapp", "signal", "matrix",
"email", "sms", "teams", "feishu", "wecom", "line", "google_chat"):
return "gateway"
if s == "tui":
return "tui"
if s in ("api", "api_server", "openai_api"):
return "api"
if s == "cron":
return "cron"
if s == "batch":
return "batch"
if s == "acp":
return "acp"
return "cli"
# ── post_api_request -> model_call event ────────────────────────────────────
@_safe
def _on_post_api_request(**kw: Any) -> None:
from agent.telemetry import emitter, spans
from agent.telemetry.events import ModelCallEvent
session_id = kw.get("session_id") or ""
platform = kw.get("platform") or ""
run = _ensure_run(session_id, kw.get("task_id"), platform)
usage = kw.get("usage") or {}
duration = kw.get("api_duration")
latency_ms = int(duration * 1000) if isinstance(duration, (int, float)) else None
evt = ModelCallEvent(
span_id=spans.new_span_id(),
run_id=run["run_id"],
provider=kw.get("provider"), # raw
model=kw.get("model"), # raw
base_url=kw.get("base_url"),
input_tokens=int(usage.get("input_tokens") or 0),
output_tokens=int(usage.get("output_tokens") or 0),
cache_read_tokens=int(usage.get("cache_read_tokens") or 0),
cache_write_tokens=int(usage.get("cache_write_tokens") or 0),
reasoning_tokens=int(usage.get("reasoning_tokens") or 0),
latency_ms=latency_ms,
end_reason="completed",
)
with _runs_lock:
run["model_call_count"] += 1
emitter.emit(evt)
# ── api_request_error -> error event ────────────────────────────────────────
@_safe
def _on_api_request_error(**kw: Any) -> None:
from agent.telemetry import emitter
from agent.telemetry.events import ErrorEvent
session_id = kw.get("session_id") or ""
run = _ensure_run(session_id, kw.get("task_id"), kw.get("platform") or "")
error_class = _coarse_error_class(kw.get("error_type") or kw.get("error") or "")
with _runs_lock:
run["error_count"] += 1
emitter.emit(ErrorEvent(
run_id=run["run_id"],
error_class=error_class,
subsystem="model_api",
recovery=None,
))
def _coarse_error_class(raw: Any) -> str:
s = str(raw).lower()
if "timeout" in s:
return "provider_timeout"
if "rate" in s and "limit" in s:
return "rate_limit"
if any(k in s for k in ("auth", "401", "403", "unauthorized", "forbidden")):
return "auth"
if any(k in s for k in ("connection", "network", "dns", "socket")):
return "network"
if "context" in s and ("length" in s or "overflow" in s or "token" in s):
return "context_overflow"
if any(k in s for k in ("500", "502", "503", "server error", "provider")):
return "provider_error"
return "unknown"
# ── post_tool_call -> tool_call event ───────────────────────────────────────
@_safe
def _on_post_tool_call(**kw: Any) -> None:
from agent.telemetry import emitter, spans
from agent.telemetry.events import ToolCallEvent
session_id = kw.get("session_id") or ""
run = _ensure_run(session_id, kw.get("task_id"), kw.get("platform") or "")
function_name = kw.get("function_name") or kw.get("tool_name")
duration_ms = kw.get("duration_ms")
result = kw.get("result")
result_class = _tool_result_class(result)
with _runs_lock:
run["tool_call_count"] += 1
if result_class == "error":
run["error_count"] += 1
emitter.emit(ToolCallEvent(
span_id=spans.new_span_id(),
run_id=run["run_id"],
tool_name=function_name, # raw tool name
duration_ms=int(duration_ms) if isinstance(duration_ms, (int, float)) else None,
result_class=result_class,
))
def _tool_result_class(result: Any) -> str:
"""Classify a tool result without retaining content — error vs ok vs blocked."""
try:
import json
if isinstance(result, str):
r = result.strip()
if r.startswith("{"):
obj = json.loads(r)
if isinstance(obj, dict):
if obj.get("error") or obj.get("blocked"):
return "blocked" if obj.get("blocked") else "error"
if obj.get("timeout"):
return "timeout"
return "ok"
if isinstance(result, dict):
if result.get("error"):
return "error"
return "ok"
except Exception:
return "ok"
return "ok"
# ── on_session_finalize -> finalize the run row ─────────────────────────────
@_safe
def _on_session_finalize(**kw: Any) -> None:
from agent.telemetry import emitter, spans
from agent.telemetry.events import RunEvent
session_id = kw.get("session_id") or ""
key = _run_key(session_id, kw.get("task_id"))
with _runs_lock:
run = _runs.pop(key, None)
if run is None:
run = _ensure_run(session_id, kw.get("task_id"), kw.get("platform") or "")
with _runs_lock:
_runs.pop(key, None)
end_reason = _coarse_end_reason(kw)
emitter.emit(RunEvent(
run_id=run["run_id"],
trace_id=run["trace_id"],
entrypoint=run.get("entrypoint", "cli"),
session_id=run.get("session_id"),
platform=run.get("platform"),
start_ns=run.get("start_ns", time.time_ns()),
end_ns=time.time_ns(),
end_reason=end_reason,
model_call_count=run.get("model_call_count", 0),
tool_call_count=run.get("tool_call_count", 0),
error_count=run.get("error_count", 0),
estimated_cost_usd=_as_float(kw.get("estimated_cost_usd")),
cost_status=kw.get("cost_status"),
))
spans.clear_run()
def _coarse_end_reason(kw: Dict[str, Any]) -> str:
if kw.get("interrupted"):
return "interrupted"
if kw.get("failed"):
return "failed"
reason = str(kw.get("turn_exit_reason") or "").lower()
if "max_iteration" in reason:
return "max_iterations"
if "timeout" in reason:
return "timeout"
return "completed"
def _as_float(v: Any) -> Optional[float]:
try:
return float(v) if v is not None else None
except (TypeError, ValueError):
return None
# ── subagent lineage (reserved) ─────────────────────────────────────────────
@_safe
def _on_subagent_start(**kw: Any) -> None:
# Subagents inherit the run context via contextvars; no explicit handling needed.
return None
@_safe
def _on_subagent_stop(**kw: Any) -> None:
return None
# ── registration ────────────────────────────────────────────────────────────
def register(ctx) -> None:
ctx.register_hook("on_session_start", _on_session_start)
ctx.register_hook("post_api_request", _on_post_api_request)
ctx.register_hook("api_request_error", _on_api_request_error)
ctx.register_hook("post_tool_call", _on_post_tool_call)
ctx.register_hook("on_session_finalize", _on_session_finalize)
ctx.register_hook("subagent_start", _on_subagent_start)
ctx.register_hook("subagent_stop", _on_subagent_stop)
logger.debug("telemetry plugin registered 7 hooks")

12
plugins/telemetry/plugin.yaml Normal file
View file

@ -0,0 +1,12 @@
name: telemetry
version: 1.0.0
description: "Local-first telemetry & observability. Records runs, model calls, tool calls, and errors to a local event log + state.db index via plugin hooks — no agent action, no content, no network. Powers /usage, /insights, and local dashboards."
author: NousResearch
hooks:
- post_api_request
- api_request_error
- post_tool_call
- on_session_start
- on_session_finalize
- subagent_start
- subagent_stop