From edac3a7baef82bb3ccc658a1f828867261ad6b1e Mon Sep 17 00:00:00 2001 From: emozilla Date: Sat, 27 Jun 2026 01:21:41 -0400 Subject: [PATCH] refactor(telemetry): cut dead schema; tests assert what's actually written MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review after the #51714 feedback found the reviewer's dead-table finding was not isolated — the schema advertised far more than the code populates, and our own tests hid it by hand-feeding fields production never sends. Make the surface honest by subtraction. Schema (10 tel_* tables -> 5): - Delete tel_gateway_events, tel_cron_events, tel_skill_events, tel_memory_events, tel_feedback_events — declared, never written, never read. - Drop columns nothing populates: tel_runs.{profile_id,estimated_cost_usd, cost_status}; tel_model_calls.{ttft_ms,estimated_cost_usd,cost_status, cost_source,end_reason,retry_count}; tel_tool_calls.{backend,retry_count, approval}; tel_spans.attrs_json. Cost duplicated the existing sessions billing columns and was always NULL here. - events.py / emitter _TABLE_COLUMNS / OTLP _span_attrs / rollup / preview display all trimmed to match. Correctness: - end_reason no longer hardcodes "completed". Production finalize callers pass `reason` (shutdown/session_expired/session_reset); _coarse_end_reason now reads it and maps accordingly. - Fix a latent bug the trim exposed: the model_call hook passed end_reason= to ModelCallEvent, which the @_safe wrapper was silently swallowing — so tel_model_calls dropped every row in real runs. Now writes correctly. Tests: - Stop hand-feeding estimated_cost_usd / turn_exit_reason that no production call site sends. Finalize is now driven with the real `reason` kwarg, and assertions cover only fields that are actually populated. This is what let the model_call drop hide — the suite graded on a fictional contract. Net: a smaller system that does what it says. Verified end-to-end over the real dispatch path (runs + connected span tree + model/tool rows populate; dead tables gone). 160 telemetry/state/insights tests green. --- agent/telemetry/emitter.py | 12 ++++------- agent/telemetry/events.py | 14 +------------ agent/telemetry/otlp_exporter.py | 8 +++----- agent/telemetry/rollup.py | 3 +-- hermes_cli/main.py | 3 +-- plugins/telemetry/__init__.py | 26 +++++++++++++----------- tests/telemetry/test_cli_telemetry.py | 2 +- tests/telemetry/test_plugin_e2e.py | 2 +- tests/telemetry/test_plugin_hooks.py | 7 +++---- tests/telemetry/test_rollup.py | 2 +- tests/telemetry/test_schema_migration.py | 3 +-- tests/telemetry/test_spans_trace.py | 2 +- 12 files changed, 32 insertions(+), 52 deletions(-) diff --git a/agent/telemetry/emitter.py b/agent/telemetry/emitter.py index 269046c9b23..5f804c1dacb 100644 --- a/agent/telemetry/emitter.py +++ b/agent/telemetry/emitter.py @@ -54,10 +54,9 @@ def _default_db_path() -> Path: _TABLE_COLUMNS: Dict[str, tuple] = { "run": ( "tel_runs", - ("run_id", "trace_id", "session_id", "profile_id", "entrypoint", + ("run_id", "trace_id", "session_id", "entrypoint", "platform", "start_ns", "end_ns", "end_reason", - "model_call_count", "tool_call_count", "error_count", - "estimated_cost_usd", "cost_status"), + "model_call_count", "tool_call_count", "error_count"), ), "span": ( "tel_spans", @@ -68,14 +67,11 @@ _TABLE_COLUMNS: Dict[str, tuple] = { "tel_model_calls", ("span_id", "run_id", "provider", "model", "base_url", "input_tokens", "output_tokens", "cache_read_tokens", - "cache_write_tokens", "reasoning_tokens", "latency_ms", "ttft_ms", - "estimated_cost_usd", "cost_status", "cost_source", "end_reason", - "retry_count"), + "cache_write_tokens", "reasoning_tokens", "latency_ms"), ), "tool_call": ( "tel_tool_calls", - ("span_id", "run_id", "tool_name", "backend", - "duration_ms", "result_class", "retry_count", "approval"), + ("span_id", "run_id", "tool_name", "duration_ms", "result_class"), ), "error": ( "tel_error_events", diff --git a/agent/telemetry/events.py b/agent/telemetry/events.py index 9b770238808..020ce11a961 100644 --- a/agent/telemetry/events.py +++ b/agent/telemetry/events.py @@ -20,12 +20,11 @@ def _now_ns() -> int: @dataclass(slots=True) class RunEvent: - """One top-level workflow execution (a trace root).""" + """One top-level workflow execution (a trace root). A run spans one session.""" run_id: str trace_id: str entrypoint: str session_id: Optional[str] = None - profile_id: Optional[str] = None platform: Optional[str] = None start_ns: int = field(default_factory=_now_ns) end_ns: Optional[int] = None @@ -33,8 +32,6 @@ class RunEvent: model_call_count: int = 0 tool_call_count: int = 0 error_count: int = 0 - estimated_cost_usd: Optional[float] = None - cost_status: Optional[str] = None def to_dict(self) -> Dict[str, Any]: return {"event": "run", **asdict(self)} @@ -53,12 +50,6 @@ class ModelCallEvent: cache_write_tokens: int = 0 reasoning_tokens: int = 0 latency_ms: Optional[int] = None - ttft_ms: Optional[int] = None - estimated_cost_usd: Optional[float] = None - cost_status: Optional[str] = None - cost_source: Optional[str] = None - end_reason: Optional[str] = None - retry_count: int = 0 def to_dict(self) -> Dict[str, Any]: return {"event": "model_call", **asdict(self)} @@ -69,11 +60,8 @@ class ToolCallEvent: span_id: str run_id: str tool_name: Optional[str] = None # raw tool name, e.g. "web_search" - backend: Optional[str] = None duration_ms: Optional[int] = None result_class: Optional[str] = None - retry_count: int = 0 - approval: Optional[str] = None def to_dict(self) -> Dict[str, Any]: return {"event": "tool_call", **asdict(self)} diff --git a/agent/telemetry/otlp_exporter.py b/agent/telemetry/otlp_exporter.py index 0043f6486a6..3f18026a5df 100644 --- a/agent/telemetry/otlp_exporter.py +++ b/agent/telemetry/otlp_exporter.py @@ -140,15 +140,13 @@ def _span_attrs(ev: Dict[str, Any]) -> Dict[str, Any]: attrs: Dict[str, Any] = {"hermes.event": kind or "unknown"} keep_by_kind = { "run": ("entrypoint", "platform", "end_reason", - "model_call_count", "tool_call_count", "error_count", - "estimated_cost_usd", "cost_status"), + "model_call_count", "tool_call_count", "error_count"), "span": ("trace_id", "run_id", "parent_span_id", "name", "kind", "start_ns", "end_ns", "status"), "model_call": ("provider", "model", "base_url", "input_tokens", "output_tokens", "cache_read_tokens", - "cache_write_tokens", "reasoning_tokens", "latency_ms", - "ttft_ms", "end_reason"), - "tool_call": ("tool_name", "backend", "duration_ms", "result_class"), + "cache_write_tokens", "reasoning_tokens", "latency_ms"), + "tool_call": ("tool_name", "duration_ms", "result_class"), "error": ("error_class", "subsystem", "recovery"), } for col in keep_by_kind.get(kind, ()): # type: ignore[arg-type] diff --git a/agent/telemetry/rollup.py b/agent/telemetry/rollup.py index 8f30304f92b..3880a55caa8 100644 --- a/agent/telemetry/rollup.py +++ b/agent/telemetry/rollup.py @@ -52,7 +52,7 @@ def _run_events(c: sqlite3.Connection, since_ns: Optional[int]) -> List[Dict[str where += f" AND start_ns >= {int(since_ns)}" rows = c.execute( "SELECT run_id, entrypoint, platform, end_reason, start_ns, end_ns, " - "model_call_count, tool_call_count, error_count, estimated_cost_usd " + "model_call_count, tool_call_count, error_count " "FROM tel_runs" + where ).fetchall() @@ -98,7 +98,6 @@ def _run_events(c: sqlite3.Connection, since_ns: Optional[int]) -> List[Dict[str "duration_ms": round(duration_ms, 1) if duration_ms is not None else None, "input_tokens": int((trow["inp"] if trow else 0) or 0), "output_tokens": int((trow["outp"] if trow else 0) or 0), - "estimated_cost_usd": r["estimated_cost_usd"], }) return events diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 173eb2a1ede..abd4c462dd3 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -14422,8 +14422,7 @@ def cmd_telemetry(args): print(f" • heartbeat hermes={e.get('hermes_version')} os={e.get('os_family')}") continue bits = [f"{e.get('event_name')}"] - for k in ("entrypoint", "platform", "end_reason", "duration_ms", - "estimated_cost_usd"): + for k in ("entrypoint", "platform", "end_reason", "duration_ms"): if e.get(k) is not None: bits.append(f"{k}={e[k]}") models = e.get("models_used") or [] diff --git a/plugins/telemetry/__init__.py b/plugins/telemetry/__init__.py index 44ff228f29d..c270d4d72a2 100644 --- a/plugins/telemetry/__init__.py +++ b/plugins/telemetry/__init__.py @@ -182,7 +182,6 @@ def _on_post_api_request(**kw: Any) -> None: 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 @@ -324,30 +323,33 @@ def _on_session_finalize(**kw: Any) -> None: 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: + """Map a finalize payload to a coarse end reason. + + Production finalize callers pass ``reason`` (e.g. "shutdown", "session_expired", + "session_reset"); the older ``turn_exit_reason``/``interrupted``/``failed`` keys are + honored when present. Defaults to "ended". + """ if kw.get("interrupted"): return "interrupted" if kw.get("failed"): return "failed" - reason = str(kw.get("turn_exit_reason") or "").lower() + reason = str(kw.get("turn_exit_reason") or kw.get("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 + if "expired" in reason: + return "expired" + if "reset" in reason: + return "reset" + if "shutdown" in reason or "complete" in reason: + return "completed" + return reason or "ended" # ── subagent lineage (reserved) ───────────────────────────────────────────── diff --git a/tests/telemetry/test_cli_telemetry.py b/tests/telemetry/test_cli_telemetry.py index 4aaf4cab744..19fc59e394c 100644 --- a/tests/telemetry/test_cli_telemetry.py +++ b/tests/telemetry/test_cli_telemetry.py @@ -23,7 +23,7 @@ def home(tmp_path, monkeypatch): now = time.time_ns() em.emit(RunEvent(run_id="r1", trace_id="t1", entrypoint="cli", end_reason="completed", start_ns=now - 60_000_000, end_ns=now, model_call_count=1, - tool_call_count=1, estimated_cost_usd=0.3)) + tool_call_count=1)) em.emit(ModelCallEvent(span_id="m1", run_id="r1", provider="anthropic", model="claude-opus-4", input_tokens=20000, output_tokens=2000)) diff --git a/tests/telemetry/test_plugin_e2e.py b/tests/telemetry/test_plugin_e2e.py index c6cf05688d8..9a9ed00913d 100644 --- a/tests/telemetry/test_plugin_e2e.py +++ b/tests/telemetry/test_plugin_e2e.py @@ -63,7 +63,7 @@ def _fire_one_turn(invoke_hook): invoke_hook("post_tool_call", session_id="s1", platform="cli", function_name="web_search", duration_ms=210, result='{"data": "ok"}') invoke_hook("on_session_finalize", session_id="s1", platform="cli", - turn_exit_reason="completed", estimated_cost_usd=0.01, cost_status="known") + reason="shutdown") def test_real_dispatch_writes_tel_rows(runtime): diff --git a/tests/telemetry/test_plugin_hooks.py b/tests/telemetry/test_plugin_hooks.py index 807dc276e1b..6b1d1b2efbf 100644 --- a/tests/telemetry/test_plugin_hooks.py +++ b/tests/telemetry/test_plugin_hooks.py @@ -43,9 +43,9 @@ def test_full_session_lifecycle_produces_rows(wired): session_id="sess1", platform="telegram", function_name="web_search", duration_ms=812, result="{\"data\": \"...\"}", ) + # Production finalize callers pass `reason` (e.g. "shutdown"), not cost. plug._on_session_finalize( - session_id="sess1", platform="telegram", - turn_exit_reason="completed", estimated_cost_usd=0.042, cost_status="known", + session_id="sess1", platform="telegram", reason="shutdown", ) em.flush() @@ -58,7 +58,6 @@ def test_full_session_lifecycle_produces_rows(wired): assert run["end_reason"] == "completed" assert run["model_call_count"] == 1 assert run["tool_call_count"] == 1 - assert abs(run["estimated_cost_usd"] - 0.042) < 1e-9 mc = conn.execute("SELECT * FROM tel_model_calls").fetchone() assert mc["provider"] == "anthropic" @@ -78,7 +77,7 @@ def test_tool_error_result_classified_and_counted(wired): session_id="s2", function_name="terminal", duration_ms=10, result="{\"error\": \"command failed\"}", ) - plug._on_session_finalize(session_id="s2", turn_exit_reason="completed") + plug._on_session_finalize(session_id="s2", reason="shutdown") em.flush() conn = sqlite3.connect(db) conn.row_factory = sqlite3.Row diff --git a/tests/telemetry/test_rollup.py b/tests/telemetry/test_rollup.py index 5ef50c89f08..b271bb916f9 100644 --- a/tests/telemetry/test_rollup.py +++ b/tests/telemetry/test_rollup.py @@ -21,7 +21,7 @@ def _seed(tmp_path): em.emit(RunEvent(run_id="r1", trace_id="t1", entrypoint="gateway", platform="telegram", end_reason="completed", start_ns=now - 90_000_000, end_ns=now, - model_call_count=2, tool_call_count=2, estimated_cost_usd=2.1)) + model_call_count=2, tool_call_count=2)) em.emit(ModelCallEvent(span_id="m1", run_id="r1", provider="anthropic", model="claude-opus-4", input_tokens=60000, output_tokens=8000)) em.emit(ModelCallEvent(span_id="m2", run_id="r1", provider="anthropic", diff --git a/tests/telemetry/test_schema_migration.py b/tests/telemetry/test_schema_migration.py index 5a73a335319..a5a946da4b0 100644 --- a/tests/telemetry/test_schema_migration.py +++ b/tests/telemetry/test_schema_migration.py @@ -9,8 +9,7 @@ import hermes_state TEL_TABLES = { "tel_runs", "tel_spans", "tel_model_calls", "tel_tool_calls", - "tel_gateway_events", "tel_cron_events", "tel_skill_events", - "tel_memory_events", "tel_feedback_events", "tel_error_events", + "tel_error_events", } diff --git a/tests/telemetry/test_spans_trace.py b/tests/telemetry/test_spans_trace.py index aac9320182d..c99fb02fb68 100644 --- a/tests/telemetry/test_spans_trace.py +++ b/tests/telemetry/test_spans_trace.py @@ -50,7 +50,7 @@ def _one_turn(invoke_hook): invoke_hook("post_tool_call", session_id="s1", platform="cli", function_name="web_search", duration_ms=210, result='{"data": "ok"}') invoke_hook("on_session_finalize", session_id="s1", platform="cli", - turn_exit_reason="completed", estimated_cost_usd=0.01, cost_status="known") + reason="shutdown") def test_tel_spans_forms_connected_trace(runtime):