diff --git a/README.md b/README.md
index 07a1ebf..674ff72 100644
--- a/README.md
+++ b/README.md
@@ -415,6 +415,26 @@ pattern.
+
+OpenTelemetry export (spans in Phoenix or any OTel backend)
+
+
+
+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.
+
+
+
Lower-level recorder (EnvelopeRecorder)
diff --git a/chronicle/__init__.py b/chronicle/__init__.py
index f44b2fd..38899c0 100644
--- a/chronicle/__init__.py
+++ b/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",
diff --git a/chronicle/otel.py b/chronicle/otel.py
new file mode 100644
index 0000000..59049f9
--- /dev/null
+++ b/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
diff --git a/chronicle/session.py b/chronicle/session.py
index 3210a6b..831c025 100644
--- a/chronicle/session.py
+++ b/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:
diff --git a/tests/test_otel.py b/tests/test_otel.py
new file mode 100644
index 0000000..132566b
--- /dev/null
+++ b/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"