feat: first-class OpenTelemetry export (one span per boundary)

README.md

@@ -415,6 +415,26 @@ pattern.
 
 </details>
 
+<details>
+<summary><b>OpenTelemetry export (spans in Phoenix or any OTel backend)</b></summary>
+
+<br>
+
+Emit one OpenTelemetry span per boundary crossing, using OpenInference semantic
+conventions, so recorded runs show up in Phoenix or any OTel backend as LLM and tool
+spans (input/output, model, token counts), nested by the run's call graph:
+
+```python
+with chronicle.record("run-1") as session:
+    chronicle.instrument_otel()      # attaches to the active recording session
+    run_agent(...)
+```
+
+Call it inside the `record` block, or pass `session=`. Needs the OTel extra
+(`pip install agent-chronicle[phoenix]`); the base install imports no OpenTelemetry.
+
+</details>
+
 <details>
 <summary><b>Lower-level recorder (<code>EnvelopeRecorder</code>)</b></summary>
 

chronicle/__init__.py

@@ -28,6 +28,17 @@
 
 __version__ = "0.3.0"
 
+
+def __getattr__(name: str):
+    # Lazy so the base install never imports opentelemetry. `chronicle.instrument_otel`
+    # (and the attribute mapper) load the optional OTel export on first access.
+    if name in ("instrument_otel", "envelope_span_attributes"):
+        from chronicle import otel
+
+        return getattr(otel, name)
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
 __all__ = [
     "ActionResult",
     "BoundaryMode",
@@ -50,8 +61,10 @@
     "apply_redactors",
     "boundary",
     "default_redactors",
+    "envelope_span_attributes",
     "get_session",
     "instrument_langgraph",
+    "instrument_otel",
     "open_store",
     "record",
     "redact_secrets",

chronicle/otel.py

@@ -0,0 +1,134 @@
+"""First-class OpenTelemetry export: one span per boundary crossing.
+
+Chronicle records each boundary crossing as an Envelope. ``instrument_otel`` turns each
+recorded Envelope into an OpenTelemetry span using OpenInference semantic conventions, so
+recorded runs land in Phoenix (or any OTel backend) with the right shape: LLM and tool
+spans carrying input/output, model, and token counts, nested by the run's call graph.
+
+    with chronicle.record("run-1") as session:
+        chronicle.instrument_otel()      # emit spans for this run (uses the active session)
+        run_agent(...)
+
+Call it inside the ``record`` block (so it attaches to the recording session), or pass
+``session=`` explicitly. Requires the OpenTelemetry SDK and OpenInference conventions:
+``pip install agent-chronicle[phoenix]``. Nothing here is imported by ``import chronicle``,
+so the base install needs neither package.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Callable
+
+from chronicle.envelope.schema import Envelope
+from chronicle.session import ChronicleSession, get_session
+
+
+def _require_trace():
+    try:
+        from opentelemetry import trace
+
+        return trace
+    except ImportError as exc:  # pragma: no cover - exercised only without the extra
+        raise ImportError(
+            "OpenTelemetry export needs the OTel SDK and OpenInference conventions: "
+            "pip install agent-chronicle[phoenix]"
+        ) from exc
+
+
+def _span_kind(boundary_kind: str) -> str:
+    from openinference.semconv.trace import OpenInferenceSpanKindValues as Kind
+
+    return {"llm": Kind.LLM.value, "tool": Kind.TOOL.value}.get(boundary_kind, Kind.CHAIN.value)
+
+
+def _input_value(envelope: Envelope) -> Any:
+    state = envelope.input_state
+    return state.messages or state.graph_state or {}
+
+
+def _output_value(envelope: Envelope) -> Any:
+    action = envelope.action_result
+    if action.error:
+        return {"error": action.error, "error_type": action.error_type}
+    if action.tool_calls:
+        return [tc.model_dump() if hasattr(tc, "model_dump") else tc for tc in action.tool_calls]
+    if action.completion is not None:
+        return action.completion
+    if action.raw_response is not None:
+        return action.raw_response
+    return {}
+
+
+def _as_json(value: Any) -> str:
+    if isinstance(value, str):
+        return value
+    try:
+        return json.dumps(value, default=str)
+    except (TypeError, ValueError):
+        return str(value)
+
+
+def envelope_span_attributes(envelope: Envelope) -> dict[str, Any]:
+    """Map one Envelope to OpenInference span attributes."""
+    from openinference.semconv.trace import SpanAttributes as S
+
+    attributes: dict[str, Any] = {
+        S.OPENINFERENCE_SPAN_KIND: _span_kind(envelope.boundary_kind),
+        S.INPUT_VALUE: _as_json(_input_value(envelope)),
+        S.OUTPUT_VALUE: _as_json(_output_value(envelope)),
+        "chronicle.envelope_id": envelope.envelope_id,
+        "chronicle.trace_id": envelope.trace_id,
+        "chronicle.invocation_index": envelope.invocation_index,
+    }
+    if envelope.metadata.build_id:
+        attributes["chronicle.build_id"] = envelope.metadata.build_id
+    if envelope.boundary_kind == "llm":
+        if envelope.metadata.model_version:
+            attributes[S.LLM_MODEL_NAME] = envelope.metadata.model_version
+        usage = envelope.action_result.token_usage or {}
+        prompt = usage.get("prompt_tokens", usage.get("input_tokens"))
+        completion = usage.get("completion_tokens", usage.get("output_tokens"))
+        if prompt is not None:
+            attributes[S.LLM_TOKEN_COUNT_PROMPT] = int(prompt)
+        if completion is not None:
+            attributes[S.LLM_TOKEN_COUNT_COMPLETION] = int(completion)
+    if envelope.boundary_kind == "tool":
+        attributes[S.TOOL_NAME] = envelope.node_id
+    return attributes
+
+
+def instrument_otel(
+    tracer: Any | None = None,
+    *,
+    session: ChronicleSession | None = None,
+) -> Callable[[], None]:
+    """Emit one OpenTelemetry span per recorded boundary crossing.
+
+    Attaches to ``session`` (default: the active session) via its ``on_record`` hook.
+    Spans nest by the run's parent linkage and carry OpenInference attributes. Returns a
+    callable that removes the instrumentation.
+    """
+    trace = _require_trace()
+    tracer = tracer or trace.get_tracer("chronicle")
+    active = session or get_session()
+    spans: dict[str, Any] = {}  # envelope_id -> span, for parent linkage
+
+    def on_record(envelope: Envelope) -> None:
+        parent = spans.get(envelope.parent_envelope_id) if envelope.parent_envelope_id else None
+        context = trace.set_span_in_context(parent) if parent is not None else None
+        span = tracer.start_span(envelope.node_id, context=context)
+        for key, value in envelope_span_attributes(envelope).items():
+            span.set_attribute(key, value)
+        if envelope.action_result.error:
+            span.set_status(trace.Status(trace.StatusCode.ERROR, envelope.action_result.error))
+        span.end()
+        spans[envelope.envelope_id] = span
+
+    active.on_record = on_record
+
+    def uninstrument() -> None:
+        if active.on_record is on_record:
+            active.on_record = None
+
+    return uninstrument

chronicle/session.py

@@ -63,6 +63,10 @@ class ChronicleSession:
     # whether the function returned or raised. Signature:
     # (boundary_id, kind, input_state) -> None
     on_leave: Callable[[str, str, InputState], None] | None = None
+    # Optional observer fired with the full Envelope right after it is recorded
+    # (LIVE). Used by exporters (e.g. OpenTelemetry) to emit one span per crossing.
+    # Signature: (envelope) -> None
+    on_record: Callable[[Envelope], None] | None = None
     # Applied to each envelope before it is retained or stored, so secrets never
     # reach a committed fixture. Empty by default; set to default_redactors() or
     # your own. Signature: (str) -> str. See chronicle.redaction.
@@ -188,6 +192,8 @@ def record_envelope(
         self._call_log.append(
             CallRecord(boundary_id, invocation_index, "record", envelope.envelope_id)
         )
+        if self.on_record is not None:
+            self.on_record(envelope)
         return envelope
 
     def _fixture_for(self, boundary_id: str) -> Envelope:

tests/test_otel.py

@@ -0,0 +1,131 @@
+"""First-class OpenTelemetry export: one span per boundary crossing, carrying
+OpenInference attributes and nested by the run's call graph. Uses an in-memory exporter.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+
+import pytest
+
+pytest.importorskip("opentelemetry")
+pytest.importorskip("openinference.semconv")
+
+from openinference.semconv.trace import OpenInferenceSpanKindValues as Kind
+from openinference.semconv.trace import SpanAttributes as S
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+from opentelemetry.trace import StatusCode
+
+import chronicle
+from chronicle import boundary
+
+
+def _tracer_and_exporter():
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    return provider.get_tracer("test"), exporter
+
+
+def _run_agent():
+    @boundary("agent", kind="llm")
+    def agent(state):
+        return {**state, "completion": "plan", "tool_calls": [], "finish_reason": "tool_calls"}
+
+    @boundary("refund", kind="tool")
+    def refund(order_id, amount_cents):
+        return {"status": "blocked", "blocked": True, "amount_cents": amount_cents}
+
+    @boundary("agent", kind="llm")
+    def finalize(state, tool_result):
+        return {**state, "completion": "done", "blocked": tool_result["blocked"]}
+
+    state = agent({"messages": []})
+    tool_result = refund("o1", 999)
+    return finalize(state, tool_result)
+
+
+def test_one_span_per_crossing_with_openinference_attrs():
+    tracer, exporter = _tracer_and_exporter()
+    with chronicle.record("t-otel"):
+        chronicle.instrument_otel(tracer=tracer)
+        _run_agent()
+
+    spans = exporter.get_finished_spans()
+    assert [s.name for s in spans] == ["agent", "refund", "agent"]
+    kinds = [s.attributes[S.OPENINFERENCE_SPAN_KIND] for s in spans]
+    assert kinds == [Kind.LLM.value, Kind.TOOL.value, Kind.LLM.value]
+    for span in spans:
+        assert S.INPUT_VALUE in span.attributes
+        assert S.OUTPUT_VALUE in span.attributes
+        assert span.attributes["chronicle.trace_id"] == "t-otel"
+    assert spans[1].attributes[S.TOOL_NAME] == "refund"
+
+
+def test_spans_nest_by_parent_linkage():
+    tracer, exporter = _tracer_and_exporter()
+    with chronicle.record("t-nest"):
+        chronicle.instrument_otel(tracer=tracer)
+        _run_agent()
+
+    agent1, refund, agent2 = exporter.get_finished_spans()
+    assert agent1.parent is None
+    assert refund.parent.span_id == agent1.context.span_id
+    assert agent2.parent.span_id == refund.context.span_id
+
+
+def test_error_boundary_sets_error_status():
+    tracer, exporter = _tracer_and_exporter()
+
+    @boundary("boom", kind="tool")
+    def boom():
+        raise ValueError("nope")
+
+    with chronicle.record("t-err"):
+        chronicle.instrument_otel(tracer=tracer)
+        with pytest.raises(ValueError):
+            boom()
+
+    spans = exporter.get_finished_spans()
+    assert len(spans) == 1
+    assert spans[0].status.status_code == StatusCode.ERROR
+
+
+def test_uninstrument_stops_spans():
+    tracer, exporter = _tracer_and_exporter()
+    with chronicle.record("t-un") as session:
+        stop = chronicle.instrument_otel(tracer=tracer, session=session)
+        _run_agent()
+        stop()
+        _run_agent()
+    assert len(exporter.get_finished_spans()) == 3  # only the first run emitted spans
+
+
+def test_attribute_mapping_includes_model_and_tokens():
+    from chronicle.envelope.schema import ActionResult, ContextMetadata, Envelope, InputState
+
+    env = Envelope(
+        node_id="llm",
+        boundary_kind="llm",
+        trace_id="t",
+        metadata=ContextMetadata(model_version="gpt-4o", build_id="b"),
+        input_state=InputState(messages=[{"role": "user", "content": "hi"}]),
+        action_result=ActionResult(
+            completion="hey", token_usage={"prompt_tokens": 3, "completion_tokens": 2}
+        ),
+    )
+    attrs = chronicle.envelope_span_attributes(env)
+    assert attrs[S.OPENINFERENCE_SPAN_KIND] == Kind.LLM.value
+    assert attrs[S.LLM_MODEL_NAME] == "gpt-4o"
+    assert attrs[S.LLM_TOKEN_COUNT_PROMPT] == 3
+    assert attrs[S.LLM_TOKEN_COUNT_COMPLETION] == 2
+
+
+def test_import_chronicle_does_not_import_opentelemetry():
+    code = "import chronicle, sys; assert 'chronicle.otel' not in sys.modules; print('ok')"
+    proc = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True)
+    assert proc.returncode == 0, proc.stderr
+    assert proc.stdout.strip() == "ok"