diff --git a/pyproject.toml b/pyproject.toml
index f7d4d297..26c61740 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -48,7 +48,13 @@ Documentation = "https://bub.build"
 bub = "bub.__main__:app"
 
 [project.optional-dependencies]
+trace = [
+    "opentelemetry-api>=1.39.0",
+    "opentelemetry-sdk>=1.39.0",
+    "opentelemetry-exporter-otlp-proto-http>=1.39.0",
+]
 logfire = [
+    "bub[trace]",
     "logfire>=4.31.0",
 ]
 
diff --git a/src/bub/__main__.py b/src/bub/__main__.py
index 97242cc9..5494d737 100644
--- a/src/bub/__main__.py
+++ b/src/bub/__main__.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import sys
 
 import typer
@@ -22,6 +23,15 @@ def _instrument_bub(level: str) -> None:
     logger.remove()
     logger.add(sys.stderr, level=level, colorize=True, diagnose=False)
 
+    if os.getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") or os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT"):
+        from bub.tracing import configure_otlp
+
+        try:
+            configure_otlp()
+        except Exception as exc:
+            logger.warning("OTLP instrumentation disabled: {}", exc)
+        return
+
     try:
         import logfire
         from logfire.integrations.loguru import LogfireHandler
diff --git a/src/bub/builtin/agent.py b/src/bub/builtin/agent.py
index 3924c4d9..5bdb7ad2 100644
--- a/src/bub/builtin/agent.py
+++ b/src/bub/builtin/agent.py
@@ -7,8 +7,8 @@
 import re
 import shlex
 import time
-from collections.abc import AsyncGenerator, AsyncIterator, Callable, Collection, Coroutine, Iterable
-from contextlib import AsyncExitStack
+from collections.abc import AsyncGenerator, AsyncIterator, Collection, Iterable
+from contextlib import AsyncExitStack, aclosing
 from dataclasses import dataclass, replace
 from datetime import UTC, datetime
 from functools import cached_property
@@ -34,6 +34,7 @@
     ToolContext,
     model_tools,
 )
+from bub.tracing import Span, current_span
 from bub.turn import TurnState
 from bub.utils import workspace_from_state
 
@@ -73,19 +74,6 @@ async def generator() -> AsyncIterator:
 
         return AsyncStreamEvents(generator())
 
-    @staticmethod
-    def _events_with_callback(
-        events: AsyncStreamEvents, callback: Callable[[], Coroutine[Any, Any, Any]]
-    ) -> AsyncStreamEvents:
-        async def generator() -> AsyncIterator[StreamEvent]:
-            try:
-                async for event in events:
-                    yield event
-            finally:
-                await callback()
-
-        return AsyncStreamEvents(generator(), state=events._state)
-
     async def run_stream(
         self,
         *,
@@ -96,37 +84,87 @@ async def run_stream(
         allowed_skills: Collection[str] | None = None,
         allowed_tools: Collection[str] | None = None,
     ) -> AsyncStreamEvents:
-        if not prompt:
-            return self._events_from_iterable([
-                StreamEvent("text", {"delta": "error: empty prompt"}),
-                StreamEvent("final", {"text": "error: empty prompt", "ok": False}),
-            ])
-
-        state.setdefault("session_id", session_id)
-        tape = self.tape.session_tape(
-            session_id, workspace_from_state(state), context=replace(self.tape.context, state=state)
+        span = Span(
+            "invoke_agent bub",
+            {
+                "gen_ai.operation.name": "invoke_agent",
+                "gen_ai.agent.name": "bub",
+                "gen_ai.conversation.id": session_id,
+            },
         )
-        merge_back = not session_id.startswith("temp/")
+        span.messages("gen_ai.input.messages", [{"role": "user", "content": prompt}])
         stack = AsyncExitStack()
-        # The fork_tape context manager must not be exited until the last chunk of the stream is consumed.
-        tape = await stack.enter_async_context(tape.fork_tape(merge_back=merge_back))
-        await tape.ensure_bootstrap_anchor()
-        if isinstance(prompt, str) and prompt.strip().startswith(","):
-            result = await self._run_command(tape=tape, line=prompt.strip())
-            events = self._events_from_iterable([
-                StreamEvent("text", {"delta": result}),
-                StreamEvent("final", {"text": result, "ok": True}),
-            ])
-        else:
-            events = await self._agent_loop(
-                tape=tape,
-                prompt=prompt,
-                model=model,
-                allowed_skills=allowed_skills,
-                allowed_tools=allowed_tools,
-            )
+        try:
+            with span.activate():
+                if not prompt:
+                    events = self._events_from_iterable([
+                        StreamEvent("text", {"delta": "error: empty prompt"}),
+                        StreamEvent("final", {"text": "error: empty prompt", "ok": False}),
+                    ])
+                else:
+                    state.setdefault("session_id", session_id)
+                    tape = self.tape.session_tape(
+                        session_id, workspace_from_state(state), context=replace(self.tape.context, state=state)
+                    )
+                    # Keep the tape fork open until the stream closes, even if it is never consumed.
+                    tape = await stack.enter_async_context(
+                        tape.fork_tape(merge_back=not session_id.startswith("temp/"))
+                    )
+                    await tape.ensure_bootstrap_anchor()
+                    if isinstance(prompt, str) and prompt.strip().startswith(","):
+                        result = await self._run_command(tape=tape, line=prompt.strip())
+                        events = self._events_from_iterable([
+                            StreamEvent("text", {"delta": result}),
+                            StreamEvent("final", {"text": result, "ok": True}),
+                        ])
+                    else:
+                        events = await self._agent_loop(
+                            tape=tape,
+                            prompt=prompt,
+                            model=model,
+                            allowed_skills=allowed_skills,
+                            allowed_tools=allowed_tools,
+                        )
+        except BaseException as exc:
+            span.fail(exc)
+            try:
+                with span.activate():
+                    await stack.aclose()
+            finally:
+                span.end()
+            raise
+        return AsyncStreamEvents(
+            self._trace_events(events, span),
+            state=events._state,
+            on_close=stack.aclose,
+            span=span,
+        )
 
-        return self._events_with_callback(events, callback=stack.aclose)
+    @staticmethod
+    async def _trace_events(events: AsyncStreamEvents, span: Span) -> AsyncGenerator[StreamEvent, None]:
+        messages: list[dict[str, Any]] = []
+        text: list[str] = []
+        calls: list[dict[str, Any]] = []
+        try:
+            async with aclosing(events):
+                async for event in events:
+                    if span.recording:
+                        if event.kind == "text":
+                            text.append(str(event.data.get("delta", "")))
+                        elif event.kind == "tool_call":
+                            calls = event.data.get("tool_calls", [])
+                            messages.append({"role": "assistant", "content": "".join(text), "tool_calls": calls})
+                            text.clear()
+                        elif event.kind == "tool_result":
+                            for call, result in zip(calls, event.data.get("tool_results", []), strict=False):
+                                messages.append({"role": "tool", "tool_call_id": call["id"], "content": result})
+                        elif event.kind == "error":
+                            span.fail(RuntimeError(str(event.data.get("message", "agent failed"))))
+                    yield event
+        finally:
+            if text:
+                messages.append({"role": "assistant", "content": "".join(text)})
+            span.messages("gen_ai.output.messages", messages)
 
     async def _run_command(self, tape: Tape, *, line: str) -> str:
         line = line[1:].strip()
@@ -224,22 +262,23 @@ async def _stream_events_with_auto_handoff(
                     allowed_skills=allowed_skills,
                     allowed_tools=allowed_tools,
                 )
-                async for event in output:
-                    yield event
-                    if event.kind == "error":
-                        elapsed_ms = int((time.monotonic() - start) * 1000)
-                        await tape.append_event(
-                            "loop.step",
-                            {
-                                "step": step,
-                                "elapsed_ms": elapsed_ms,
-                                "status": "error",
-                                "error": event.data.get("message", ""),
-                                "date": datetime.now(UTC).isoformat(),
-                            },
-                        )
-                    elif event.kind == "final":
-                        should_continue = bool(event.data.get("tool_calls") or event.data.get("tool_results"))
+                async with aclosing(output):
+                    async for event in output:
+                        yield event
+                        if event.kind == "error":
+                            elapsed_ms = int((time.monotonic() - start) * 1000)
+                            await tape.append_event(
+                                "loop.step",
+                                {
+                                    "step": step,
+                                    "elapsed_ms": elapsed_ms,
+                                    "status": "error",
+                                    "error": event.data.get("message", ""),
+                                    "date": datetime.now(UTC).isoformat(),
+                                },
+                            )
+                        elif event.kind == "final":
+                            should_continue = bool(event.data.get("tool_calls") or event.data.get("tool_results"))
             except Exception as exc:
                 error_message = f"{exc!s}"
                 elapsed_ms = int((time.monotonic() - start) * 1000)
@@ -363,6 +402,12 @@ async def _run_once_stream(
         resolved_model = model or self.settings.model
 
         model_tools_for_call = model_tools(tools)
+        if (span := current_span()) and span.recording:
+            span.set(**{
+                "gen_ai.tool.definitions": [
+                    tool.to_schema()["function"] | {"type": "function"} for tool in model_tools_for_call
+                ]
+            })
         steering_inbox = self.framework.get_steering_inbox()
         steering_envelopes = await steering_inbox.drain_messages(tape.context.state) if steering_inbox else []
         steering_messages = list(
diff --git a/src/bub/builtin/model_runner.py b/src/bub/builtin/model_runner.py
index 561e0450..04cc4c85 100644
--- a/src/bub/builtin/model_runner.py
+++ b/src/bub/builtin/model_runner.py
@@ -3,8 +3,10 @@
 from __future__ import annotations
 
 import asyncio
+import inspect
 import re
 from collections.abc import AsyncGenerator, AsyncIterator, Iterable, Iterator
+from contextlib import aclosing
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from time import monotonic
@@ -40,6 +42,7 @@
 from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState
 from bub.tape import Tape
 from bub.tools import Tool, ToolContext, ToolExecutor
+from bub.tracing import Span, current_span, event
 
 CONTEXT_LENGTH_PATTERNS = re.compile(
     r"context.{0,20}(?:length|window)|maximum.{0,20}context|token.{0,10}limit|prompt.{0,10}too long|tokens? > \d+ maximum",
@@ -139,6 +142,12 @@ async def completion_response(
         completion_error: Exception | None = None
         for index, (candidate, llm) in enumerate(clients):
             try:
+                if span := current_span():
+                    span.rename(f"chat {candidate.model_id}")
+                    span.set(**{
+                        "gen_ai.provider.name": candidate.provider.value,
+                        "gen_ai.request.model": candidate.model_id,
+                    })
                 streaming = llm.SUPPORTS_COMPLETION_STREAMING
                 completion_messages = _adapt_messages_for_provider(messages, candidate.provider)
                 completion_kwargs = {
@@ -154,6 +163,7 @@ async def completion_response(
                     completion_kwargs["reasoning_effort"] = reasoning_effort
                 return cast("CompletionResult", await llm.acompletion(**completion_kwargs))
             except Exception as exc:
+                event("bub.model.attempt_failed", model=candidate.name, error=repr(exc))
                 if completion_error is None:
                     completion_error = exc
                 if index == len(clients) - 1:
@@ -221,15 +231,8 @@ async def fire_after(error: Exception | None = None) -> None:
 
             try:
                 completion_started = monotonic()
-                async with asyncio.timeout(self.settings.model_timeout_seconds):
-                    completion = await self.completion_response(
-                        model=request.model,
-                        messages=list(request.messages),
-                        tools=tools,
-                        max_tokens=request.max_tokens,
-                        reasoning_effort=tape.context.state.get("reasoning_effort"),
-                    )
-                    async for event in self._completion_events(completion, state, output):
+                async with aclosing(self._traced_completion(request, tools, tape, state, output)) as events:
+                    async for event in events:
                         yield event
                 completion_elapsed = monotonic() - completion_started
             except Exception as exc:
@@ -252,6 +255,7 @@ async def fire_after(error: Exception | None = None) -> None:
                 execution = await ToolExecutor(hooks=self.hooks).execute_async(
                     tool_invocations,
                     context=context,
+                    call_ids=[call.id for call in tool_calls],
                 )
                 await self.record_chat(
                     tape=tape,
@@ -286,6 +290,74 @@ async def fire_after(error: Exception | None = None) -> None:
 
         return AsyncStreamEvents(iterator(), state=state)
 
+    def _traced_completion(
+        self,
+        request: LlmCallRequest,
+        tools: list[Tool],
+        tape: Tape,
+        state: StreamState,
+        output: ModelOutputAccumulator,
+    ) -> AsyncStreamEvents:
+        provider, _, model = request.model.partition(":")
+        span = Span(
+            f"chat {model or request.model}",
+            {
+                "gen_ai.operation.name": "chat",
+                "gen_ai.provider.name": provider,
+                "gen_ai.request.model": model or request.model,
+                "gen_ai.request.max_tokens": request.max_tokens,
+                "gen_ai.conversation.id": tape.context.state.get("session_id"),
+                "bub.run_id": request.run_id,
+                "bub.tape": tape.name,
+            },
+        )
+        span.messages("gen_ai.input.messages", request.messages)
+        if span.recording:
+            span.set(**{
+                "gen_ai.tool.definitions": [tool.to_schema()["function"] | {"type": "function"} for tool in tools]
+            })
+
+        async def iterator() -> AsyncGenerator[StreamEvent, None]:
+            async with asyncio.timeout(self.settings.model_timeout_seconds):
+                completion = await self.completion_response(
+                    model=request.model,
+                    messages=list(request.messages),
+                    tools=tools,
+                    max_tokens=request.max_tokens,
+                    reasoning_effort=tape.context.state.get("reasoning_effort"),
+                )
+                try:
+                    async with aclosing(self._completion_events(completion, state, output)) as events:
+                        async for item in events:
+                            yield item
+                finally:
+                    close = getattr(completion, "aclose", None) or getattr(completion, "close", None)
+                    if close is not None:
+                        result = close()
+                        if inspect.isawaitable(result):
+                            await result
+
+        async def finish() -> None:
+            if not span.recording:
+                return
+            usage = state.usage or {}
+            span.set(**{
+                "gen_ai.usage.input_tokens": usage.get("prompt_tokens", usage.get("input_tokens")),
+                "gen_ai.usage.output_tokens": usage.get("completion_tokens", usage.get("output_tokens")),
+            })
+            span.messages(
+                "gen_ai.output.messages",
+                [
+                    {
+                        "role": "assistant",
+                        "content": output.text,
+                        "tool_calls": [call.model_dump(exclude_none=True) for call in output.tool_calls],
+                    }
+                ],
+            )
+
+        return AsyncStreamEvents(iterator(), state=state, span=span, on_close=finish)
+
     @staticmethod
     def generate_run_id() -> str:
         return f"run-{datetime.now(UTC).strftime('%Y%m%dT%H%M%S%fZ')}"
@@ -400,6 +472,7 @@ async def _completion_events(
         output: ModelOutputAccumulator,
     ) -> AsyncGenerator[StreamEvent, None]:
         if isinstance(completion, ChatCompletion):
+            self._trace_response(completion)
             if usage := Tape._extract_usage(completion):
                 state.usage = usage
             output.response = completion
@@ -409,9 +482,18 @@ async def _completion_events(
             return
 
         async for chunk in completion:
+            self._trace_response(chunk)
             async for event in self._completion_chunk_events(chunk, state, output):
                 yield event
 
+    @staticmethod
+    def _trace_response(response: ChatCompletion | ChatCompletionChunk) -> None:
+        if span := current_span():
+            span.set(**{"gen_ai.response.model": response.model, "gen_ai.response.id": response.id})
+            reasons = [choice.finish_reason for choice in response.choices if choice.finish_reason is not None]
+            if reasons:
+                span.set(**{"gen_ai.response.finish_reasons": reasons})
+
     def _completion_message_events(
         self,
         message: ChatCompletionMessage,
diff --git a/src/bub/builtin/tools.py b/src/bub/builtin/tools.py
index 862a88f0..0a59e309 100644
--- a/src/bub/builtin/tools.py
+++ b/src/bub/builtin/tools.py
@@ -4,6 +4,7 @@
 import json
 import uuid
 from collections.abc import Iterable
+from contextlib import aclosing
 from pathlib import Path
 from typing import TYPE_CHECKING, cast
 
@@ -340,18 +341,20 @@ async def run_subagent(param: SubAgentInput, *, context: ToolContext) -> str:
     state = {**context.state, "session_id": subagent_session}
     allowed_tools = resolve_tool_names(param.allowed_tools or None, exclude={"subagent"})
     output = ""
-    async for event in await agent.run_stream(
+    stream = await agent.run_stream(
         session_id=subagent_session,
         prompt=param.prompt,
         state=state,
         model=param.model,
         allowed_tools=allowed_tools,
         allowed_skills=param.allowed_skills,
-    ):
-        if event.kind == "error":
-            output += f"[Error: {event.data.get('message', 'unknown error')}]"
-        elif event.kind == "text":
-            output += str(event.data.get("delta", ""))
+    )
+    async with aclosing(stream):
+        async for event in stream:
+            if event.kind == "error":
+                output += f"[Error: {event.data.get('message', 'unknown error')}]"
+            elif event.kind == "text":
+                output += str(event.data.get("delta", ""))
     return output
 
 
diff --git a/src/bub/errors.py b/src/bub/errors.py
index a1aae96f..141bd03f 100644
--- a/src/bub/errors.py
+++ b/src/bub/errors.py
@@ -19,9 +19,9 @@ class ErrorKind(StrEnum):
     UNKNOWN = "unknown"
 
 
-@dataclass(frozen=True)
+@dataclass
 class BubError(Exception):
-    """Public error type for Bub execution failures."""
+    """Public error type; exception machinery must be able to assign traceback fields."""
 
     kind: ErrorKind
     message: str
diff --git a/src/bub/framework.py b/src/bub/framework.py
index 72aca545..d46f27de 100644
--- a/src/bub/framework.py
+++ b/src/bub/framework.py
@@ -220,16 +220,19 @@ async def _run_model(
         else:
             parts: list[str] = []
             events = self._channel_router.wrap_stream(inbound, stream) if self._channel_router is not None else stream
-            async for event in events:
-                if event.kind == "text":
-                    parts.append(str(event.data.get("delta", "")))
-                elif event.kind == "error":
-                    # Turn "kind" to enum type otherwise BubError's __str__ won't work well.
-                    data = {
-                        **event.data,
-                        "kind": ErrorKind(event.data.get("kind", "unknown")),
-                    }
-                    await self._hook_runtime.notify_error(stage="run_model", error=BubError(**data), message=inbound)
+            async with contextlib.aclosing(stream):
+                async for event in events:
+                    if event.kind == "text":
+                        parts.append(str(event.data.get("delta", "")))
+                    elif event.kind == "error":
+                        # Turn "kind" to enum type otherwise BubError's __str__ won't work well.
+                        data = {
+                            **event.data,
+                            "kind": ErrorKind(event.data.get("kind", "unknown")),
+                        }
+                        await self._hook_runtime.notify_error(
+                            stage="run_model", error=BubError(**data), message=inbound
+                        )
             return "".join(parts)
 
     def hook_report(self) -> dict[str, list[str]]:
diff --git a/src/bub/hooks/runtime.py b/src/bub/hooks/runtime.py
index f972143c..2b0bf112 100644
--- a/src/bub/hooks/runtime.py
+++ b/src/bub/hooks/runtime.py
@@ -4,6 +4,7 @@
 
 import inspect
 from collections.abc import AsyncGenerator
+from contextlib import aclosing
 from typing import Any
 
 import pluggy
@@ -173,9 +174,10 @@ async def run_model(self, prompt: str | list[dict], session_id: str, state: Turn
             if hasattr(plugin, "run_model_stream"):
                 stream = await self.call_first("run_model_stream", prompt=prompt, session_id=session_id, state=state)
                 text = ""
-                async for event in stream:
-                    if event.kind == "text":
-                        text += str(event.data.get("delta", ""))
+                async with aclosing(stream):
+                    async for event in stream:
+                        if event.kind == "text":
+                            text += str(event.data.get("delta", ""))
                 return text
         return None
 
diff --git a/src/bub/streaming.py b/src/bub/streaming.py
index f2c57d42..73392b6e 100644
--- a/src/bub/streaming.py
+++ b/src/bub/streaming.py
@@ -2,11 +2,13 @@
 
 from __future__ import annotations
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Awaitable, Callable
+from contextlib import nullcontext
 from dataclasses import dataclass
 from typing import Any, Literal
 
 from bub.errors import BubError
+from bub.tracing import Span
 
 
 @dataclass
@@ -22,12 +24,64 @@ class StreamEvent:
 
 
 class AsyncStreamEvents:
-    def __init__(self, iterator: AsyncIterator[StreamEvent], *, state: StreamState | None = None) -> None:
+    def __init__(
+        self,
+        iterator: AsyncIterator[StreamEvent],
+        *,
+        state: StreamState | None = None,
+        on_close: Callable[[], Awaitable[None]] | None = None,
+        span: Span | None = None,
+    ) -> None:
         self._iterator = iterator
         self._state = state or StreamState()
+        self._on_close = on_close
+        self._span = span
+        self._closed = False
 
     def __aiter__(self) -> AsyncIterator[StreamEvent]:
-        return self._iterator
+        return self
+
+    async def __anext__(self) -> StreamEvent:
+        if self._closed:
+            raise StopAsyncIteration
+        try:
+            # Detach before returning an event to the consumer, even across tasks.
+            with self._span.activate() if self._span else nullcontext():
+                return await anext(self._iterator)
+        except StopAsyncIteration:
+            await self._close()
+            raise
+        except BaseException as exc:
+            if self._span:
+                self._span.fail(exc)
+            await self._close()
+            raise
+
+    async def aclose(self) -> None:
+        """Close the source and release resources, including before first iteration."""
+        if not self._closed and self._span:
+            self._span.set(**{"bub.cancelled": True})
+        await self._close()
+
+    async def _close(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        try:
+            with self._span.activate() if self._span else nullcontext():
+                try:
+                    if close := getattr(self._iterator, "aclose", None):
+                        await close()
+                finally:
+                    if self._on_close is not None:
+                        await self._on_close()
+        except BaseException as exc:
+            if self._span:
+                self._span.fail(exc)
+            raise
+        finally:
+            if self._span:
+                self._span.end()
 
     @property
     def error(self) -> BubError | None:
diff --git a/src/bub/tape.py b/src/bub/tape.py
index fb2ac5e3..627a074a 100644
--- a/src/bub/tape.py
+++ b/src/bub/tape.py
@@ -14,6 +14,7 @@
 
 from pydantic import BaseModel
 
+from bub import tracing
 from bub.errors import BubError
 from bub.sidecars import TapeSidecar, sidecar_tape_name
 
@@ -298,7 +299,8 @@ async def search(self, query: TapeQuery[AsyncTapeStore]) -> list[TapeEntry]:
         return list(await self.store.fetch_all(query))
 
     async def append_event(self, name: str, payload: dict[str, Any], **meta: Any) -> None:
-        await self.store.append(self.name, TapeEntry.event(name, payload, **meta))
+        tracing.event(f"bub.{name}", **payload)
+        await self.store.append(self.name, TapeEntry.event(name, payload, **(tracing.correlation() | meta)))
 
     async def read_messages(self) -> list[dict[str, Any]]:
         query = self.context.build_query(self.query())
@@ -317,6 +319,8 @@ async def handoff(
         **meta: Any,
     ) -> list[TapeEntry]:
         tape_name = self.name
+        meta = tracing.correlation() | meta
+        tracing.event("bub.handoff", anchor=name, state=state)
         entry = TapeEntry.anchor(name, state=state, **meta)
         event = TapeEntry.event("handoff", {"name": name, "state": state or {}}, **meta)
         await self.store.append(tape_name, entry)
@@ -340,7 +344,7 @@ async def record_chat(  # noqa: C901
         usage: dict[str, Any] | None = None,
     ) -> None:
         tape_name = self.name
-        meta = {"run_id": run_id}
+        meta = {"run_id": run_id, **tracing.correlation()}
         if system_prompt:
             await self.store.append(tape_name, TapeEntry.system(system_prompt, **meta))
         if context_error is not None:
diff --git a/src/bub/tools.py b/src/bub/tools.py
index a5881fc5..4e7fae33 100644
--- a/src/bub/tools.py
+++ b/src/bub/tools.py
@@ -16,6 +16,7 @@
 from bub.errors import BubError, ErrorKind
 from bub.hooks.interception import ToolCall, ToolCallResult
 from bub.tape import Tape
+from bub.tracing import Span
 
 if TYPE_CHECKING:
     from bub.hooks.interception import AgentHooks
@@ -201,6 +202,7 @@ async def execute_async(
         invocations: Sequence[tuple[Tool, dict[str, Any]]],
         *,
         context: ToolContext | None = None,
+        call_ids: Sequence[str] | None = None,
     ) -> ToolExecution:
         if not invocations:
             return ToolExecution(tool_results=[])
@@ -208,7 +210,10 @@ async def execute_async(
         results: list[Any] = []
         error: BubError | None = None
         gathered = await asyncio.gather(
-            *(self._handle_tool_response_async(tool_obj, tool_args, context) for tool_obj, tool_args in invocations),
+            *(
+                self._trace_tool_response(tool_obj, tool_args, context, call_ids[index] if call_ids else None)
+                for index, (tool_obj, tool_args) in enumerate(invocations)
+            ),
             return_exceptions=True,
         )
         for result in gathered:
@@ -225,6 +230,41 @@ async def execute_async(
 
         return ToolExecution(tool_results=results, error=error)
 
+    async def _trace_tool_response(
+        self,
+        tool: Tool,
+        arguments: dict[str, Any],
+        context: ToolContext | None,
+        call_id: str | None,
+    ) -> Any:
+        span = Span(
+            f"execute_tool {tool.name}",
+            {
+                "gen_ai.operation.name": "execute_tool",
+                "gen_ai.tool.name": tool.name,
+                "gen_ai.tool.type": "function",
+                "gen_ai.tool.call.id": call_id,
+                "bub.run_id": context.run_id if context else None,
+                "gen_ai.conversation.id": context.state.get("session_id") if context else None,
+            },
+        )
+        try:
+            with span.activate():
+                result = await self._handle_tool_response_async(tool, arguments, context, span=span)
+                if isinstance(result, _FailedToolResult):
+                    span.fail(result.error)
+                    span.set(**{
+                        "gen_ai.tool.call.result": result.error.as_dict() if result.result is None else result.result
+                    })
+                else:
+                    span.set(**{"gen_ai.tool.call.result": result})
+                return result
+        except BaseException as exc:
+            span.fail(exc)
+            raise
+        finally:
+            span.end()
+
     def _invoke_tool(
         self,
         *,
@@ -244,6 +284,8 @@ async def _handle_tool_response_async(
         tool_obj: Tool,
         tool_args: dict[str, Any],
         context: ToolContext | None,
+        *,
+        span: Span | None = None,
     ) -> Any:
         tool_name = tool_obj.name
         call = ToolCall(
@@ -255,10 +297,11 @@ async def _handle_tool_response_async(
         if self._hooks is not None and context is not None:
             hook_state["_runtime_tape"] = context.tape
         started = time.monotonic()
-        if self._hooks is not None:
-            call, short_circuit = await self._apply_before_tool_call(call, hook_state, started)
-            if short_circuit is not None:
-                return short_circuit()
+        call, short_circuit = await self._apply_before_tool_call(call, hook_state, started)
+        if span is not None:
+            span.set(**{"gen_ai.tool.call.arguments": call.arguments})
+        if short_circuit is not None:
+            return short_circuit()
 
         try:
             result = await self._invoke_normalized(tool_obj, call, context)
diff --git a/src/bub/tracing.py b/src/bub/tracing.py
new file mode 100644
index 00000000..5716fab7
--- /dev/null
+++ b/src/bub/tracing.py
@@ -0,0 +1,256 @@
+"""Optional GenAI telemetry. Importing Bub never configures an exporter."""
+
+from __future__ import annotations
+
+import asyncio
+import importlib
+import json
+import os
+from collections.abc import Iterator, Mapping
+from contextlib import contextmanager, suppress
+from contextvars import ContextVar
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    import opentelemetry.trace as otel
+else:
+    try:
+        otel: Any = importlib.import_module("opentelemetry.trace")
+    except ImportError:
+        otel = None
+
+_OPENINFERENCE_ATTRIBUTES = {
+    "gen_ai.provider.name": "llm.provider",
+    "gen_ai.request.model": "llm.model_name",
+    "gen_ai.response.model": "llm.model_name",
+    "gen_ai.usage.input_tokens": "llm.token_count.prompt",
+    "gen_ai.usage.output_tokens": "llm.token_count.completion",
+    "gen_ai.tool.name": "tool.name",
+    "gen_ai.tool.call.arguments": "input.value",
+    "gen_ai.tool.call.result": "output.value",
+}
+
+
+def configure_otlp() -> None:
+    """Configure opt-in HTTP export; preserve application-owned providers."""
+    if (
+        otel is None
+        or not (os.getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") or os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT"))
+        or os.getenv("OTEL_SDK_DISABLED", "").lower() == "true"
+        or not isinstance(otel.get_tracer_provider(), otel.ProxyTracerProvider)
+    ):
+        return
+    try:
+        from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+        from opentelemetry.sdk.trace import TracerProvider
+        from opentelemetry.sdk.trace.export import BatchSpanProcessor
+    except ImportError:
+        return
+
+    protocol = os.getenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL") or os.getenv(
+        "OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf"
+    )
+    if protocol != "http/protobuf":
+        raise ValueError("Bub's trace extra supports OTLP http/protobuf only.")
+
+    exporter = OTLPSpanExporter()
+    # The SDK reads resource/sampler settings and drains the batch queue at process exit.
+    provider = TracerProvider()
+    provider.add_span_processor(BatchSpanProcessor(exporter))
+    otel.set_tracer_provider(provider)
+
+
+def _json(value: Any) -> str:
+    try:
+        return json.dumps(value, ensure_ascii=False, default=str)
+    except (TypeError, ValueError, RecursionError):
+        return '"[unserializable value]"'
+
+
+def _parts(message: Mapping[str, Any]) -> list[dict[str, Any]]:
+    content = message.get("content")
+    if message.get("role") == "tool":
+        return [{"type": "tool_call_response", "id": message.get("tool_call_id", ""), "response": content}]
+    parts: list[dict[str, Any]] = []
+    if isinstance(content, str) and content:
+        parts.append({"type": "text", "content": content})
+    elif isinstance(content, list):
+        for part in content:
+            if isinstance(part, dict) and part.get("type") == "text":
+                parts.append({"type": "text", "content": part.get("text", "")})
+            elif isinstance(part, dict):
+                # Preserve media structure without copying inline binary payloads.
+                parts.append({"type": "text", "content": f"[{part.get('type', 'media')} omitted]"})
+    for call in message.get("tool_calls") or []:
+        function = call.get("function", {})
+        arguments = function.get("arguments", {})
+        if isinstance(arguments, str):
+            with suppress(ValueError):
+                arguments = json.loads(arguments)
+        parts.append({
+            "type": "tool_call",
+            "id": call.get("id", ""),
+            "name": function.get("name", ""),
+            "arguments": arguments,
+        })
+    return parts
+
+
+class Span:
+    """A span whose lifetime is independent of its task-local activation."""
+
+    def __init__(self, name: str, attributes: Mapping[str, Any] | None = None) -> None:
+        self._operation = str((attributes or {}).get("gen_ai.operation.name", ""))
+        if otel is not None:
+            self._span = otel.get_tracer("bub").start_span(
+                name,
+                kind=otel.SpanKind.CLIENT if self._operation == "chat" else otel.SpanKind.INTERNAL,
+                attributes={k: v for k, v in (attributes or {}).items() if isinstance(v, str | bool | int | float)},
+            )
+        else:
+            self._span = None
+        self._ended = False
+        self.set(**(attributes or {}))
+        self.set(**{
+            "openinference.span.kind": {
+                "invoke_agent": "AGENT",
+                "chat": "LLM",
+                "execute_tool": "TOOL",
+            }.get(self._operation)
+        })
+
+    @property
+    def recording(self) -> bool:
+        return self._span is not None and self._span.is_recording()
+
+    def set(self, **attributes: Any) -> None:
+        if not self.recording:
+            return
+        for key, alias in _OPENINFERENCE_ATTRIBUTES.items():
+            if (value := attributes.get(key)) is not None:
+                if alias in {"input.value", "output.value"}:
+                    attributes[alias] = value if isinstance(value, str) else _json(value)
+                    attributes[alias.replace("value", "mime_type")] = (
+                        "text/plain" if isinstance(value, str) else "application/json"
+                    )
+                else:
+                    attributes[alias] = value
+        for key, value in attributes.items():
+            if value is not None:
+                self._span.set_attribute(
+                    key,
+                    value
+                    if isinstance(value, str | bool | int | float)
+                    or (
+                        key == "gen_ai.response.finish_reasons"
+                        and isinstance(value, list | tuple)
+                        and all(isinstance(v, str) for v in value)
+                    )
+                    else _json(value),
+                )
+
+    def rename(self, name: str) -> None:
+        if self.recording:
+            self._span.update_name(name)
+
+    def messages(self, key: str, messages: list[dict[str, Any]]) -> None:
+        if self.recording:
+            normalized = [{"role": m.get("role", "user"), "parts": _parts(m)} for m in messages]
+            direction = "input" if key == "gen_ai.input.messages" else "output"
+            self.set(**{
+                key: _json(normalized),
+                f"{direction}.value": _json(normalized),
+                f"{direction}.mime_type": "application/json",
+            })
+            if self._operation == "chat":
+                self.set(**_openinference_messages(direction, normalized))
+
+    def event(self, name: str, attributes: Mapping[str, Any]) -> None:
+        if self.recording:
+            self._span.add_event(
+                name,
+                {
+                    k: v if isinstance(v, str | bool | int | float) else _json(v)
+                    for k, v in attributes.items()
+                    if v is not None
+                },
+            )
+
+    def fail(self, error: BaseException) -> None:
+        if not self.recording:
+            return
+        if isinstance(error, asyncio.CancelledError | GeneratorExit):
+            self.set(**{"bub.cancelled": True})
+        else:
+            self._span.record_exception(error)
+            self._span.set_status(otel.Status(otel.StatusCode.ERROR, str(error)))
+            self.set(**{"error.type": type(error).__name__})
+
+    def end(self) -> None:
+        if not self._ended:
+            self._ended = True
+            if self._span is not None:
+                self._span.end()
+
+    @contextmanager
+    def activate(self) -> Iterator[None]:
+        token = _CURRENT.set(self)
+        try:
+            if self._span is None:
+                yield
+            else:
+                with otel.use_span(
+                    self._span, end_on_exit=False, record_exception=False, set_status_on_exception=False
+                ):
+                    yield
+        finally:
+            _CURRENT.reset(token)
+
+    def correlation(self) -> dict[str, str]:
+        if self._span is None:
+            return {}
+        context = self._span.get_span_context()
+        if not context.is_valid:
+            return {}
+        return {"trace_id": f"{context.trace_id:032x}", "span_id": f"{context.span_id:016x}"}
+
+
+_CURRENT: ContextVar[Span | None] = ContextVar("bub_trace_span", default=None)
+
+
+def _openinference_messages(direction: str, messages: list[dict[str, Any]]) -> dict[str, Any]:
+    attributes: dict[str, Any] = {}
+    for index, message in enumerate(messages):
+        prefix = f"llm.{direction}_messages.{index}.message"
+        attributes[f"{prefix}.role"] = message["role"]
+        text: list[str] = []
+        call_index = 0
+        for part in message["parts"]:
+            if part["type"] == "text":
+                text.append(part["content"])
+            elif part["type"] == "tool_call_response":
+                attributes[f"{prefix}.tool_call_id"] = part["id"]
+                text.append(part["response"] if isinstance(part["response"], str) else _json(part["response"]))
+            elif part["type"] == "tool_call":
+                call_prefix = f"{prefix}.tool_calls.{call_index}.tool_call"
+                attributes[f"{call_prefix}.id"] = part["id"]
+                attributes[f"{call_prefix}.function.name"] = part["name"]
+                attributes[f"{call_prefix}.function.arguments"] = _json(part["arguments"])
+                call_index += 1
+        if text:
+            attributes[f"{prefix}.content"] = "\n".join(text)
+    return attributes
+
+
+def current_span() -> Span | None:
+    return _CURRENT.get()
+
+
+def correlation() -> dict[str, str]:
+    span = current_span()
+    return span.correlation() if span else {}
+
+
+def event(event_name: str, **attributes: Any) -> None:
+    if span := current_span():
+        span.event(event_name, attributes)
diff --git a/tests/test_errors.py b/tests/test_errors.py
new file mode 100644
index 00000000..bf3628eb
--- /dev/null
+++ b/tests/test_errors.py
@@ -0,0 +1,19 @@
+from collections.abc import Iterator
+from contextlib import contextmanager
+
+import pytest
+
+from bub.errors import BubError, ErrorKind
+
+
+def test_error_survives_generator_context_manager_with_identity_and_details() -> None:
+    @contextmanager
+    def boundary() -> Iterator[None]:
+        yield
+
+    error = BubError(ErrorKind.NOT_FOUND, "missing", {"handle": "unknown"})
+    with pytest.raises(BubError) as caught, boundary():
+        raise error
+
+    assert caught.value is error
+    assert caught.value.as_dict() == {"kind": "not_found", "message": "missing", "details": {"handle": "unknown"}}
diff --git a/tests/test_otlp.py b/tests/test_otlp.py
new file mode 100644
index 00000000..caa563ab
--- /dev/null
+++ b/tests/test_otlp.py
@@ -0,0 +1,177 @@
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from collections.abc import Iterator
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from queue import Queue
+from threading import Thread
+
+import pytest
+
+
+def run_python(code: str, tmp_path: Path, **variables: str) -> None:
+    env = {k: v for k, v in os.environ.items() if not k.startswith(("OTEL_", "LOGFIRE_", "BUB_"))}
+    env.update(variables)
+    result = subprocess.run(
+        [sys.executable, "-c", code], cwd=tmp_path, env=env, capture_output=True, text=True, timeout=20
+    )
+    assert result.returncode == 0, result.stderr
+
+
+@pytest.fixture
+def collector() -> Iterator[tuple[str, Queue[tuple[str, str | None, bytes]]]]:
+    pytest.importorskip("opentelemetry.sdk.trace")
+    requests: Queue[tuple[str, str | None, bytes]] = Queue()
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_POST(self) -> None:
+            requests.put((
+                self.path,
+                self.headers.get("x-project-name"),
+                self.rfile.read(int(self.headers["Content-Length"])),
+            ))
+            self.send_response(200)
+            self.end_headers()
+
+        def log_message(self, fmt: str, *args: object) -> None:
+            pass
+
+    with ThreadingHTTPServer(("127.0.0.1", 0), Handler) as server:
+        thread = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True)
+        thread.start()
+        try:
+            yield f"http://127.0.0.1:{server.server_port}", requests
+        finally:
+            server.shutdown()
+            thread.join()
+
+
+@pytest.mark.parametrize("generic_endpoint", [False, True])
+def test_cli_exports_once_and_flushes_at_exit_without_configuring_logfire(
+    tmp_path: Path, collector: tuple[str, Queue[tuple[str, str | None, bytes]]], generic_endpoint: bool
+) -> None:
+    pytest.importorskip("opentelemetry.sdk.trace")
+    from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
+
+    url, requests = collector
+    endpoint = (
+        {"OTEL_EXPORTER_OTLP_ENDPOINT": url}
+        if generic_endpoint
+        else {"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": f"{url}/v1/traces"}
+    )
+    run_python(
+        """
+try:
+    import logfire
+except ImportError:
+    pass
+else:
+    # Pydantic may import Logfire's plugin, but Bub must not configure its exporter.
+    def unexpected_configuration(*args, **kwargs):
+        raise SystemExit('OTLP must not configure Logfire')
+    logfire.configure = unexpected_configuration
+from bub.__main__ import _instrument_bub
+from bub.tracing import Span
+_instrument_bub('INFO')  # CLI import already configured it; repeat must not duplicate exports.
+root = Span('invoke_agent bub', {'gen_ai.operation.name': 'invoke_agent'})
+with root.activate():
+    child = Span('chat test', {'gen_ai.operation.name': 'chat'})
+    child.end()
+root.end()
+# No explicit flush: the process must export queued spans on normal exit.
+""",
+        tmp_path,
+        **endpoint,
+        OTEL_SERVICE_NAME="bub-test",
+        OTEL_EXPORTER_OTLP_HEADERS="x-project-name=test-project",
+        OTEL_BSP_SCHEDULE_DELAY="600000",
+    )
+    path, project, body = requests.get(timeout=2)
+    assert path == "/v1/traces"
+    assert project == "test-project"
+    request = ExportTraceServiceRequest.FromString(body)
+    (resource,) = request.resource_spans
+    assert any(a.key == "service.name" and a.value.string_value == "bub-test" for a in resource.resource.attributes)
+    spans = [span for scope in resource.scope_spans for span in scope.spans]
+    assert {span.name for span in spans} == {"invoke_agent bub", "chat test"}
+    root = next(span for span in spans if span.name == "invoke_agent bub")
+    child = next(span for span in spans if span.name == "chat test")
+    assert child.parent_span_id == root.span_id
+    assert child.trace_id == root.trace_id
+    assert requests.empty()
+
+
+@pytest.mark.parametrize("mode", ["no_endpoint", "disabled", "missing_sdk"])
+def test_otlp_stays_noop_when_not_enabled_or_dependencies_are_missing(tmp_path: Path, mode: str) -> None:
+    env = {} if mode == "no_endpoint" else {"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": "http://127.0.0.1:1/v1/traces"}
+    if mode == "disabled":
+        env["OTEL_SDK_DISABLED"] = "true"
+    blocker = (
+        """
+import sys
+from importlib.abc import MetaPathFinder
+class BlockSDK(MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname.startswith(('opentelemetry.sdk', 'opentelemetry.exporter')):
+            raise ModuleNotFoundError(fullname)
+sys.meta_path.insert(0, BlockSDK())
+"""
+        if mode == "missing_sdk"
+        else ""
+    )
+    run_python(
+        blocker
+        + """
+from bub.tracing import Span, configure_otlp
+configure_otlp()
+span = Span('noop')
+assert not span.recording
+span.end()
+""",
+        tmp_path,
+        **env,
+    )
+
+
+def test_otlp_preserves_application_provider(tmp_path: Path) -> None:
+    pytest.importorskip("opentelemetry.sdk.trace")
+    run_python(
+        """
+from opentelemetry import trace
+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 bub.tracing import Span, configure_otlp
+provider = TracerProvider()
+exporter = InMemorySpanExporter()
+provider.add_span_processor(SimpleSpanProcessor(exporter))
+trace.set_tracer_provider(provider)
+configure_otlp()
+assert trace.get_tracer_provider() is provider
+Span('existing').end()
+assert len(exporter.get_finished_spans()) == 1
+""",
+        tmp_path,
+        OTEL_EXPORTER_OTLP_TRACES_ENDPOINT="http://127.0.0.1:1/v1/traces",
+    )
+
+
+def test_otlp_rejects_unsupported_protocol(tmp_path: Path) -> None:
+    pytest.importorskip("opentelemetry.sdk.trace")
+    run_python(
+        """
+from bub.tracing import configure_otlp
+try:
+    configure_otlp()
+except ValueError as exc:
+    assert 'http/protobuf' in str(exc)
+else:
+    raise AssertionError('unsupported protocol silently accepted')
+""",
+        tmp_path,
+        OTEL_EXPORTER_OTLP_TRACES_ENDPOINT="http://127.0.0.1:1",
+        OTEL_EXPORTER_OTLP_TRACES_PROTOCOL="grpc",
+    )
diff --git a/tests/test_tracing.py b/tests/test_tracing.py
new file mode 100644
index 00000000..581eff4f
--- /dev/null
+++ b/tests/test_tracing.py
@@ -0,0 +1,454 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import subprocess
+import sys
+from collections.abc import AsyncIterator, Iterator
+from contextlib import aclosing
+from pathlib import Path
+from typing import Any
+
+import pytest
+from any_llm.constants import LLMProvider
+from any_llm.types.completion import ChatCompletion, ChatCompletionChunk
+
+from bub import tracing
+from bub.builtin.agent import Agent
+from bub.builtin.context import default_tape_context
+from bub.builtin.model_runner import ModelRunner
+from bub.builtin.settings import AgentSettings, ModelCandidate
+from bub.framework import BubFramework
+from bub.hooks import hookimpl
+from bub.hooks.interception import LlmCallDecision, ToolCallDecision, ToolCallResult
+from bub.store import AsyncTapeStoreAdapter, InMemoryTapeStore
+from bub.streaming import AsyncStreamEvents, StreamEvent
+from bub.tape import Tape
+from bub.tools import REGISTRY, Tool, ToolContext, ToolExecutor
+from bub.utils import workspace_from_state
+
+
+@pytest.fixture
+def spans(monkeypatch: pytest.MonkeyPatch) -> Iterator[Any]:
+    sdk = pytest.importorskip("opentelemetry.sdk.trace")
+    from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+    from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+
+    provider = sdk.TracerProvider()
+    exporter = InMemorySpanExporter()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    monkeypatch.setattr(tracing.otel, "get_tracer", provider.get_tracer)
+    yield exporter
+    provider.shutdown()
+
+
+@pytest.fixture
+def agent(tmp_path: Path) -> Agent:
+    framework = BubFramework(config_file=tmp_path / "config.yml")
+    framework.load_builtin_hooks()
+    agent = Agent(framework)
+    agent.settings = AgentSettings.model_construct(model="openai:test", api_key="unused", api_base=None)
+    agent.model_runner = ModelRunner(agent.settings, hooks=framework.get_agent_hooks())
+    agent.__dict__["tape"] = Tape(tmp_path, AsyncTapeStoreAdapter(InMemoryTapeStore()), default_tape_context())
+    return agent
+
+
+def completion(text: str = "done", calls: list[dict[str, Any]] | None = None) -> ChatCompletion:
+    return ChatCompletion.model_validate({
+        "id": "response-1",
+        "object": "chat.completion",
+        "created": 0,
+        "model": "actual-model",
+        "choices": [
+            {
+                "index": 0,
+                "finish_reason": "tool_calls" if calls else "stop",
+                "message": {"role": "assistant", "content": text, "tool_calls": calls},
+            }
+        ],
+        "usage": {"prompt_tokens": 10, "completion_tokens": 4, "total_tokens": 14},
+    })
+
+
+def call(name: str, call_id: str, arguments: dict[str, Any] | None = None) -> dict[str, Any]:
+    return {"id": call_id, "type": "function", "function": {"name": name, "arguments": json.dumps(arguments or {})}}
+
+
+@pytest.mark.asyncio
+async def test_agent_trajectory_has_parallel_tools_messages_and_tape_links(
+    spans: Any,
+    agent: Agent,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    started = 0
+    both_started = asyncio.Event()
+
+    async def handler() -> dict[str, str]:
+        nonlocal started
+        started += 1
+        if started == 2:
+            both_started.set()
+        await asyncio.wait_for(both_started.wait(), 1)
+        return {"value": "result"}
+
+    monkeypatch.setitem(REGISTRY, "trace_tool", Tool(name="trace_tool", handler=handler))
+    replies = iter([completion("checking", [call("trace_tool", "call-1"), call("trace_tool", "call-2")]), completion()])
+
+    async def respond(**kwargs: Any) -> ChatCompletion:
+        return next(replies)
+
+    monkeypatch.setattr(agent.model_runner, "completion_response", respond)
+    events = await agent.run_stream(session_id="trace-test", prompt="hello", state={}, allowed_tools=["trace_tool"])
+    async for _ in events:
+        assert tracing.current_span() is None
+        assert not tracing.otel.get_current_span().get_span_context().is_valid
+
+    finished = spans.get_finished_spans()
+    root = next(s for s in finished if s.name == "invoke_agent bub")
+    models = [s for s in finished if s.name.startswith("chat ")]
+    tools = [s for s in finished if s.name.startswith("execute_tool ")]
+    assert len(models) == len(tools) == 2
+    assert all(s.parent.span_id == root.context.span_id for s in [*models, *tools])
+    assert tools[0].start_time < tools[1].end_time and tools[1].start_time < tools[0].end_time
+    assert {s.attributes["gen_ai.tool.call.id"] for s in tools} == {"call-1", "call-2"}
+    assert all(s.attributes["gen_ai.response.model"] == "actual-model" for s in models)
+    assert all(s.attributes["gen_ai.usage.input_tokens"] == 10 for s in models)
+    assert "gen_ai.usage.input_tokens" not in root.attributes  # No double counting.
+    assert root.attributes["gen_ai.conversation.id"] == "trace-test"
+    assert root.attributes["openinference.span.kind"] == "AGENT"
+    assert all(s.attributes["openinference.span.kind"] == "LLM" for s in models)
+    assert all(s.attributes["openinference.span.kind"] == "TOOL" for s in tools)
+    assert models[0].attributes["llm.token_count.prompt"] == 10
+    assert models[0].attributes["llm.model_name"] == "actual-model"
+    assert models[0].attributes["llm.output_messages.0.message.tool_calls.0.tool_call.id"] == "call-1"
+    assert json.loads(tools[0].attributes["output.value"]) == {"value": "result"}
+    messages = json.loads(root.attributes["gen_ai.output.messages"])
+    assert [m["role"] for m in messages] == ["assistant", "tool", "tool", "assistant"]
+    assert messages[-1]["parts"] == [{"type": "text", "content": "done"}]
+    assert any(e.name == "bub.loop.step" for e in root.events)
+    entries = await agent.tape.store.fetch_all(agent.tape.session_tape("trace-test", workspace_from_state({})).query())
+    assert any(e.meta.get("trace_id") == f"{root.context.trace_id:032x}" for e in entries)
+
+
+@pytest.mark.asyncio
+async def test_subagent_is_nested_under_its_tool(spans: Any, agent: Agent, monkeypatch: pytest.MonkeyPatch) -> None:
+    replies = iter([
+        completion("", [call("subagent", "child", {"prompt": "child task"})]),
+        completion("child done"),
+        completion(),
+    ])
+
+    async def respond(**kwargs: Any) -> ChatCompletion:
+        return next(replies)
+
+    monkeypatch.setattr(agent.model_runner, "completion_response", respond)
+    events = await agent.run_stream(
+        session_id="parent", prompt="delegate", state={"_runtime_agent": agent}, allowed_tools=["subagent"]
+    )
+    async for _ in events:
+        pass
+    finished = spans.get_finished_spans()
+    roots = [s for s in finished if s.name == "invoke_agent bub"]
+    tool = next(s for s in finished if s.name == "execute_tool subagent")
+    assert len(roots) == 2
+    child = next(s for s in roots if s.parent is not None)
+    parent = next(s for s in roots if s.parent is None)
+    assert child.parent.span_id == tool.context.span_id
+    assert tool.parent.span_id == parent.context.span_id
+    assert child.context.trace_id == parent.context.trace_id
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("cancel", [False, True])
+async def test_stream_close_and_cancel_finish_spans_and_provider(
+    spans: Any,
+    agent: Agent,
+    monkeypatch: pytest.MonkeyPatch,
+    cancel: bool,
+) -> None:
+    closed = asyncio.Event()
+    waiting = asyncio.Event()
+
+    async def chunks() -> AsyncIterator[ChatCompletionChunk]:
+        try:
+            yield ChatCompletionChunk.model_validate({
+                "id": "chunk-1",
+                "object": "chat.completion.chunk",
+                "created": 0,
+                "model": "actual",
+                "choices": [{"index": 0, "delta": {"content": "first"}}],
+            })
+            waiting.set()
+            await asyncio.Event().wait()
+        finally:
+            closed.set()
+
+    async def respond(**kwargs: Any) -> AsyncIterator[ChatCompletionChunk]:
+        return chunks()
+
+    monkeypatch.setattr(agent.model_runner, "completion_response", respond)
+    events = await agent.run_stream(session_id="cancel-test", prompt="hello", state={}, allowed_tools=[])
+    assert (await anext(events)).kind == "text"
+    assert tracing.current_span() is None
+    assert len(spans.get_finished_spans()) == 0
+    if cancel:
+        task = asyncio.create_task(anext(events))
+        await waiting.wait()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+    else:
+        await events.aclose()
+    assert closed.is_set()
+    assert len(spans.get_finished_spans()) == 2
+    assert all(s.attributes.get("bub.cancelled") for s in spans.get_finished_spans())
+    assert tracing.current_span() is None
+    assert not tracing.otel.get_current_span().get_span_context().is_valid
+    await events.aclose()
+    assert len(spans.get_finished_spans()) == 2
+
+
+@pytest.mark.asyncio
+async def test_close_before_first_iteration_releases_tape(spans: Any, agent: Agent) -> None:
+    events = await agent.run_stream(session_id="unused", prompt="hello", state={}, allowed_tools=[])
+    await events.aclose()
+    assert len(spans.get_finished_spans()) == 1
+    # Fork contents were merged by the close callback even though the generator never started.
+    entries = await agent.tape.store.fetch_all(agent.tape.session_tape("unused", workspace_from_state({})).query())
+    assert any(e.payload.get("name") == "loop.start" for e in entries)
+
+
+@pytest.mark.asyncio
+async def test_setup_failure_releases_tape_and_finishes_agent_span(
+    spans: Any, agent: Agent, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    async def fail_setup(*, tape: Tape, **kwargs: Any) -> AsyncStreamEvents:
+        await tape.append_event("setup.started", {})
+        raise ValueError("setup failed")
+
+    monkeypatch.setattr(agent, "_agent_loop", fail_setup)
+    with pytest.raises(ValueError, match="setup failed"):
+        await agent.run_stream(session_id="setup-failure", prompt="hello", state={})
+    (span,) = spans.get_finished_spans()
+    assert span.status.status_code.name == "ERROR"
+    assert span.attributes["error.type"] == "ValueError"
+    entries = await agent.tape.store.fetch_all(
+        agent.tape.session_tape("setup-failure", workspace_from_state({})).query()
+    )
+    assert any(e.payload.get("name") == "setup.started" for e in entries)
+    assert tracing.current_span() is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("consume", [False, True])
+async def test_cleanup_failure_finishes_span_without_leaking_context(spans: Any, consume: bool) -> None:
+    async def source() -> AsyncIterator[StreamEvent]:
+        yield StreamEvent("text", {"delta": "ok"})
+
+    async def cleanup() -> None:
+        raise RuntimeError("cleanup failed")
+
+    events = AsyncStreamEvents(source(), span=tracing.Span("cleanup"), on_close=cleanup)
+    with pytest.raises(RuntimeError, match="cleanup failed"):
+        if consume:
+            async for _ in events:
+                pass
+        else:
+            await events.aclose()
+    await events.aclose()
+    (span,) = spans.get_finished_spans()
+    assert span.status.status_code.name == "ERROR"
+    assert span.attributes["error.type"] == "RuntimeError"
+    assert bool(span.attributes.get("bub.cancelled")) is not consume
+    assert tracing.current_span() is None
+    assert not tracing.otel.get_current_span().get_span_context().is_valid
+
+
+@pytest.mark.asyncio
+async def test_llm_short_circuit_has_no_phantom_model_span(spans: Any, agent: Agent) -> None:
+    class Finish:
+        @hookimpl
+        def before_llm_call(self) -> LlmCallDecision:
+            return LlmCallDecision.finish("stopped by policy")
+
+    agent.framework.register_plugin(Finish())
+    events = await agent.run_stream(session_id="policy", prompt="hello", state={}, allowed_tools=[])
+    assert [e.data["delta"] async for e in events if e.kind == "text"] == ["stopped by policy"]
+    assert [s.name for s in spans.get_finished_spans()] == ["invoke_agent bub"]
+
+
+@pytest.mark.asyncio
+async def test_failed_tool_records_effective_result_and_original_failure(spans: Any, agent: Agent) -> None:
+    class Policy:
+        @hookimpl(tryfirst=True)
+        def before_tool_call(self) -> ToolCallDecision:
+            return ToolCallDecision.deny("denied")
+
+        @hookimpl
+        def after_tool_call(self, result: ToolCallResult) -> None:
+            result.result = "bounded failure"
+
+    agent.framework.register_plugin(Policy())
+    execution = await ToolExecutor(agent.framework.get_agent_hooks()).execute_async(
+        [(Tool(name="denied", handler=lambda: pytest.fail("must not run")), {})],
+        context=ToolContext(agent.tape, "run-1"),
+        call_ids=["denied-1"],
+    )
+    assert execution.error is not None
+    (span,) = spans.get_finished_spans()
+    assert span.status.status_code.name == "ERROR"
+    assert span.attributes["gen_ai.tool.call.result"] == "bounded failure"
+    assert span.attributes["gen_ai.tool.call.id"] == "denied-1"
+
+
+@pytest.mark.asyncio
+async def test_provider_fallback_records_actual_model(
+    spans: Any, agent: Agent, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    class Client:
+        SUPPORTS_COMPLETION_STREAMING = False
+
+        async def acompletion(self, **kwargs: Any) -> ChatCompletion:
+            if kwargs["model"] == "unavailable":
+                raise RuntimeError("try next")
+            return completion()
+
+    candidates = [
+        ModelCandidate(name=f"openai:{name}", provider=LLMProvider.OPENAI, model_id=name)
+        for name in ["unavailable", "fallback"]
+    ]
+    monkeypatch.setattr(agent.model_runner, "iter_llm_clients", lambda model: iter((c, Client()) for c in candidates))
+    events = await agent.run_stream(session_id="fallback", prompt="hello", state={}, allowed_tools=[])
+    async for _ in events:
+        pass
+    model = next(s for s in spans.get_finished_spans() if s.name.startswith("chat "))
+    assert model.attributes["gen_ai.request.model"] == "fallback"
+    assert model.attributes["gen_ai.provider.name"] == "openai"
+    assert any(e.name == "bub.model.attempt_failed" for e in model.events)
+
+
+def test_missing_optional_dependencies_import_and_execute_as_noop() -> None:
+    code = """
+import asyncio
+import sys
+from importlib.abc import MetaPathFinder
+class BlockTelemetry(MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname.split('.')[0] in {'opentelemetry', 'logfire'}:
+            raise ModuleNotFoundError(fullname)
+sys.meta_path.insert(0, BlockTelemetry())
+from bub import tracing
+from bub.builtin.agent import Agent
+from bub.tools import Tool, ToolExecutor
+assert tracing.otel is None
+async def main():
+    span = tracing.Span('noop')
+    assert not span.recording
+    with span.activate():
+        result = await ToolExecutor().execute_async([(Tool(name='echo', handler=lambda: 'ok'), {})])
+        assert result.tool_results == ['ok']
+    span.end()
+    assert tracing.correlation() == {}
+asyncio.run(main())
+"""
+    result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True)
+    assert result.returncode == 0, result.stderr
+
+
+@pytest.mark.asyncio
+async def test_streaming_usage_arrives_on_last_chunk(spans: Any, agent: Agent, monkeypatch: pytest.MonkeyPatch) -> None:
+    async def chunks() -> AsyncIterator[ChatCompletionChunk]:
+        yield ChatCompletionChunk.model_validate({
+            "id": "stream",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": "stream-model",
+            "choices": [{"index": 0, "delta": {"content": "done"}, "finish_reason": "stop"}],
+        })
+        yield ChatCompletionChunk.model_validate({
+            "id": "stream",
+            "object": "chat.completion.chunk",
+            "created": 0,
+            "model": "stream-model",
+            "choices": [],
+            "usage": {"prompt_tokens": 19, "completion_tokens": 3, "total_tokens": 22},
+        })
+
+    async def respond(**kwargs: Any) -> AsyncIterator[ChatCompletionChunk]:
+        return chunks()
+
+    monkeypatch.setattr(agent.model_runner, "completion_response", respond)
+    events = await agent.run_stream(session_id="stream", prompt="hello", state={}, allowed_tools=[])
+    assert (await anext(events)).data == {"delta": "done"}
+    assert not spans.get_finished_spans()
+    async for _ in events:
+        pass
+    model = next(s for s in spans.get_finished_spans() if s.name.startswith("chat "))
+    assert model.attributes["gen_ai.usage.input_tokens"] == 19
+    assert model.attributes["gen_ai.usage.output_tokens"] == 3
+    assert model.attributes["gen_ai.response.finish_reasons"] == ("stop",)
+    assert "bub.cancelled" not in model.attributes
+
+
+@pytest.mark.asyncio
+async def test_timeout_marks_model_and_agent_as_failed(
+    spans: Any, agent: Agent, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    agent.settings.model_timeout_seconds = 0
+
+    async def respond(**kwargs: Any) -> ChatCompletion:
+        await asyncio.Event().wait()
+        return completion()
+
+    monkeypatch.setattr(agent.model_runner, "completion_response", respond)
+    events = await agent.run_stream(session_id="timeout", prompt="hello", state={}, allowed_tools=[])
+    with pytest.raises(TimeoutError):
+        async for _ in events:
+            pass
+    finished = spans.get_finished_spans()
+    assert len(finished) == 2
+    assert all(s.status.status_code.name == "ERROR" for s in finished)
+    assert all(s.attributes["error.type"] == "TimeoutError" for s in finished)
+
+
+@pytest.mark.asyncio
+async def test_concurrent_sessions_have_independent_traces(
+    spans: Any, agent: Agent, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    async def respond(**kwargs: Any) -> ChatCompletion:
+        await asyncio.sleep(0)
+        return completion()
+
+    monkeypatch.setattr(agent.model_runner, "completion_response", respond)
+
+    async def run(session: str) -> None:
+        events = await agent.run_stream(session_id=session, prompt="hello", state={}, allowed_tools=[])
+        async for _ in events:
+            assert tracing.current_span() is None
+
+    await asyncio.gather(run("one"), run("two"))
+    roots = [s for s in spans.get_finished_spans() if s.name == "invoke_agent bub"]
+    assert len({s.context.trace_id for s in roots}) == 2
+    for model in [s for s in spans.get_finished_spans() if s.name.startswith("chat ")]:
+        root = next(s for s in roots if s.context.trace_id == model.context.trace_id)
+        assert model.parent.span_id == root.context.span_id
+        assert model.attributes["gen_ai.conversation.id"] == root.attributes["gen_ai.conversation.id"]
+
+
+@pytest.mark.asyncio
+async def test_noop_stream_still_closes_when_dependencies_are_absent(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(tracing, "otel", None)
+    closed = False
+
+    async def source() -> AsyncIterator[StreamEvent]:
+        nonlocal closed
+        try:
+            yield StreamEvent("text", {"delta": "ok"})
+        finally:
+            closed = True
+
+    events = AsyncStreamEvents(source(), span=tracing.Span("noop"))
+    async with aclosing(events):
+        assert (await anext(events)).data == {"delta": "ok"}
+    assert closed
+    assert tracing.current_span() is None
diff --git a/uv.lock b/uv.lock
index 1acb233a..61698618 100644
--- a/uv.lock
+++ b/uv.lock
@@ -243,6 +243,14 @@ dependencies = [
 [package.optional-dependencies]
 logfire = [
     { name = "logfire" },
+    { name = "opentelemetry-api" },
+    { name = "opentelemetry-exporter-otlp-proto-http" },
+    { name = "opentelemetry-sdk" },
+]
+trace = [
+    { name = "opentelemetry-api" },
+    { name = "opentelemetry-exporter-otlp-proto-http" },
+    { name = "opentelemetry-sdk" },
 ]
 
 [package.dev-dependencies]
@@ -265,6 +273,12 @@ requires-dist = [
     { name = "inquirer-textual", specifier = ">=0.5.1" },
     { name = "logfire", marker = "extra == 'logfire'", specifier = ">=4.31.0" },
     { name = "loguru", specifier = ">=0.7.2" },
+    { name = "opentelemetry-api", marker = "extra == 'logfire'", specifier = ">=1.39.0" },
+    { name = "opentelemetry-api", marker = "extra == 'trace'", specifier = ">=1.39.0" },
+    { name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'logfire'", specifier = ">=1.39.0" },
+    { name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'trace'", specifier = ">=1.39.0" },
+    { name = "opentelemetry-sdk", marker = "extra == 'logfire'", specifier = ">=1.39.0" },
+    { name = "opentelemetry-sdk", marker = "extra == 'trace'", specifier = ">=1.39.0" },
     { name = "pluggy", specifier = ">=1.6.0" },
     { name = "prompt-toolkit", specifier = ">=3.0.0" },
     { name = "pydantic", specifier = ">=2.0.0" },
@@ -276,7 +290,7 @@ requires-dist = [
     { name = "typer", specifier = ">=0.9.0" },
     { name = "typing-extensions", specifier = ">=4.13.0" },
 ]
-provides-extras = ["logfire"]
+provides-extras = ["logfire", "trace"]
 
 [package.metadata.requires-dev]
 dev = [
diff --git a/website/src/content/docs/docs/tutorials/observability.mdx b/website/src/content/docs/docs/tutorials/observability.mdx
index 00a40864..21375da5 100644
--- a/website/src/content/docs/docs/tutorials/observability.mdx
+++ b/website/src/content/docs/docs/tutorials/observability.mdx
@@ -1,14 +1,14 @@
 ---
-title: Observe Bub with tapes and Phoenix
-description: Use Bub's tape as a first observability surface, then export OpenTelemetry GenAI traces to Phoenix.
+title: Observe Bub with tapes, Phoenix, and Logfire
+description: Inspect Bub's tape and export native agent, model, and tool traces to Logfire or an OTLP backend.
 sidebar:
   order: 1
 ---
 
-This tutorial gives you two observability paths for one Bub workspace:
+This tutorial gives you local tape inspection and native OpenTelemetry traces for one Bub workspace:
 
 1. Run a small natural-language task, then ask Bub about the tape it just wrote. This works without a tracing backend because Bub records each session as an append-only tape.
-2. Send OpenTelemetry telemetry to Phoenix while running the same kind of task. Use this when you want GenAI traces outside the local workspace.
+2. Send the trajectory to Phoenix or Logfire, including model calls, parallel tools, and nested subagents.
 
 By the end, you will have a quick local health check and a Phoenix trace view for agent, model, and tool activity.
 
@@ -19,17 +19,13 @@ You need:
 - Bub installed and runnable with `bub --help`.
 - One workspace where `bub run "What tools do you have?"` can call your configured model.
 - Docker or Podman if you want to run Phoenix locally.
-- The Logfire extra installed before starting Bub with Phoenix:
+- The `trace` extra installed in Bub's activated virtual environment before starting Bub with Phoenix:
 
 ```bash
-uv sync --extra logfire
+uv pip install "bub[trace]"
 ```
 
-- The `bub-tapestore-otel` contrib plugin installed for richer tape and agent spans:
-
-```bash
-bub install bub-tapestore-otel@main
-```
+The `trace` extra includes the OpenTelemetry API, SDK, and HTTP/protobuf OTLP exporter. The Phoenix walkthrough below requires no Logfire installation or configuration.
 
 ## 1. Ask Bub for its current tape
 
@@ -101,37 +97,33 @@ Open the UI:
 http://localhost:6006
 ```
 
-## 4. Run Bub with contrib and OTLP
+## 4. Run Bub with OTLP
 
-Bub already supports Logfire during CLI startup when the `logfire` extra is installed. The `bub-tapestore-otel` contrib plugin adds GenAI-oriented spans from Bub's tape store, including `invoke_agent bub`, `bub.agent.step`, chat, and tool execution spans. This tutorial sends both to Phoenix through OTLP.
+Bub exports native GenAI spans directly to Phoenix using the OpenTelemetry SDK. Logfire is not required.
 
 In another terminal, run:
 
 ```bash
-LOGFIRE_SEND_TO_LOGFIRE=false \
-LOGFIRE_SERVICE_NAME=bub \
-BUB_TAPESTORE_OTEL_ENABLED=true \
-BUB_TAPESTORE_OTEL_SERVICE_NAME=bub \
-BUB_TAPESTORE_OTEL_AGENT_NAME=bub \
+OTEL_SERVICE_NAME=bub \
 OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://localhost:6006/v1/traces \
 OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/protobuf \
-uv run --extra logfire bub run "What tools do you have, and what small tasks are they useful for?"
+bub run "What tools do you have, and what small tasks are they useful for?"
 ```
 
 Then run the local tape check with the same telemetry settings:
 
 ```bash
-LOGFIRE_SEND_TO_LOGFIRE=false \
-LOGFIRE_SERVICE_NAME=bub \
-BUB_TAPESTORE_OTEL_ENABLED=true \
-BUB_TAPESTORE_OTEL_SERVICE_NAME=bub \
-BUB_TAPESTORE_OTEL_AGENT_NAME=bub \
+OTEL_SERVICE_NAME=bub \
 OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://localhost:6006/v1/traces \
 OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/protobuf \
-uv run --extra logfire bub run ",tape.info"
+bub run ",tape.info"
 ```
 
-Use `LOGFIRE_SEND_TO_LOGFIRE=false` for local Phoenix tutorials so Bub does not try to send telemetry to the hosted Logfire backend. `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` points OpenTelemetry exporters at Phoenix's OTLP HTTP endpoint.
+`OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` is the complete trace URL. Alternatively, set `OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:6006`; the exporter appends `/v1/traces`. The trace-specific variable takes precedence. Bub supports `http/protobuf` (the default), with headers, timeouts, resource attributes, and sampling configured through standard OpenTelemetry environment variables.
+
+An explicit OTLP endpoint selects direct export even if Logfire is installed; Bub does not also configure Logfire. Existing tracer providers are preserved, repeated initialization does not add exporters, and `OTEL_SDK_DISABLED=true` disables this initialization. The SDK batches exports and drains queued spans on normal process exit.
+
+For an authenticated destination, set `OTEL_EXPORTER_OTLP_TRACES_HEADERS` (or `OTEL_EXPORTER_OTLP_HEADERS`). To select a Phoenix project, include `x-project-name=my-project` in those headers; otherwise Phoenix uses its default project.
 
 ## 5. Inspect the trace in Phoenix
 
@@ -139,7 +131,7 @@ In Phoenix:
 
 1. Open the default project.
 2. Open the most recent trace.
-3. Look for `invoke_agent bub`, `bub.agent.step`, `chat <model>`, and `execute_tool <tool>` spans.
+3. Look for `invoke_agent bub`, `chat <model>`, and `execute_tool <tool>` spans.
 
 ![A Phoenix screenshot showing Bub GenAI telemetry exported through OTLP](/docs/observability/phoenix-otel.png)
 
@@ -150,6 +142,55 @@ This path complements tape inspection:
 
 Use both when debugging production behavior: start with `,tape.info` to understand the session state, then use Phoenix to inspect timing, errors, model calls, and tool calls.
 
+## Send trajectories to Logfire
+
+To send trajectories to Logfire, install the `logfire` extra, which includes `bub[trace]` and Logfire. With no explicit OTLP endpoint configured, Bub's CLI configures Logfire at startup; no tracing plugin is required.
+
+When switching from Phoenix, clear both OTLP endpoint variables, authenticate with your Logfire project, then run Bub:
+
+```bash
+unset OTEL_EXPORTER_OTLP_TRACES_ENDPOINT OTEL_EXPORTER_OTLP_ENDPOINT
+uv pip install "bub[logfire]"
+logfire auth
+logfire projects use
+bub run "Inspect this repository and summarize its structure."
+```
+
+For deployment, provide `LOGFIRE_TOKEN` through your environment. In Logfire's Agents view, look for `bub`. Each invocation contains `chat <model>` and `execute_tool <tool>` spans. Subagents are nested under the tool that launched them, and runs in the same session share `gen_ai.conversation.id`.
+
+## Trace contents and lifecycle
+
+The spans record model inputs and outputs, tool arguments and effective results (after result hooks and spill), token usage when supplied by the provider, and errors. Inline media payloads are omitted from message telemetry. Model fallback updates the model attributes and records failed attempts as events. Token usage is recorded on model spans only, avoiding duplicate counts on agent spans.
+
+Tape events such as loop steps, handoffs, and spill writes appear on the active span. Tape event and chat metadata contain `trace_id` and `span_id` for correlation. Tape storage remains independent of telemetry export.
+
+Without the optional dependencies, tracing is a no-op. Installing `bub[trace]` alone does not enable export: configure an OTLP endpoint or supply your own tracer provider. Importing Bub does not configure telemetry. Embedders can call `configure_otlp()` after setting the endpoint environment variable, or configure their own provider:
+
+```python
+from bub.tracing import configure_otlp
+
+configure_otlp()  # Reads OTEL_EXPORTER_OTLP_TRACES_ENDPOINT or OTEL_EXPORTER_OTLP_ENDPOINT.
+# Create and run Bub using your normal embedding entrypoint.
+```
+
+Streams expose `aclose()`. Embedders that may stop consuming early should use `contextlib.aclosing(stream)` so provider streams, tape forks, and spans close promptly. Trace context is activated only while advancing or closing a stream; it does not remain attached while consumer code handles an event. Cancellation closes spans and sets `bub.cancelled`; model failures and timeouts set error status.
+
+### Migrating from `bub-tapestore-otel`
+
+Use native tracing for live agent trajectories. The contrib plugin projects batches of committed tape entries into spans after the work has finished; its span timestamps describe the projection, while step duration is stored separately as an attribute. Native spans measure execution directly, preserve parallel tool timing and subagent parentage, and close on cancellation even when no terminal tape entry is committed.
+
+The plugin remains useful when the intended observation is committed tape writes. Native tracing does not replace tape persistence or reconstruct past traces from existing tapes.
+
+To migrate:
+
+1. Install the `trace` extra for Phoenix/OTLP, or the `logfire` extra for Logfire, and configure the destination as shown here.
+2. Set `BUB_TAPESTORE_OTEL_ENABLED=false` if the contrib plugin is still installed.
+3. Keep your existing tape backend. Native tracing does not wrap or replace it.
+
+Native spans also include OpenInference classification, model/token, message, and tool input/output attributes for Phoenix. These compatibility attributes are separate from the standard `gen_ai.*` attributes. There is no separate `bub.agent.step` span: loop steps are events on the agent span.
+
+See the [contrib plugin implementation](https://github.com/bubbuild/bub-contrib/tree/main/packages/bub-tapestore-otel) for the tape-projection approach.
+
 ## Clean up
 
 Stop Phoenix with `Ctrl+C` if it is running in the foreground. If it is detached, remove it:
diff --git a/website/src/content/docs/zh-cn/docs/tutorials/observability.mdx b/website/src/content/docs/zh-cn/docs/tutorials/observability.mdx
index bf0384d6..0c58b5d4 100644
--- a/website/src/content/docs/zh-cn/docs/tutorials/observability.mdx
+++ b/website/src/content/docs/zh-cn/docs/tutorials/observability.mdx
@@ -1,14 +1,14 @@
 ---
-title: 使用 tape 与 Phoenix 观察 Bub
-description: 先把 Bub 的 tape 当作第一层可观测性，再把 OpenTelemetry GenAI trace 导出到 Phoenix。
+title: 使用 tape、Phoenix 与 Logfire 观察 Bub
+description: 检查 Bub 的 tape，并将原生 agent、模型和工具 trace 导出到 Logfire 或 OTLP 后端。
 sidebar:
   order: 1
 ---
 
-本教程提供两条观察同一个 Bub workspace 的路径：
+本教程介绍同一个 Bub workspace 的本地 tape 检查和原生 OpenTelemetry trace：
 
 1. 先运行一个小的英文自然语言任务，再询问 Bub 刚写入的 tape。由于 Bub 会把每个 session 记录为 append-only tape，这条路径不依赖外部 tracing backend。
-2. 在运行同类任务时，将 OpenTelemetry telemetry 发送到 Phoenix。需要把 GenAI trace 放到本地或生产可观测平台时使用这条路径。
+2. 将执行轨迹发送到 Phoenix 或 Logfire，包括模型调用、并发工具和嵌套子 agent。
 
 完成后，你会得到一个本地快速健康检查方式，以及一个用于查看 agent、model 和 tool 活动的 Phoenix trace 视图。
 
@@ -19,17 +19,13 @@ sidebar:
 - Bub 已安装，且 `bub --help` 可以运行。
 - 一个 workspace，其中 `bub run "What tools do you have?"` 能调用已配置的模型。
 - 如果要本地运行 Phoenix，需要 Docker 或 Podman。
-- 启动带 Phoenix 的 Bub 之前，安装 Logfire extra：
+- 启动带 Phoenix 的 Bub 之前，在已激活的 Bub 虚拟环境中安装 `trace` extra：
 
 ```bash
-uv sync --extra logfire
+uv pip install "bub[trace]"
 ```
 
-- 安装 `bub-tapestore-otel` contrib 插件，获得更丰富的 tape 与 agent span：
-
-```bash
-bub install bub-tapestore-otel@main
-```
+`trace` extra 包含 OpenTelemetry API、SDK 和 HTTP/protobuf OTLP exporter。以下 Phoenix 流程无需安装或配置 Logfire。
 
 ## 1. 询问 Bub 当前 tape
 
@@ -101,37 +97,33 @@ docker run --rm --name bub-phoenix \
 http://localhost:6006
 ```
 
-## 4. 使用 contrib 与 OTLP 运行 Bub
+## 4. 使用 OTLP 运行 Bub
 
-安装 `logfire` extra 后，Bub 在 CLI 启动时已经支持 Logfire。`bub-tapestore-otel` contrib 插件会从 Bub tape store 生成面向 GenAI 的 span，包括 `invoke_agent bub`、`bub.agent.step`、chat 和 tool execution span。本教程通过 OTLP 把两者发送到 Phoenix。
+Bub 通过 OpenTelemetry SDK 将原生 GenAI span 直接发送到 Phoenix，无需 Logfire。
 
 在另一个终端运行：
 
 ```bash
-LOGFIRE_SEND_TO_LOGFIRE=false \
-LOGFIRE_SERVICE_NAME=bub \
-BUB_TAPESTORE_OTEL_ENABLED=true \
-BUB_TAPESTORE_OTEL_SERVICE_NAME=bub \
-BUB_TAPESTORE_OTEL_AGENT_NAME=bub \
+OTEL_SERVICE_NAME=bub \
 OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://localhost:6006/v1/traces \
 OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/protobuf \
-uv run --extra logfire bub run "What tools do you have, and what small tasks are they useful for?"
+bub run "What tools do you have, and what small tasks are they useful for?"
 ```
 
 然后用相同 telemetry 设置运行本地 tape 检查：
 
 ```bash
-LOGFIRE_SEND_TO_LOGFIRE=false \
-LOGFIRE_SERVICE_NAME=bub \
-BUB_TAPESTORE_OTEL_ENABLED=true \
-BUB_TAPESTORE_OTEL_SERVICE_NAME=bub \
-BUB_TAPESTORE_OTEL_AGENT_NAME=bub \
+OTEL_SERVICE_NAME=bub \
 OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://localhost:6006/v1/traces \
 OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/protobuf \
-uv run --extra logfire bub run ",tape.info"
+bub run ",tape.info"
 ```
 
-本地 Phoenix 教程建议设置 `LOGFIRE_SEND_TO_LOGFIRE=false`，避免 Bub 尝试把 telemetry 发送到托管 Logfire backend。`OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` 指向 Phoenix 的 OTLP HTTP endpoint。
+`OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` 是完整的 trace URL。也可以设置 `OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:6006`，exporter 会追加 `/v1/traces`；trace 专用变量优先。Bub 支持 `http/protobuf`（默认值），headers、timeout、resource attributes 和采样均通过标准 OpenTelemetry 环境变量配置。
+
+显式 OTLP endpoint 会选择直接导出，即使安装了 Logfire 也不会再配置 Logfire。已有 tracer provider 会被保留，重复初始化不会重复添加 exporter，`OTEL_SDK_DISABLED=true` 会禁用这条初始化路径。SDK 批量导出，并在进程正常退出时发送队列中剩余的 span。
+
+需要认证时，设置 `OTEL_EXPORTER_OTLP_TRACES_HEADERS`（或 `OTEL_EXPORTER_OTLP_HEADERS`）。如需指定 Phoenix 项目，在 headers 中加入 `x-project-name=my-project`；否则使用 default 项目。
 
 ## 5. 在 Phoenix 中检查 trace
 
@@ -139,7 +131,7 @@ uv run --extra logfire bub run ",tape.info"
 
 1. 打开 default project。
 2. 打开最近的 trace。
-3. 查找 `invoke_agent bub`、`bub.agent.step`、`chat <model>` 和 `execute_tool <tool>` span。
+3. 查找 `invoke_agent bub`、`chat <model>` 和 `execute_tool <tool>` span。
 
 ![展示 Bub GenAI telemetry 通过 OTLP 导出到 Phoenix 的截图](/docs/observability/phoenix-otel.png)
 
@@ -150,6 +142,55 @@ uv run --extra logfire bub run ",tape.info"
 
 排查生产行为时建议两者一起使用：先用 `,tape.info` 判断 session 状态，再用 Phoenix 查看耗时、错误、model call 和 tool call。
 
+## 将执行轨迹发送到 Logfire
+
+如需将执行轨迹发送到 Logfire，安装包含 `bub[trace]` 和 Logfire 的 `logfire` extra。没有显式配置 OTLP endpoint 时，Bub CLI 会在启动时配置 Logfire，无需额外的 tracing 插件。
+
+从 Phoenix 切换时，先清除两个 OTLP endpoint 环境变量，再登录并选择 Logfire 项目，然后运行 Bub：
+
+```bash
+unset OTEL_EXPORTER_OTLP_TRACES_ENDPOINT OTEL_EXPORTER_OTLP_ENDPOINT
+uv pip install "bub[logfire]"
+logfire auth
+logfire projects use
+bub run "检查这个仓库并总结它的结构。"
+```
+
+部署时通过环境变量提供 `LOGFIRE_TOKEN`。在 Logfire 的 Agents 页面查找 `bub`。每次执行包含 `chat <model>` 和 `execute_tool <tool>` span；子 agent 位于启动它的工具 span 下，同一 session 的执行通过 `gen_ai.conversation.id` 关联。
+
+## Trace 内容与生命周期
+
+Span 记录模型输入输出、工具参数和最终结果（结果 hook 与 spill 处理之后）、provider 返回的 token usage，以及错误。消息中的内联媒体内容不会上传。模型 fallback 会更新模型属性，并把失败尝试记录为事件。Token usage 只写入模型 span，避免与 agent span 重复计数。
+
+Loop step、handoff 和 spill 写入等 tape 事件会附加到当前 span。Tape 事件及聊天记录的 metadata 包含 `trace_id` 和 `span_id`，便于关联查询。Tape 存储与遥测导出相互独立。
+
+缺少可选依赖时 tracing 自动成为 no-op。仅安装 `bub[trace]` 不会启用导出，还需要配置 OTLP endpoint 或提供自己的 tracer provider。导入 Bub 不会自动配置遥测。嵌入使用方可以在设置 endpoint 环境变量后调用 `configure_otlp()`，或自行配置 provider：
+
+```python
+from bub.tracing import configure_otlp
+
+configure_otlp()  # 读取 OTEL_EXPORTER_OTLP_TRACES_ENDPOINT 或 OTEL_EXPORTER_OTLP_ENDPOINT。
+# 通过应用现有的嵌入入口创建并运行 Bub。
+```
+
+流提供 `aclose()`。嵌入使用方若可能提前停止消费，应使用 `contextlib.aclosing(stream)`，及时关闭 provider 流、tape fork 和 span。Trace 上下文仅在恢复迭代或关闭流时激活，不会残留在消费者处理事件的代码中。取消会关闭 span 并设置 `bub.cancelled`；模型失败和超时会设置错误状态。
+
+### 从 `bub-tapestore-otel` 迁移
+
+实时 agent trajectory 推荐使用原生 tracing。Contrib 插件会在工作完成、tape 提交后，把一批记录投影为 span；其 span 时间戳反映投影过程，step 耗时另存为属性。原生 span 直接测量执行耗时，保留并发工具的时间关系和子 agent 的父子关系，并且在未提交终止 tape 事件的取消场景中也能关闭。
+
+如果关注的是已提交的 tape 写入，旧插件仍有用途。原生 tracing 不替代 tape 持久化，也不会从已有 tape 重建历史 trace。
+
+迁移步骤：
+
+1. Phoenix/OTLP 安装 `trace` extra，Logfire 安装 `logfire` extra，并按本页配置目标。
+2. 如果仍安装着 contrib 插件，设置 `BUB_TAPESTORE_OTEL_ENABLED=false`。
+3. 保留现有 tape backend。原生 tracing 不包装或替换它。
+
+原生 span 同时提供 Phoenix 所需的 OpenInference 分类、模型/token、消息和工具输入输出属性。这些兼容属性与标准 `gen_ai.*` 属性分开。原生方案不生成独立的 `bub.agent.step` span，loop step 记录为 agent span 上的事件。
+
+Tape 投影方式的细节见 [contrib 插件实现](https://github.com/bubbuild/bub-contrib/tree/main/packages/bub-tapestore-otel)。
+
 ## 清理
 
 如果 Phoenix 在前台运行，用 `Ctrl+C` 停止。若以 detached 方式运行，删除容器：