From 7c80b79bb01bbd3f14cf82f3bcfdd9453dd27e35 Mon Sep 17 00:00:00 2001 From: emozilla Date: Sat, 27 Jun 2026 00:52:40 -0400 Subject: [PATCH] =?UTF-8?q?feat(telemetry):=20write=20tel=5Fspans=20?= =?UTF-8?q?=E2=80=94=20reconstructable=20run=20->=20calls=20trace?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the review on #51714: the trace/span layer was declared but unwired — tel_spans was never written, call rows had no timestamp, and nothing set parent lineage, so the store was metrics-only and couldn't reconstruct a trace. Wire the span layer (keeping the praised star-schema shape): - New SpanEvent (span_id/trace_id/run_id/parent_span_id/name/kind/start_ns/end_ns) mapped into tel_spans via the emitter's _TABLE_COLUMNS. - The plugin mints a root span per run and, on each model/tool call, emits a SpanEvent (timing + parent = the run's root) keyed by the SAME span_id as the detail row, so tel_model_calls / tel_tool_calls JOIN to their span. - Call hooks fire on completion, so end_ns = now and start_ns is reconstructed from the measured latency/duration. The run's root span is emitted at finalize with the true run start/end. Result: tel_spans is a connected, single-trace_id, run -> calls tree a desktop waterfall (or any reader) can render directly, ordered by start_ns. Existing metrics rows (tel_runs/model_calls/tool_calls) are unchanged. OTLP: spans now flow to the exporter with their trace/parent/timing attributes. The exporter still emits one OTel span per event rather than reconstructing OTel SpanContexts into a connected trace tree; that projection is left for a follow-up and the module docstring now says so plainly instead of over-claiming. Adds test_spans_trace.py (connected-tree + detail-row JOIN) over the real dispatch path. Accurate (pre-hook) start times, real OTLP SpanContexts, and subagent cross-run lineage remain follow-ups. --- agent/telemetry/emitter.py | 5 ++ agent/telemetry/events.py | 24 ++++++ agent/telemetry/otlp_exporter.py | 19 +++-- plugins/telemetry/__init__.py | 77 +++++++++++++++++--- tests/telemetry/test_spans_trace.py | 109 ++++++++++++++++++++++++++++ 5 files changed, 218 insertions(+), 16 deletions(-) create mode 100644 tests/telemetry/test_spans_trace.py diff --git a/agent/telemetry/emitter.py b/agent/telemetry/emitter.py index 6b628ea5626..269046c9b23 100644 --- a/agent/telemetry/emitter.py +++ b/agent/telemetry/emitter.py @@ -59,6 +59,11 @@ _TABLE_COLUMNS: Dict[str, tuple] = { "model_call_count", "tool_call_count", "error_count", "estimated_cost_usd", "cost_status"), ), + "span": ( + "tel_spans", + ("span_id", "trace_id", "run_id", "parent_span_id", "name", "kind", + "start_ns", "end_ns", "status"), + ), "model_call": ( "tel_model_calls", ("span_id", "run_id", "provider", "model", "base_url", diff --git a/agent/telemetry/events.py b/agent/telemetry/events.py index 4125aaff414..9b770238808 100644 --- a/agent/telemetry/events.py +++ b/agent/telemetry/events.py @@ -79,6 +79,29 @@ class ToolCallEvent: return {"event": "tool_call", **asdict(self)} +@dataclass(slots=True) +class SpanEvent: + """A timed span — the timing/lineage backbone of a trace. + + One row per run (the root, ``parent_span_id=None``) and one per model/tool call + (``parent_span_id`` = the run's root span). Detail rows in ``tel_model_calls`` / + ``tel_tool_calls`` share the ``span_id`` and are joined here for ordering and + placement on a timeline. + """ + span_id: str + trace_id: str + run_id: str + name: str + kind: str # "run" | "model" | "tool" + start_ns: int + end_ns: Optional[int] = None + parent_span_id: Optional[str] = None + status: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + return {"event": "span", **asdict(self)} + + @dataclass(slots=True) class ErrorEvent: run_id: Optional[str] @@ -95,5 +118,6 @@ __all__ = [ "RunEvent", "ModelCallEvent", "ToolCallEvent", + "SpanEvent", "ErrorEvent", ] diff --git a/agent/telemetry/otlp_exporter.py b/agent/telemetry/otlp_exporter.py index 32912b9da27..0043f6486a6 100644 --- a/agent/telemetry/otlp_exporter.py +++ b/agent/telemetry/otlp_exporter.py @@ -1,8 +1,8 @@ """Export telemetry to an OpenTelemetry Collector over OTLP/HTTP. -Maps telemetry events (which carry trace_id/run_id/span_id/parent_span_id) to OTel -spans and sends them to the endpoint configured under ``telemetry.export.otlp``. Lets -an operator stream Hermes telemetry into their own observability stack. +Maps the local tel_* events to OTel spans and sends them to the endpoint configured +under ``telemetry.export.otlp``. Lets an operator stream Hermes telemetry into their +own observability stack. Notes: * The destination is operator-configured; this module only sends to that endpoint. @@ -15,7 +15,13 @@ Notes: * The continuous subscriber runs in the emitter's writer thread after durable writes and is fail-isolated, so an export error cannot affect a run. -Spans carry structural telemetry by default. Message content is included only when the +Each event is exported as a span carrying its recorded attributes (provider, model, +tokens, duration, etc.). The timing/parent linkage captured in tel_spans +(trace_id/span_id/parent_span_id/start_ns/end_ns) is not yet reconstructed into OTel +SpanContexts here, so spans currently arrive as independent records rather than a +connected trace tree; building the connected-trace projection is tracked separately. + +Spans carry structural telemetry by default. Message content is included only when trajectories is enabled, and always passes through the export redaction pipeline. """ @@ -136,6 +142,8 @@ def _span_attrs(ev: Dict[str, Any]) -> Dict[str, Any]: "run": ("entrypoint", "platform", "end_reason", "model_call_count", "tool_call_count", "error_count", "estimated_cost_usd", "cost_status"), + "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", @@ -195,7 +203,8 @@ def _read_events(db_path: Optional[Path], since_ns: Optional[int]) -> List[Dict[ out: List[Dict[str, Any]] = [] try: table_event = { - "tel_runs": "run", "tel_model_calls": "model_call", + "tel_runs": "run", "tel_spans": "span", + "tel_model_calls": "model_call", "tel_tool_calls": "tool_call", "tel_error_events": "error", } for table, evkind in table_event.items(): diff --git a/plugins/telemetry/__init__.py b/plugins/telemetry/__init__.py index 0f5157295c9..44ff228f29d 100644 --- a/plugins/telemetry/__init__.py +++ b/plugins/telemetry/__init__.py @@ -10,12 +10,15 @@ swallowed by core, and we additionally guard each callback so a telemetry bug ca disturb a session. No content, no network: local telemetry 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) + on_session_start -> begin a run context (trace_id/run_id + root span id) + post_api_request -> one model_call event + its timing span (tokens, latency) 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) + post_tool_call -> one tool_call event + its timing span (duration, result class) + on_session_finalize -> finalize the run row + emit the run's root span subagent_start/stop -> (reserved) lineage markers + +Each model/tool call emits a SpanEvent (timing + parent = the run's root span) keyed by +the same span_id as its detail row, so tel_spans reconstructs a run -> calls trace tree. """ from __future__ import annotations @@ -56,15 +59,18 @@ def _on_session_start(**kw: Any) -> None: session_id = kw.get("session_id") or "" platform = kw.get("platform") or kw.get("source") or "" ctx = spans.start_run() + now = time.time_ns() + root_span_id = spans.new_span_id() key = _run_key(session_id, kw.get("task_id")) with _runs_lock: _runs[key] = { "run_id": ctx.run_id, "trace_id": ctx.trace_id, + "root_span_id": root_span_id, "session_id": session_id or None, "entrypoint": _entrypoint_for(platform, kw.get("source")), "platform": platform or None, - "start_ns": time.time_ns(), + "start_ns": now, "model_call_count": 0, "tool_call_count": 0, "error_count": 0, @@ -84,6 +90,7 @@ def _ensure_run(session_id: Optional[str], task_id: Optional[str], platform: str run = { "run_id": rid, "trace_id": tid, + "root_span_id": spans.new_span_id(), "session_id": session_id or None, "entrypoint": _entrypoint_for(platform), "platform": platform or None, @@ -96,6 +103,33 @@ def _ensure_run(session_id: Optional[str], task_id: Optional[str], platform: str return run +def _emit_call_span(run: Dict[str, Any], span_id: str, name: str, kind: str, + duration_ms: Optional[int], status: Optional[str]) -> None: + """Emit the timing/lineage span for a model or tool call. + + The call hooks fire on completion, so end_ns is ~now and start_ns is reconstructed + from the measured duration (end - duration). The span is parented to the run's root + so a 2-level run -> calls waterfall can be reconstructed from tel_spans. + """ + from agent.telemetry import emitter + from agent.telemetry.events import SpanEvent + + end_ns = time.time_ns() + dur_ns = int(duration_ms) * 1_000_000 if isinstance(duration_ms, (int, float)) else 0 + start_ns = end_ns - dur_ns + emitter.emit(SpanEvent( + span_id=span_id, + trace_id=run["trace_id"], + run_id=run["run_id"], + parent_span_id=run.get("root_span_id"), + name=name, + kind=kind, + start_ns=start_ns, + end_ns=end_ns, + status=status, + )) + + def _entrypoint_for(platform: Optional[str], source: Optional[str] = None) -> str: """Coarse entrypoint label (cli / gateway / tui / api / cron …). @@ -135,8 +169,9 @@ def _on_post_api_request(**kw: Any) -> None: duration = kw.get("api_duration") latency_ms = int(duration * 1000) if isinstance(duration, (int, float)) else None + span_id = spans.new_span_id() evt = ModelCallEvent( - span_id=spans.new_span_id(), + span_id=span_id, run_id=run["run_id"], provider=kw.get("provider"), # raw model=kw.get("model"), # raw @@ -151,6 +186,8 @@ def _on_post_api_request(**kw: Any) -> None: ) with _runs_lock: run["model_call_count"] += 1 + _emit_call_span(run, span_id, name=kw.get("model") or "model_call", + kind="model", duration_ms=latency_ms, status="ok") emitter.emit(evt) @@ -209,11 +246,15 @@ def _on_post_tool_call(**kw: Any) -> None: if result_class == "error": run["error_count"] += 1 + dur_int = int(duration_ms) if isinstance(duration_ms, (int, float)) else None + span_id = spans.new_span_id() + _emit_call_span(run, span_id, name=function_name or "tool_call", + kind="tool", duration_ms=dur_int, status=result_class) emitter.emit(ToolCallEvent( - span_id=spans.new_span_id(), + span_id=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, + duration_ms=dur_int, result_class=result_class, )) @@ -245,7 +286,7 @@ def _tool_result_class(result: Any) -> str: @_safe def _on_session_finalize(**kw: Any) -> None: from agent.telemetry import emitter, spans - from agent.telemetry.events import RunEvent + from agent.telemetry.events import RunEvent, SpanEvent session_id = kw.get("session_id") or "" key = _run_key(session_id, kw.get("task_id")) @@ -256,15 +297,29 @@ def _on_session_finalize(**kw: Any) -> None: with _runs_lock: _runs.pop(key, None) + end_ns = time.time_ns() + start_ns = run.get("start_ns", end_ns) end_reason = _coarse_end_reason(kw) + # Root span for the run — the trace root the call spans hang off of. + emitter.emit(SpanEvent( + span_id=run.get("root_span_id") or spans.new_span_id(), + trace_id=run["trace_id"], + run_id=run["run_id"], + parent_span_id=None, + name=f"run:{run.get('entrypoint', 'cli')}", + kind="run", + start_ns=start_ns, + end_ns=end_ns, + status=end_reason, + )) 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(), + start_ns=start_ns, + end_ns=end_ns, end_reason=end_reason, model_call_count=run.get("model_call_count", 0), tool_call_count=run.get("tool_call_count", 0), diff --git a/tests/telemetry/test_spans_trace.py b/tests/telemetry/test_spans_trace.py new file mode 100644 index 00000000000..aac9320182d --- /dev/null +++ b/tests/telemetry/test_spans_trace.py @@ -0,0 +1,109 @@ +"""Trace/span layer: tel_spans is populated as a connected run -> calls tree. + +Drives the real dispatch chain (discover_plugins -> invoke_hook) and asserts the +timing/lineage backbone in tel_spans: + - one root span per run (kind="run", parent_span_id NULL), + - one child span per model/tool call parented to the root, + - a single trace_id across the run, + - call detail rows (tel_model_calls / tel_tool_calls) JOIN to their span by span_id, + - reconstructed durations match the reported latency/duration. + +This is the regression guard for the waterfall a desktop trace viewer renders. +""" + +from __future__ import annotations + +import sqlite3 +import time + +import pytest + +import hermes_state + + +@pytest.fixture +def runtime(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + db = tmp_path / "state.db" + hermes_state.SessionDB(db_path=db) + import hermes_cli.plugins as plugins_mod + monkeypatch.setattr(plugins_mod, "_plugin_manager", None, raising=False) + from agent.telemetry import emitter as emitter_mod + emitter_mod.reset_emitter_for_tests(None) + import plugins.telemetry as plug + plug._runs.clear() + yield db, plugins_mod, emitter_mod + try: + emitter_mod.get_emitter().flush() + except Exception: + pass + emitter_mod.reset_emitter_for_tests(None) + monkeypatch.setattr(plugins_mod, "_plugin_manager", None, raising=False) + + +def _one_turn(invoke_hook): + invoke_hook("on_session_start", session_id="s1", + model="anthropic/claude-opus-4", platform="cli") + invoke_hook("post_api_request", session_id="s1", platform="cli", + provider="anthropic", model="claude-opus-4", api_duration=0.9, + usage={"input_tokens": 1000, "output_tokens": 120}) + 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") + + +def test_tel_spans_forms_connected_trace(runtime): + db, plugins_mod, emitter_mod = runtime + plugins_mod.discover_plugins(force=True) + _one_turn(plugins_mod.invoke_hook) + time.sleep(0.5) + emitter_mod.get_emitter().flush() + + conn = sqlite3.connect(db) + conn.row_factory = sqlite3.Row + spans = conn.execute( + "SELECT span_id, parent_span_id, kind, name, start_ns, end_ns, status, trace_id " + "FROM tel_spans" + ).fetchall() + + # root + model + tool + assert len(spans) == 3 + roots = [s for s in spans if s["parent_span_id"] is None] + children = [s for s in spans if s["parent_span_id"] is not None] + assert len(roots) == 1 + assert roots[0]["kind"] == "run" + assert len(children) == 2 + + # single trace, all children parented to the root + assert len({s["trace_id"] for s in spans}) == 1 + assert all(c["parent_span_id"] == roots[0]["span_id"] for c in children) + + # spans are time-ordered and carry real durations + by_kind = {s["kind"]: s for s in spans} + assert (by_kind["model"]["end_ns"] - by_kind["model"]["start_ns"]) == 900 * 1_000_000 + assert (by_kind["tool"]["end_ns"] - by_kind["tool"]["start_ns"]) == 210 * 1_000_000 + assert by_kind["run"]["end_ns"] >= by_kind["run"]["start_ns"] + + +def test_detail_rows_join_to_spans(runtime): + db, plugins_mod, emitter_mod = runtime + plugins_mod.discover_plugins(force=True) + _one_turn(plugins_mod.invoke_hook) + time.sleep(0.5) + emitter_mod.get_emitter().flush() + + conn = sqlite3.connect(db) + conn.row_factory = sqlite3.Row + mc = conn.execute( + "SELECT m.model, s.kind, s.trace_id FROM tel_model_calls m " + "JOIN tel_spans s ON m.span_id = s.span_id" + ).fetchone() + assert mc is not None and mc["model"] == "claude-opus-4" and mc["kind"] == "model" + + tc = conn.execute( + "SELECT t.tool_name, s.kind FROM tel_tool_calls t " + "JOIN tel_spans s ON t.span_id = s.span_id" + ).fetchone() + assert tc is not None and tc["tool_name"] == "web_search" and tc["kind"] == "tool" + conn.close()