From 74ab798b498fd5f3831f89b5e113a9b128bb2355 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 26 Jun 2026 04:36:26 -0500 Subject: [PATCH] feat(trace): derive OTel-style execution traces from the session store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reconstruct per-session and per-turn span trees (AGENT/LLM/TOOL, with subagents nested under their delegate_task span) straight from SQLite — no new write path. Turn-scoping splits on real user prompts and folds synthetic continuations ([ASYNC DELEGATION ...], [IMPORTANT: ...]) into the turn that spawned them. Exports to OTLP/JSON and Chrome Trace formats. --- agent/trace_builder.py | 610 ++++++++++++++++++++++++++++++ agent/trace_export.py | 170 +++++++++ hermes_state.py | 21 + tests/agent/test_trace_builder.py | 331 ++++++++++++++++ 4 files changed, 1132 insertions(+) create mode 100644 agent/trace_builder.py create mode 100644 agent/trace_export.py create mode 100644 tests/agent/test_trace_builder.py diff --git a/agent/trace_builder.py b/agent/trace_builder.py new file mode 100644 index 00000000000..4540e93b694 --- /dev/null +++ b/agent/trace_builder.py @@ -0,0 +1,610 @@ +"""Derive OpenTelemetry-style traces from the Hermes session store. + +Hermes already persists everything a trace needs: ``sessions`` rows carry +server-side ``started_at`` / ``ended_at`` and full token accounting, and +``messages`` rows carry a server-side ``timestamp`` plus ``tool_calls`` (the +OpenAI tool-call JSON) and ``tool_call_id`` so a tool call can be paired with +its result. Every subagent is itself a session linked by +``parent_session_id``. That means a complete, accurately-timed span tree can be +reconstructed for *any* session — historical or live — with zero extra +instrumentation. + +This module is the read-side "derive-on-read" trace builder. It turns a session +(and its subagent descendants) into a provider-neutral :class:`Trace` of +:class:`Span` objects. ``agent/trace_export.py`` renders that into OTLP/JSON +(OpenInference conventions, ingestible by Arize Phoenix / any OTel backend) or +the Chrome Trace Event format (viewable in https://ui.perfetto.dev). + +Accuracy note: the Hermes agent loop runs tool calls sequentially, so inferring +span durations from consecutive message timestamps matches real execution. The +only inferred link is a ``delegate_task`` tool call → its child session, matched +by start-time proximity; a future precision pass can persist the spawning +``tool_call_id`` to make that exact. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Protocol + +logger = logging.getLogger(__name__) + +# OpenInference span kinds (what Phoenix and other OTel-GenAI viewers expect). +KIND_AGENT = "AGENT" +KIND_LLM = "LLM" +KIND_TOOL = "TOOL" +KIND_CHAIN = "CHAIN" + +# Status codes, mirroring OTLP (1 = OK, 2 = ERROR). +STATUS_OK = "ok" +STATUS_ERROR = "error" +STATUS_UNSET = "unset" + +# Tool names that spawn subagent sessions. A span with one of these names gets +# its matched child session's subtree nested underneath it. +_DELEGATE_TOOL_NAMES = frozenset({"delegate_task"}) + +# How long after the last message a session's ``ended_at`` may sit and still be +# trusted as real activity (vs a cleanup/orphan reaper firing much later). +_END_GRACE_SECONDS = 300.0 + + +class _SessionStore(Protocol): + """The slice of ``SessionDB`` the builder depends on (keeps it testable).""" + + def get_session(self, session_id: str) -> Optional[Dict[str, Any]]: ... + + def get_messages( + self, session_id: str, include_inactive: bool = False + ) -> List[Dict[str, Any]]: ... + + def get_child_session_ids(self, parent_session_id: str) -> List[str]: ... + + +@dataclass +class Span: + """A single unit of work on the trace timeline. + + Times are epoch seconds (float) to match ``messages.timestamp``. Exporters + convert to their own units (OTLP nanoseconds, Chrome microseconds). + """ + + span_id: str + parent_id: Optional[str] + name: str + kind: str + start: float + end: float + status: str = STATUS_UNSET + session_id: Optional[str] = None + attributes: Dict[str, Any] = field(default_factory=dict) + + @property + def duration(self) -> float: + return max(0.0, self.end - self.start) + + def to_dict(self) -> Dict[str, Any]: + return { + "span_id": self.span_id, + "parent_id": self.parent_id, + "name": self.name, + "kind": self.kind, + "start": self.start, + "end": self.end, + "duration": self.duration, + "status": self.status, + "session_id": self.session_id, + "attributes": {k: v for k, v in self.attributes.items() if v is not None}, + } + + +@dataclass +class Trace: + """A full span tree rooted at one session (plus its subagent descendants).""" + + trace_id: str + root_session_id: str + spans: List[Span] = field(default_factory=list) + root_span_id: Optional[str] = None + metadata: Dict[str, Any] = field(default_factory=dict) + + @property + def start(self) -> float: + return min((s.start for s in self.spans), default=0.0) + + @property + def end(self) -> float: + return max((s.end for s in self.spans), default=0.0) + + @property + def duration(self) -> float: + return max(0.0, self.end - self.start) + + def to_dict(self) -> Dict[str, Any]: + return { + "trace_id": self.trace_id, + "root_session_id": self.root_session_id, + "root_span_id": self.root_span_id, + "start": self.start, + "end": self.end, + "duration": self.duration, + "metadata": {k: v for k, v in self.metadata.items() if v is not None}, + "spans": [s.to_dict() for s in self.spans], + } + + +# ── tool-call shape helpers ────────────────────────────────────────────────── + + +def _tool_call_id(call: Dict[str, Any]) -> str: + return str(call.get("id") or call.get("tool_call_id") or "") + + +def _tool_call_name(call: Dict[str, Any]) -> str: + fn = call.get("function") + if isinstance(fn, dict) and fn.get("name"): + return str(fn["name"]) + return str(call.get("name") or "tool") + + +def _tool_call_args(call: Dict[str, Any]) -> Any: + fn = call.get("function") + raw = fn.get("arguments") if isinstance(fn, dict) else call.get("arguments") + if isinstance(raw, str): + try: + return json.loads(raw) + except (json.JSONDecodeError, TypeError): + return raw + return raw + + +def _as_text(content: Any) -> str: + if content is None: + return "" + if isinstance(content, str): + return content + try: + return json.dumps(content, ensure_ascii=False, default=str) + except (TypeError, ValueError): + return str(content) + + +def _looks_like_error(message: Dict[str, Any]) -> bool: + """Best-effort error detection on a tool-result message.""" + text = _as_text(message.get("content")).lstrip() + if not text: + return False + head = text[:400].lower() + if text.startswith("{"): + try: + obj = json.loads(text) + if isinstance(obj, dict): + if obj.get("error") or obj.get("success") is False: + return True + status = str(obj.get("status", "")).lower() + if status in {"error", "failed", "failure"}: + return True + except (json.JSONDecodeError, TypeError): + pass + return any( + marker in head + for marker in ("traceback (most recent call last)", "error:", "exception:") + ) + + +# ── builder ────────────────────────────────────────────────────────────────── + + +def _short(value: str, limit: int = 120) -> str: + flat = " ".join(value.split()) + return flat if len(flat) <= limit else flat[: limit - 1] + "…" + + +def _llm_span_name() -> str: + """Label for an LLM (assistant-turn) span. A plain structural "llm" (matching + the OTel/Langfuse convention of short, uniform LLM labels) — the model and + the response text live in the span's attributes / detail panel, not the row. + """ + return "llm" + + +def _mk_span_id(prefix: str, session_id: str, key: Any) -> str: + return f"{prefix}:{session_id}:{key}" + + +def _trace_id(session_id: str) -> str: + return f"trace:{session_id}" + + +def build_trace( + store: _SessionStore, + session_id: str, + *, + include_subagents: bool = True, + _depth: int = 0, + _max_depth: int = 8, +) -> Optional[Trace]: + """Reconstruct a :class:`Trace` for ``session_id`` from the session store. + + Returns ``None`` when the session does not exist. Walks delegate subagent + descendants (``parent_session_id``) and nests each under the + ``delegate_task`` tool span that spawned it. + """ + session = store.get_session(session_id) + if not session: + return None + + trace = Trace( + trace_id=_trace_id(session_id), + root_session_id=session_id, + metadata={ + "source": session.get("source"), + "model": session.get("model"), + "cwd": session.get("cwd"), + "git_branch": session.get("git_branch"), + }, + ) + + root_span = _build_session_spans(store, session, parent_span_id=None, trace=trace) + trace.root_span_id = root_span.span_id if root_span else None + + if include_subagents and root_span and _depth < _max_depth: + _attach_subagents(store, session_id, trace, _depth=_depth, _max_depth=_max_depth) + + trace.spans.sort(key=lambda s: (s.start, s.span_id)) + return trace + + +# Synthetic re-injections that re-enter the conversation as ``user`` messages +# but are CONTINUATIONS of earlier work, not a fresh prompt: async-delegation +# completions (`[ASYNC DELEGATION …]`) and background-process notifications +# (`[IMPORTANT: …]`). They must not open a new turn — otherwise a background +# subagent dispatched in turn N shows up as its own orphan "[ASYNC DELEGATION]" +# turn when it finishes, instead of folding into the group that spawned it. This +# mirrors the desktop live view, which only resets the live turn on a real +# ``prompt.submit``. +_CONTINUATION_PREFIXES = ( + "[ASYNC DELEGATION", + "[IMPORTANT:", +) + + +def _is_continuation(message: Dict[str, Any]) -> bool: + """True for a synthetic re-injection that should merge into the current turn.""" + if message.get("role") != "user": + return False + return _as_text(message.get("content")).lstrip().startswith(_CONTINUATION_PREFIXES) + + +def _split_turns(messages: List[Dict[str, Any]]) -> List[tuple]: + """Split messages into turns. A turn begins at each *real* ``user`` message and + runs until the next one. Synthetic continuations (async-delegation / + background-process re-injections) do NOT start a turn — they merge into the + one that spawned the work. Leading non-user messages (e.g. system) join the + first turn. Returns ``[(start_idx, end_idx), ...]`` index ranges. + """ + bounds: List[tuple] = [] + start = 0 + for i, m in enumerate(messages): + if i > 0 and m.get("role") == "user" and not _is_continuation(m): + bounds.append((start, i)) + start = i + bounds.append((start, len(messages))) + return bounds + + +def build_session_turns( + store: _SessionStore, + session_id: str, + *, + include_subagents: bool = True, +) -> List[Trace]: + """Build one :class:`Trace` per turn for a session. + + A turn (one user prompt → the agent's full response, subagents included) is + the natural trace unit — it has no inter-turn idle gaps, so each renders as a + tight waterfall. Returns traces in chronological order. + """ + session = store.get_session(session_id) + if not session: + return [] + messages = store.get_messages(session_id) + if not messages: + return [] + + meta = { + "source": session.get("source"), + "model": session.get("model"), + "cwd": session.get("cwd"), + "git_branch": session.get("git_branch"), + } + out: List[Trace] = [] + for ti, (a, b) in enumerate(_split_turns(messages)): + slice_msgs = messages[a:b] + if not slice_msgs: + continue + trace = Trace( + trace_id=f"{_trace_id(session_id)}:t{ti}", + root_session_id=session_id, + metadata={**meta, "turn": ti}, + ) + root = _build_session_spans( + store, + session, + parent_span_id=None, + trace=trace, + messages=slice_msgs, + agent_key=f"turn{ti}", + ) + if not root: + continue + trace.root_span_id = root.span_id + if include_subagents: + _attach_subagents( + store, + session_id, + trace, + agent_key=f"turn{ti}", + window=(root.start, root.end), + _depth=0, + _max_depth=8, + ) + trace.spans.sort(key=lambda s: (s.start, s.span_id)) + out.append(trace) + + return out + + +def _build_session_spans( + store: _SessionStore, + session: Dict[str, Any], + *, + parent_span_id: Optional[str], + trace: Trace, + messages: Optional[List[Dict[str, Any]]] = None, + agent_key: str = "root", +) -> Optional[Span]: + """Append the AGENT span for one session (or a turn slice of it) plus its + LLM/TOOL child spans. + + ``messages`` lets a caller pass a turn-scoped slice; ``agent_key`` keeps the + AGENT span id unique per turn. Returns the session's root (AGENT) span, or + ``None`` for an empty session. + """ + session_id = session["id"] + if messages is None: + messages = store.get_messages(session_id) + if not messages: + return None + + msg_start = min(float(m["timestamp"]) for m in messages) + msg_end = max(float(m["timestamp"]) for m in messages) + started_at = float(session.get("started_at") or msg_start) + + # Clamp the AGENT span to real message activity. Session ``started_at`` / + # ``ended_at`` are unreliable for trace timing: ``ended_at`` can be a + # cleanup/orphan reaper firing hours later (e.g. ``ws_orphan_reap``), and on + # a turn slice ``started_at`` is the whole-session start, far before this + # turn. Snap to them only when they sit right at the slice's edges, so a + # span never balloons into inter-turn idle. + activity_start = msg_start + if msg_start - _END_GRACE_SECONDS <= started_at <= msg_start: + activity_start = started_at + activity_end = msg_end + raw_ended = session.get("ended_at") + if raw_ended is not None: + ended_at = float(raw_ended) + if msg_end <= ended_at <= msg_end + _END_GRACE_SECONDS: + activity_end = ended_at + + goal = _session_goal(messages, session) + agent_span = Span( + span_id=_mk_span_id("agent", session_id, agent_key), + parent_id=parent_span_id, + name=goal, + kind=KIND_AGENT, + start=activity_start, + end=activity_end, + status=STATUS_ERROR if session.get("end_reason") in {"error", "failed"} else STATUS_OK, + session_id=session_id, + attributes={ + "session.id": session_id, + "session.source": session.get("source"), + "llm.model_name": session.get("model"), + "llm.token_count.prompt": session.get("input_tokens"), + "llm.token_count.completion": session.get("output_tokens"), + "llm.token_count.reasoning": session.get("reasoning_tokens"), + "session.message_count": session.get("message_count"), + "session.tool_call_count": session.get("tool_call_count"), + "session.end_reason": session.get("end_reason"), + }, + ) + trace.spans.append(agent_span) + + # Pre-index tool results by tool_call_id so calls pair with their output. + results_by_id: Dict[str, Dict[str, Any]] = {} + for m in messages: + if m.get("role") == "tool" and m.get("tool_call_id"): + results_by_id[str(m["tool_call_id"])] = m + + # Walk the turn timeline. An assistant message closes the LLM span that + # began at the previous boundary; each tool_call it carries becomes a TOOL + # span ending at its paired result. + prev_boundary = activity_start + for m in messages: + role = m.get("role") + ts = float(m["timestamp"]) + + if role == "assistant": + llm_span = Span( + span_id=_mk_span_id("llm", session_id, m["id"]), + parent_id=agent_span.span_id, + name=_llm_span_name(), + kind=KIND_LLM, + start=prev_boundary, + end=ts, + status=STATUS_OK, + session_id=session_id, + attributes={ + "llm.model_name": session.get("model"), + "llm.token_count.completion": m.get("token_count"), + "output.value": _short(_as_text(m.get("content")), 2000), + "hermes.finish_reason": m.get("finish_reason"), + "hermes.has_reasoning": bool( + m.get("reasoning") or m.get("reasoning_content") + ), + }, + ) + trace.spans.append(llm_span) + + for call in m.get("tool_calls") or []: + if not isinstance(call, dict): + continue + _append_tool_span( + trace=trace, + session=session, + parent_span_id=agent_span.span_id, + call=call, + call_ts=ts, + results_by_id=results_by_id, + fallback_end=activity_end, + ) + + prev_boundary = ts + elif role in {"user", "tool"}: + # User input and tool results define the next LLM span's start. + prev_boundary = ts + + return agent_span + + +def _append_tool_span( + *, + trace: Trace, + session: Dict[str, Any], + parent_span_id: str, + call: Dict[str, Any], + call_ts: float, + results_by_id: Dict[str, Dict[str, Any]], + fallback_end: float, +) -> None: + session_id = session["id"] + call_id = _tool_call_id(call) + name = _tool_call_name(call) + result = results_by_id.get(call_id) if call_id else None + end = float(result["timestamp"]) if result else fallback_end + status = STATUS_OK + if result and _looks_like_error(result): + status = STATUS_ERROR + elif not result: + status = STATUS_UNSET + + args = _tool_call_args(call) + span = Span( + span_id=_mk_span_id("tool", session_id, call_id or f"{call_ts}:{name}"), + parent_id=parent_span_id, + name=name, + kind=KIND_TOOL, + start=call_ts, + end=max(end, call_ts), + status=status, + session_id=session_id, + attributes={ + "tool.name": name, + "tool.call_id": call_id or None, + "input.value": _short(_as_text(args), 2000), + "output.value": _short(_as_text(result.get("content")), 2000) if result else None, + "hermes.is_delegate": name in _DELEGATE_TOOL_NAMES, + }, + ) + trace.spans.append(span) + + +def _attach_subagents( + store: _SessionStore, + session_id: str, + trace: Trace, + *, + agent_key: str = "root", + window: Optional[tuple] = None, + _depth: int, + _max_depth: int, +) -> None: + """Nest each delegate child session under the tool span that spawned it. + + Children are matched to ``delegate_task`` tool spans by start-time proximity + (each child consumed once). Unmatched children attach to the session's AGENT + span so they are never dropped from the trace. ``window`` (start, end) limits + attachment to children spawned during a turn slice. + """ + child_ids = store.get_child_session_ids(session_id) + if not child_ids: + return + + delegate_spans = sorted( + ( + s + for s in trace.spans + if s.session_id == session_id + and s.kind == KIND_TOOL + and s.attributes.get("hermes.is_delegate") + ), + key=lambda s: s.start, + ) + agent_span_id = _mk_span_id("agent", session_id, agent_key) + + children = [] + for cid in child_ids: + csess = store.get_session(cid) + if not csess: + continue + if window is not None: + cstart = float(csess.get("started_at") or 0.0) + if not (window[0] - 1.0 <= cstart <= window[1] + 1.0): + continue + children.append(csess) + children.sort(key=lambda c: float(c.get("started_at") or 0.0)) + + used: set = set() + for csess in children: + cstart = float(csess.get("started_at") or 0.0) + parent_span_id = agent_span_id + best = None + best_gap = None + for ds in delegate_spans: + if ds.span_id in used: + continue + gap = abs(ds.start - cstart) + if best_gap is None or gap < best_gap: + best, best_gap = ds, gap + if best is not None: + used.add(best.span_id) + parent_span_id = best.span_id + + child_root = _build_session_spans( + store, csess, parent_span_id=parent_span_id, trace=trace + ) + if child_root and _depth + 1 < _max_depth: + _attach_subagents( + store, csess["id"], trace, _depth=_depth + 1, _max_depth=_max_depth + ) + + +def _session_goal(messages: List[Dict[str, Any]], session: Dict[str, Any]) -> str: + """A human-readable label for an AGENT span. + + Prefer the first user message in the given slice so per-turn spans get their + own prompt as a label (the session title is identical across every turn). + Fall back to the session title, then a short id. + """ + for m in messages: + if m.get("role") == "user": + text = _as_text(m.get("content")).strip() + if text: + return _short(text, 120) + title = session.get("title") + if title: + return _short(str(title), 120) + return f"session {str(session.get('id', ''))[:8]}" diff --git a/agent/trace_export.py b/agent/trace_export.py new file mode 100644 index 00000000000..6e07459520d --- /dev/null +++ b/agent/trace_export.py @@ -0,0 +1,170 @@ +"""Render a :class:`~agent.trace_builder.Trace` into portable file formats. + +Two formats, both hand-built (no OTel SDK dependency): + +* **OTLP/JSON** with OpenInference semantic conventions — the industry standard + for LLM/agent traces. Ingestible by Arize Phoenix and any OpenTelemetry + backend, so we can confirm our spans are correct in a real viewer. +* **Chrome Trace Event format** — each session becomes its own track, viewable + by dropping the file into https://ui.perfetto.dev or ``chrome://tracing``. + +Keeping these as plain dict/JSON builders means the trace layer has zero new +third-party dependencies. +""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any, Dict, List + +from agent.trace_builder import STATUS_ERROR, STATUS_OK, Trace + +_SERVICE_NAME = "hermes-agent" +_SCOPE_NAME = "hermes.tracing" + + +def _hex_id(value: str, nbytes: int) -> str: + """Deterministic hex id of ``nbytes`` bytes from an arbitrary string.""" + digest = hashlib.sha1(value.encode("utf-8")).hexdigest() + return digest[: nbytes * 2] + + +def _otlp_any_value(value: Any) -> Dict[str, Any]: + if isinstance(value, bool): + return {"boolValue": value} + if isinstance(value, int): + return {"intValue": str(value)} + if isinstance(value, float): + return {"doubleValue": value} + return {"stringValue": str(value)} + + +def _otlp_attributes(attrs: Dict[str, Any]) -> List[Dict[str, Any]]: + out: List[Dict[str, Any]] = [] + for key, value in attrs.items(): + if value is None: + continue + out.append({"key": key, "value": _otlp_any_value(value)}) + return out + + +def _otlp_status(status: str) -> Dict[str, Any]: + if status == STATUS_OK: + return {"code": 1} + if status == STATUS_ERROR: + return {"code": 2} + return {"code": 0} + + +def _to_nanos(seconds: float) -> str: + return str(int(seconds * 1_000_000_000)) + + +def to_otlp_json(trace: Trace) -> Dict[str, Any]: + """Build an OTLP/JSON ``TracesData`` document with OpenInference attributes.""" + trace_hex = _hex_id(trace.trace_id, 16) + otlp_spans: List[Dict[str, Any]] = [] + + for span in trace.spans: + attributes = dict(span.attributes) + # OpenInference: the GenAI span kind travels as an attribute; the OTLP + # SpanKind stays INTERNAL (1). + attributes["openinference.span.kind"] = span.kind + if span.session_id: + attributes.setdefault("session.id", span.session_id) + + otlp_span: Dict[str, Any] = { + "traceId": trace_hex, + "spanId": _hex_id(span.span_id, 8), + "name": span.name, + "kind": 1, + "startTimeUnixNano": _to_nanos(span.start), + "endTimeUnixNano": _to_nanos(span.end), + "attributes": _otlp_attributes(attributes), + "status": _otlp_status(span.status), + } + if span.parent_id: + otlp_span["parentSpanId"] = _hex_id(span.parent_id, 8) + otlp_spans.append(otlp_span) + + return { + "resourceSpans": [ + { + "resource": { + "attributes": _otlp_attributes( + { + "service.name": _SERVICE_NAME, + "session.id": trace.root_session_id, + "hermes.source": trace.metadata.get("source"), + } + ) + }, + "scopeSpans": [ + { + "scope": {"name": _SCOPE_NAME}, + "spans": otlp_spans, + } + ], + } + ] + } + + +def to_chrome_trace(trace: Trace) -> Dict[str, Any]: + """Build a Chrome Trace Event document (one track per session).""" + base = trace.start + events: List[Dict[str, Any]] = [] + + # Stable, compact track ids per session — each session is its own lane. + tids: Dict[str, int] = {} + + def tid_for(session_id: str) -> int: + if session_id not in tids: + tids[session_id] = len(tids) + 1 + return tids[session_id] + + for span in trace.spans: + sid = span.session_id or trace.root_session_id + events.append( + { + "name": span.name, + "cat": span.kind, + "ph": "X", + "ts": (span.start - base) * 1_000_000, + "dur": max(0.0, span.duration) * 1_000_000, + "pid": 1, + "tid": tid_for(sid), + "args": {k: v for k, v in span.attributes.items() if v is not None}, + } + ) + + # Name each track after its session (root first) for legible lanes. + for session_id, tid in tids.items(): + label = "root" if session_id == trace.root_session_id else f"subagent {session_id[:8]}" + events.append( + { + "name": "thread_name", + "ph": "M", + "pid": 1, + "tid": tid, + "args": {"name": label}, + } + ) + + return {"traceEvents": events, "displayTimeUnit": "ms"} + + +def dumps(trace: Trace, fmt: str = "otlp", *, indent: int = 2) -> str: + """Serialize ``trace`` to a JSON string in the requested format.""" + fmt = (fmt or "otlp").lower() + if fmt in {"otlp", "otlp-json", "openinference"}: + doc = to_otlp_json(trace) + elif fmt in {"chrome", "perfetto", "trace-event"}: + doc = to_chrome_trace(trace) + else: + raise ValueError(f"unknown trace format: {fmt!r} (use 'otlp' or 'chrome')") + return json.dumps(doc, ensure_ascii=False, indent=indent, default=str) + + +__all__ = ["dumps", "to_chrome_trace", "to_otlp_json"] diff --git a/hermes_state.py b/hermes_state.py index a7938f7167f..e444a0790e6 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -1931,6 +1931,27 @@ class SessionDB: row = cursor.fetchone() return dict(row) if row else None + def get_child_session_ids(self, parent_session_id: str) -> List[str]: + """Return subagent child session ids spawned by ``parent_session_id``. + + Scoped to *ephemeral* children — delegate/subagent runs — using the same + predicate that hides them from session pickers. Branch forks and + compression continuations (which also carry ``parent_session_id``) are + intentionally excluded: a branch is a separate trace, and a compression + continuation is the same conversation rather than a nested subagent. + Ordered by ``started_at`` so callers can match them to spawn order. + """ + if not parent_session_id: + return [] + with self._lock: + cursor = self._conn.execute( + f"SELECT id FROM sessions s " + f"WHERE s.parent_session_id = ? AND {_ephemeral_child_sql('s')} " + f"ORDER BY s.started_at ASC, s.id ASC", + (parent_session_id,), + ) + return [row["id"] for row in cursor.fetchall()] + def resolve_session_id(self, session_id_or_prefix: str) -> Optional[str]: """Resolve an exact or uniquely prefixed session ID to the full ID. diff --git a/tests/agent/test_trace_builder.py b/tests/agent/test_trace_builder.py new file mode 100644 index 00000000000..70a717beb46 --- /dev/null +++ b/tests/agent/test_trace_builder.py @@ -0,0 +1,331 @@ +"""Tests for the derive-on-read trace builder and exporters. + +Builds real sessions/messages in a temp SQLite store and asserts the +reconstructed span tree, then checks the OTLP/JSON and Chrome export shapes. +""" + +from __future__ import annotations + +import json + +import pytest + +from agent.trace_builder import ( + KIND_AGENT, + KIND_LLM, + KIND_TOOL, + STATUS_ERROR, + STATUS_OK, + build_session_turns, + build_trace, +) +from agent.trace_export import to_chrome_trace, to_otlp_json +from hermes_state import SessionDB + +BASE = 1_700_000_000.0 + + +def _tool_call(call_id: str, name: str, args: dict): + return { + "id": call_id, + "type": "function", + "function": {"name": name, "arguments": json.dumps(args)}, + } + + +@pytest.fixture +def db(tmp_path): + store = SessionDB(db_path=tmp_path / "sessions.db") + yield store + store.close() + + +def _set_times(db: SessionDB, session_id: str, started: float, ended: float): + db._conn.execute( + "UPDATE sessions SET started_at = ?, ended_at = ? WHERE id = ?", + (started, ended, session_id), + ) + db._conn.commit() + + +def _build_parent_with_subagent(db: SessionDB): + """Parent session that reads a file, then delegates to a subagent.""" + db.create_session("parent", "cli", model="test-model") + db.append_message("parent", "user", "do the thing", timestamp=BASE) + db.append_message( + "parent", + "assistant", + "", + tool_calls=[_tool_call("call_read", "read_file", {"path": "a.py"})], + token_count=10, + timestamp=BASE + 1, + ) + db.append_message( + "parent", + "tool", + '{"success": true}', + tool_name="read_file", + tool_call_id="call_read", + timestamp=BASE + 2, + ) + db.append_message( + "parent", + "assistant", + "", + tool_calls=[_tool_call("call_deleg", "delegate_task", {"goal": "sub"})], + timestamp=BASE + 3, + ) + # Subagent child session. + db.create_session("child", "tool", parent_session_id="parent", model="test-model") + db.append_message("child", "user", "sub goal", timestamp=BASE + 3.5) + db.append_message( + "child", + "assistant", + "working", + tool_calls=[_tool_call("call_search", "search_files", {"q": "x"})], + timestamp=BASE + 4, + ) + db.append_message( + "child", + "tool", + "results", + tool_name="search_files", + tool_call_id="call_search", + timestamp=BASE + 4.5, + ) + db.append_message("child", "assistant", "subagent done", timestamp=BASE + 5) + _set_times(db, "child", BASE + 3.4, BASE + 5) + # Delegate result lands back in the parent, then the parent wraps up. + db.append_message( + "parent", + "tool", + "subagent done", + tool_name="delegate_task", + tool_call_id="call_deleg", + timestamp=BASE + 6, + ) + db.append_message("parent", "assistant", "all done", timestamp=BASE + 7) + _set_times(db, "parent", BASE, BASE + 7) + + +def test_build_trace_basic_shape(db): + _build_parent_with_subagent(db) + trace = build_trace(db, "parent") + + assert trace is not None + kinds = [s.kind for s in trace.spans] + assert kinds.count(KIND_AGENT) == 2 # parent + child + assert kinds.count(KIND_TOOL) == 3 # read_file, delegate_task, search_files + assert kinds.count(KIND_LLM) == 5 # 3 parent assistants + 2 child assistants + + root = next(s for s in trace.spans if s.span_id == trace.root_span_id) + assert root.kind == KIND_AGENT + assert root.parent_id is None + assert root.session_id == "parent" + + +def test_tool_span_pairs_and_times(db): + _build_parent_with_subagent(db) + trace = build_trace(db, "parent") + + read = next(s for s in trace.spans if s.attributes.get("tool.name") == "read_file") + assert read.start == pytest.approx(BASE + 1) + assert read.end == pytest.approx(BASE + 2) + assert read.status == STATUS_OK + assert read.attributes["tool.call_id"] == "call_read" + + +def test_subagent_nested_under_delegate_span(db): + _build_parent_with_subagent(db) + trace = build_trace(db, "parent") + + delegate = next( + s for s in trace.spans if s.attributes.get("tool.name") == "delegate_task" + ) + child_root = next( + s for s in trace.spans if s.kind == KIND_AGENT and s.session_id == "child" + ) + assert child_root.parent_id == delegate.span_id + # The delegate tool span should envelop the child's work. + assert delegate.start <= child_root.start + assert delegate.end >= BASE + 6 - 0.001 + + +def test_no_subagents_flag_excludes_children(db): + _build_parent_with_subagent(db) + trace = build_trace(db, "parent", include_subagents=False) + assert all(s.session_id == "parent" for s in trace.spans) + + +def test_error_status_detected(db): + db.create_session("err", "cli", model="m") + db.append_message("err", "user", "go", timestamp=BASE) + db.append_message( + "err", + "assistant", + "", + tool_calls=[_tool_call("c1", "terminal", {"cmd": "boom"})], + timestamp=BASE + 1, + ) + db.append_message( + "err", + "tool", + '{"error": "command failed", "success": false}', + tool_name="terminal", + tool_call_id="c1", + timestamp=BASE + 2, + ) + _set_times(db, "err", BASE, BASE + 2) + + trace = build_trace(db, "err") + tool = next(s for s in trace.spans if s.kind == KIND_TOOL) + assert tool.status == STATUS_ERROR + + +def test_missing_session_returns_none(db): + assert build_trace(db, "nope") is None + + +def test_otlp_export_shape(db): + _build_parent_with_subagent(db) + trace = build_trace(db, "parent") + doc = to_otlp_json(trace) + + spans = doc["resourceSpans"][0]["scopeSpans"][0]["spans"] + assert len(spans) == len(trace.spans) + one = spans[0] + assert len(one["traceId"]) == 32 # 16 bytes hex + assert len(one["spanId"]) == 16 # 8 bytes hex + keys = {a["key"] for a in one["attributes"]} + assert "openinference.span.kind" in keys + # All non-root spans carry a parentSpanId. + assert any("parentSpanId" in s for s in spans) + + +def test_ended_at_orphan_reap_does_not_balloon_root(db): + # ended_at sits hours after the last message (cleanup reaper); the AGENT + # span must clamp to real activity, not the bogus ended_at. + db.create_session("orphan", "tui", model="m") + db.append_message("orphan", "user", "go", timestamp=BASE) + db.append_message("orphan", "assistant", "done", timestamp=BASE + 10) + db._conn.execute( + "UPDATE sessions SET started_at=?, ended_at=?, end_reason=? WHERE id=?", + (BASE, BASE + 18_000, "ws_orphan_reap", "orphan"), + ) + db._conn.commit() + + trace = build_trace(db, "orphan") + root = next(s for s in trace.spans if s.kind == KIND_AGENT) + assert root.duration < 60 # ~10s of real work, not 5 hours + + +def test_build_session_turns_splits_and_tightens(db): + # Two turns separated by a long idle gap; each turn-trace must be tight. + db.create_session("multi", "tui", model="m") + db.append_message("multi", "user", "first task", timestamp=BASE) + db.append_message("multi", "assistant", "done one", timestamp=BASE + 5) + # User walks away for 800s, then a second turn. + db.append_message("multi", "user", "second task", timestamp=BASE + 805) + db.append_message("multi", "assistant", "done two", timestamp=BASE + 810) + db._conn.execute( + "UPDATE sessions SET started_at=?, ended_at=? WHERE id=?", + (BASE, BASE + 810, "multi"), + ) + db._conn.commit() + + turns = build_session_turns(db, "multi") + assert len(turns) == 2 + # Neither turn contains the 800s idle gap. + assert all(t.duration < 60 for t in turns) + assert turns[0].metadata["turn"] == 0 + assert turns[1].metadata["turn"] == 1 + + +def test_async_delegation_completion_merges_into_dispatch_turn(db): + # A background delegation dispatched in turn 0 re-enters as a synthetic + # `[ASYNC DELEGATION …]` user message. It must NOT open its own turn — it + # merges into the turn that spawned it, so the completion processing lands in + # the same group as the delegate_task call. + db.create_session("async", "tui", model="m") + db.append_message("async", "user", "kick off background work", timestamp=BASE) + db.append_message( + "async", + "assistant", + "", + tool_calls=[_tool_call("call_bg", "delegate_task", {"goal": "bg", "background": True})], + timestamp=BASE + 1, + ) + db.append_message( + "async", + "tool", + '{"delegation_id": "d1"}', + tool_name="delegate_task", + tool_call_id="call_bg", + timestamp=BASE + 2, + ) + db.append_message("async", "assistant", "dispatched, carrying on", timestamp=BASE + 3) + # Later, the background result re-enters as a synthetic continuation. + db.append_message( + "async", + "user", + "[ASYNC DELEGATION COMPLETE — d1]\nA background subagent finished.", + timestamp=BASE + 50, + ) + db.append_message("async", "assistant", "acting on the result", timestamp=BASE + 52) + db._conn.execute( + "UPDATE sessions SET started_at=?, ended_at=? WHERE id=?", + (BASE, BASE + 52, "async"), + ) + db._conn.commit() + + turns = build_session_turns(db, "async") + assert len(turns) == 1 # not split by the re-injection + # Label is the real prompt, not the async marker. + root = next(s for s in turns[0].spans if s.span_id == turns[0].root_span_id) + assert "kick off background work" in root.name + assert "ASYNC DELEGATION" not in root.name + + +def test_important_notification_merges_into_turn(db): + # Background-process notifications (`[IMPORTANT: …]`) are continuations too. + db.create_session("notif", "tui", model="m") + db.append_message("notif", "user", "run the build", timestamp=BASE) + db.append_message("notif", "assistant", "started", timestamp=BASE + 1) + db.append_message( + "notif", + "user", + "[IMPORTANT: Background process p1 exited with code 0]", + timestamp=BASE + 30, + ) + db.append_message("notif", "assistant", "build finished", timestamp=BASE + 31) + db._conn.execute( + "UPDATE sessions SET started_at=?, ended_at=? WHERE id=?", + (BASE, BASE + 31, "notif"), + ) + db._conn.commit() + + turns = build_session_turns(db, "notif") + assert len(turns) == 1 + + +def test_to_dict_round_trips_shape(db): + _build_parent_with_subagent(db) + trace = build_trace(db, "parent") + d = trace.to_dict() + assert d["root_session_id"] == "parent" + assert d["root_span_id"] == trace.root_span_id + assert len(d["spans"]) == len(trace.spans) + span = d["spans"][0] + assert {"span_id", "parent_id", "kind", "start", "end", "duration", "status"} <= set(span) + + +def test_chrome_export_one_track_per_session(db): + _build_parent_with_subagent(db) + trace = build_trace(db, "parent") + doc = to_chrome_trace(trace) + + complete = [e for e in doc["traceEvents"] if e["ph"] == "X"] + assert len(complete) == len(trace.spans) + tids = {e["tid"] for e in complete} + assert len(tids) == 2 # parent + child lanes + assert all(e["ts"] >= 0 for e in complete)