spans_to_trace

def spans_to_trace(
    spans: Sequence[SpanData],
    *,
    conventions: Sequence[SpanConvention] | None = None,
    duration_ms: int | None = None,
) -> Trace:
    """Reconstruct a :class:`Trace` from captured spans.

    Spans are ordered by start time; each is mapped to typed events by the first
    ``conventions`` entry that recognizes it (default: AG2 ``gen_ai`` + OpenInference,
    auto-detected per span). ``duration_ms`` defaults to the root span's wall-clock;
    pass an explicit value to override (e.g. the producer's measured ``ask`` duration).
    """
    ordered = sorted(spans, key=lambda s: s.start_ns)
    active = DEFAULT_CONVENTIONS if conventions is None else conventions

    events: list[BaseEvent] = []
    for span in ordered:
        for convention in active:
            mapped = convention.to_events(span)
            if mapped is not None:
                events.extend(mapped)
                break

    if ordered and not events:
        logger.warning(
            "spans_to_trace reconstructed 0 events from %d span(s) — the producer's span dialect may be "
            "unrecognized by %s.",
            len(ordered),
            "/".join(type(c).__name__ for c in active) or "(no conventions)",
        )

    resolved_duration = duration_ms if duration_ms is not None else _root_duration_ms(ordered)
    return Trace(events=events, exception=_root_exception(ordered), duration_ms=resolved_duration)