feat(trace): derive OTel-style execution traces from the session store

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

@@ -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]}"

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"]

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.
 

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)