# TraceView API


`TraceView` shows an OpenTelemetry trace as a waterfall. Each span is a row, indented under its parent, with its state (pending, running, ok, error), a bar on a shared timeline, and its duration. Click a row to see its attributes and events, fold a subtree with `−`/`+`, and press **Replay** (or drag the scrubber) to watch the trace unfold. Span *links* — "caused by, but not part of", like a queue consumer pointing at the producer — are drawn as dashed curves.


There are three ways to get spans in:



- **Live, in-process** — `view.span_processor()` is an OpenTelemetry `SpanProcessor`. Add it to a `TracerProvider` once; from then on every span made by that provider's tracers streams into the widget as it starts and ends, and running spans grow live. There is no widget-level context manager for this because OpenTelemetry already has one: `with
  tracer.start_as_current_span(...)` is the block, and the widget sees it. Needs `opentelemetry-sdk` (`pip install wigglystuff[otel]`). Anything built on OpenTelemetry, such as Logfire or Pydantic AI, can use the same processor.

- **From data** — `TraceView.from_otlp(...)` reads OTLP/JSON (a dict, a list of dicts, a file, or the JSON lines the Collector's file exporter writes), and `TraceView.from_spans(...)` takes SDK spans such as `InMemorySpanExporter.get_finished_spans()`. Loading OTLP needs no OpenTelemetry install.

- **From W&B Weave** — `with view.capture_weave():` shows the calls of any `@weave.op` function called inside the block, live, after `weave.init(...)` (`pip install wigglystuff[weave]`). Weave has no processor hook, so this context manager is the Weave stand-in for `add_span_processor`: it hooks the Weave client for the duration of the block and unhooks it afterwards. The spans themselves still come from your `@weave.op` calls, just as they come from `start_as_current_span` with OpenTelemetry. `TraceView.from_weave(...)` loads stored calls from `client.get_calls(...)` (or the service API's call dicts).


```
from opentelemetry.sdk.trace import TracerProvider
from wigglystuff import TraceView

view = TraceView(title="agent run")
provider = TracerProvider()
provider.add_span_processor(view.span_processor())
tracer = provider.get_tracer("notebook")

with tracer.start_as_current_span("agent.run"):  # this block is the trace
    ...
```


With Weave, call your op as usual inside the block:


```
import weave
from wigglystuff import TraceView

weave.init("my-project")
view = TraceView(title="support bot")
view  # display it, then in another cell:

with view.capture_weave():
    answer_ticket("Where is my order?")
```


A span with numeric `progress.done` and `progress.total` attributes gets a progress bar. Updates are sent from the thread that starts or ends a span, so spans finishing on background threads show up with the next update from the notebook's own thread. asyncio needs nothing extra: spans started inside tasks keep their parent through `contextvars`, so concurrent work shows as overlapping bars under one parent.


Picking an older trace from the menu pins it on screen. While *Follow latest* is on (the default, `show_latest=True`) the next trace that starts takes the view back to following live. Click the button, or set `show_latest=False`, to keep the picked trace pinned instead.


See also: AsyncFlow for a timeline of asyncio tasks without any instrumentation, and LiveEdit for a pass-by-pass trace of one Python function.


 Bases: `AnyWidget`


Waterfall viewer for OpenTelemetry traces, live or loaded from a file.


Each span is one row: a tree gutter shows parent/child structure, followed by the span's state, an optional progress bar, a timeline bar, and its duration. Click a row for its attributes and events; drag the replay scrubber (or press *Replay*) to watch the trace unfold.


Live, in-process (requires `opentelemetry-sdk`). Attach the processor once; after that every `with tracer.start_as_current_span(...)` block made through that provider shows up, so no widget-level context manager is needed::


```
view = TraceView(title="agent run")
provider.add_span_processor(view.span_processor())
view
```


From existing data (no OpenTelemetry install needed)::


```
TraceView.from_otlp("traces.json")              # OTLP/JSON or JSON lines
TraceView.from_spans(exporter.get_finished_spans())
TraceView.from_weave(client.get_calls(...))     # W&B Weave calls
```


W&B Weave `@weave.op` calls, live (after `weave.init(...)`). Weave has no processor hook, so this context manager plays the role of `add_span_processor` for the duration of the block::


```
with view.capture_weave():
    answer_ticket("Where is my order?")
```


A span with numeric `progress.done` and `progress.total` attributes gets a progress bar.


Picking an older trace from the menu pins it. With `show_latest=True` (the default, and the *Follow latest* button in the header) the next new trace takes the view back to following live; turning it on jumps there right away.

 Source code in `wigglystuff/trace_view.py`

```
def __init__(
    self,
    spans: list[dict[str, Any]] | None = None,
    *,
    title: str = "Trace",
    height: int = 0,
    sync_throttle_ms: int = 0,
    show_latest: bool = True,
    **kwargs: Any,
) -> None:
    if height < 0:
        raise ValueError("height must be >= 0")
    if sync_throttle_ms < 0:
        raise ValueError("sync_throttle_ms must be >= 0")
    spans = [_normalize_span(span) for span in spans or []]
    super().__init__(
        spans=spans,
        title=title,
        height=height,
        sync_throttle_ms=sync_throttle_ms,
        show_latest=show_latest,
        now_ms=time.time() * 1000,
        **kwargs,
    )
    self._lock = threading.Lock()
    self._by_id: dict[str, dict[str, Any]] = {s["span_id"]: s for s in spans}
    self._trace_ids = {s["trace_id"] for s in spans}
    # SDK spans that haven't ended yet; re-read on every sync so attribute
    # changes (like progress) made while they run show up.
    self._running: dict[str, Any] = {}
    self._last_sync = 0.0
```


## selected_span `property`


```
selected_span: dict[str, Any] | None
```


The span dict of the clicked row, or `None`.


## add_spans


```
add_spans(spans: Iterable[Any]) -> None
```


Add (or update, by `span_id`) OpenTelemetry SDK spans or span dicts.

 Source code in `wigglystuff/trace_view.py`

```
def add_spans(self, spans: Iterable[Any]) -> None:
    """Add (or update, by ``span_id``) OpenTelemetry SDK spans or span dicts."""
    with self._lock:
        for span in spans:
            data = _normalize_span(span) if isinstance(span, dict) else span_to_dict(span)
            self._by_id[data["span_id"]] = data
    self._sync()
```


## capture_weave


```
capture_weave() -> Iterator[None]
```


Show the W&B Weave `@weave.op` calls made inside this `with` block, live.


Needs `weave.init(...)` first; the calls still go to W&B as usual. Weave has no public call hook, so while the block runs this wraps the client's `create_call`/`finish_call`.

 Source code in `wigglystuff/trace_view.py`

```
@contextmanager
def capture_weave(self) -> Iterator[None]:
    """Show the W&B Weave ``@weave.op`` calls made inside this ``with`` block, live.

    Needs ``weave.init(...)`` first; the calls still go to W&B as usual.
    Weave has no public call hook, so while the block runs this wraps the
    client's ``create_call``/``finish_call``.
    """
    import weave

    client = weave.get_client()
    if client is None:
        raise RuntimeError("no Weave client: call weave.init(...) first")
    create_call, finish_call = client.create_call, client.finish_call

    def create_call_and_record(*args: Any, **kwargs: Any) -> Any:
        call = create_call(*args, **kwargs)
        self._record_call(call)
        return call

    def finish_call_and_record(call: Any, *args: Any, **kwargs: Any) -> Any:
        result = finish_call(call, *args, **kwargs)
        self._record_call(call)
        return result

    client.create_call, client.finish_call = create_call_and_record, finish_call_and_record
    try:
        yield
    finally:
        client.create_call, client.finish_call = create_call, finish_call
        self._sync()  # send any update the throttle held back
```


## clear


```
clear() -> None
```


Drop all spans and reset the selection, replay cursor, and collapsed rows.

 Source code in `wigglystuff/trace_view.py`

```
def clear(self) -> None:
    """Drop all spans and reset the selection, replay cursor, and collapsed rows."""
    with self._lock:
        self._by_id = {}
        self._running = {}
        self._trace_ids = set()
    self.spans = []
    self.trace_id = ""
    self.selected_span_id = None
    self.cursor_ms = None
    self.collapsed = []
```


## from_otlp `classmethod`


```
from_otlp(source: str | Path | dict[str, Any] | list[dict[str, Any]], **kwargs: Any) -> 'TraceView'
```


Load OTLP/JSON: a file path (a JSON document or JSON lines), a dict, or a list of dicts.

 Source code in `wigglystuff/trace_view.py`

```
@classmethod
def from_otlp(
    cls, source: str | Path | dict[str, Any] | list[dict[str, Any]], **kwargs: Any
) -> "TraceView":
    """Load OTLP/JSON: a file path (a JSON document or JSON lines), a dict, or a list of dicts."""
    if isinstance(source, (str, Path)):
        text = Path(source).read_text()
        try:
            docs = [json.loads(text)]
        except json.JSONDecodeError:  # the OTLP file exporter writes one doc per line
            docs = [json.loads(line) for line in text.splitlines() if line.strip()]
    elif isinstance(source, dict):
        docs = [source]
    else:
        docs = list(source)
    spans = [span for doc in docs for span in spans_from_otlp(doc)]
    return cls(spans, **kwargs)
```


## from_spans `classmethod`


```
from_spans(spans: Iterable[Any], **kwargs: Any) -> 'TraceView'
```


Build from OpenTelemetry SDK spans, e.g. `InMemorySpanExporter.get_finished_spans()`.

 Source code in `wigglystuff/trace_view.py`

```
@classmethod
def from_spans(cls, spans: Iterable[Any], **kwargs: Any) -> "TraceView":
    """Build from OpenTelemetry SDK spans, e.g. ``InMemorySpanExporter.get_finished_spans()``."""
    return cls([span_to_dict(span) for span in spans], **kwargs)
```


## from_weave `classmethod`


```
from_weave(calls: Iterable[Any], **kwargs: Any) -> 'TraceView'
```


Build from W&B Weave calls, e.g. `client.get_calls(filter={"trace_ids": [...]})`.


Accepts the SDK's call objects or plain dicts (service API JSON); no `weave` install needed for the dict form.

 Source code in `wigglystuff/trace_view.py`

```
@classmethod
def from_weave(cls, calls: Iterable[Any], **kwargs: Any) -> "TraceView":
    """Build from W&B Weave calls, e.g. ``client.get_calls(filter={"trace_ids": [...]})``.

    Accepts the SDK's call objects or plain dicts (service API JSON); no
    ``weave`` install needed for the dict form.
    """
    return cls([call_to_dict(call) for call in calls], **kwargs)
```


## span_processor


```
span_processor() -> Any
```


An OpenTelemetry `SpanProcessor` that streams spans into this widget.


Add it to a `TracerProvider` with `provider.add_span_processor(...)`. Updates are sent from the thread that starts or ends a span; spans that finish on background threads appear with the next update from the notebook's own thread.

 Source code in `wigglystuff/trace_view.py`

```
def span_processor(self) -> Any:
    """An OpenTelemetry ``SpanProcessor`` that streams spans into this widget.

    Add it to a ``TracerProvider`` with ``provider.add_span_processor(...)``.
    Updates are sent from the thread that starts or ends a span; spans that
    finish on background threads appear with the next update from the
    notebook's own thread.
    """
    from opentelemetry.sdk.trace import SpanProcessor

    view = self

    class TraceViewProcessor(SpanProcessor):
        def on_start(self, span: Any, parent_context: Any = None) -> None:
            view._record(span)

        def on_end(self, span: Any) -> None:
            view._record(span)

    return TraceViewProcessor()
```


## Synced traitlets


| Traitlet | Type | Notes |
| --- | --- | --- |
| `spans` | `list[dict]` | Normalized spans: `span_id`, `parent_id`, `trace_id`, `name`, `service`, `kind`, `start_ms`, `end_ms` (epoch ms, `None` while running), `status`, `status_message`, `attributes`, `events`, `links`. |
| `trace_id` | `str` | Trace on screen; `""` follows the most recently started one. |
| `show_latest` | `bool` | Follow latest button: a new trace clears a pinned `trace_id`; turning it on jumps to the latest trace. |
| `selected_span_id` | `str \\| None` | Set by clicking a row; read the span with `selected_span`. |
| `cursor_ms` | `float \\| None` | Replay position in ms since the trace started; `None` shows everything. |
| `collapsed` | `list[str]` | Span ids whose subtrees are folded. |
| `title` | `str` | Header title. |
| `height` | `int` | Max height of the rows in pixels; `0` grows to fit. |
| `now_ms` | `float` | Kernel clock at the last sync, so running bars grow correctly. |

**