feat: support Python OpenTelemetry span customizers

py/README.md

@@ -62,6 +62,78 @@ Available extras:
 - `temporal`: installs Temporal integration dependencies
 - `all`: installs all optional extras
 
+## Span export customizers
+
+Register synchronous hooks with `braintrust.set_span_customizers([...])` or
+`auto_instrument(span_customizers=[...])`. The same ordered hooks apply to native
+SDK spans and spans sent by `braintrust.otel.OtelExporter` or
+`BraintrustSpanProcessor`, including compatibility mode. No OTel dependency is
+needed for native tracing.
+
+```python
+from braintrust import SpanCustomizer, set_span_customizers
+
+class Redact(SpanCustomizer):
+    def on_span_export(self, data):
+        if "attributes" in data:  # OTel snapshot
+            data["attributes"].pop("braintrust.input_json", None)
+            data["attributes"].pop("braintrust.output_json", None)
+            data["events"] = []  # Events can also contain captured content.
+        else:  # Native, possibly incremental export record
+            data.pop("input", None)
+            data.pop("output", None)
+        return data
+
+set_span_customizers([Redact()])
+```
+
+Native hooks receive the existing mutable record (`input`, `output`, `metadata`,
+`metrics`, etc.), possibly containing SDK attachment objects. OTel hooks receive
+a detached plain dictionary, not a live `ReadableSpan` or a native conversion:
+
+| Field | OTel snapshot value |
+| --- | --- |
+| `name` | Span name string |
+| `context`, `parent` | Context dictionary; `parent` is `None` for roots |
+| `start_time`, `end_time` | Original Unix nanoseconds |
+| `kind` | Integer `SpanKind.value` (0–4) |
+| `attributes` | Dictionary of OTel scalar values or homogeneous lists |
+| `events` | List of dictionaries with `name`, `timestamp` (nanoseconds), `attributes`, `dropped_attributes` |
+| `links` | List of dictionaries with `context`, `attributes`, `dropped_attributes` |
+| `status` | Dictionary with integer `status_code` (`StatusCode.value`, 0–2) and `description` (string or `None`) |
+| `resource` | Dictionary with `attributes` and `schema_url` |
+| `instrumentation_scope` | Dictionary with `name`, `version`, `schema_url`, `attributes`, or `None` |
+| `dropped_attributes`, `dropped_events`, `dropped_links` | Integer dropped-item counts |
+
+Each context dictionary contains integer `trace_id`, `span_id`, `trace_flags`,
+boolean `is_remote`, and a `trace_state` string-to-string dictionary.
+Scope attributes require an OTel SDK version that supports them; older versions
+expose an empty dictionary. Serialization follows the installed OTel version.
+
+Hooks run after AI filtering and SDK-origin enrichment, before serialization or
+attachment upload. They affect every span reaching the Braintrust export path,
+not other exporters consuming the same OTel spans. Processor/exporter composition
+does not run hooks twice. OTel customization uses the registration active at
+export time. Transport retries reuse serialized data; a new explicit `export`
+call runs hooks again. Native records retain their logging-time registration.
+
+Return the input dictionary or a replacement; replacements are **not merged**.
+Removed fields stay removed (required transport fields use empty defaults).
+Original OTel context, parent, timing, and `attributes["braintrust.parent"]`
+routing are restored after each hook. Nested containers are detached from the
+live span. Native identity/routing/merge protection is unchanged.
+
+A raised exception or invalid return stops the chain for that record and strips
+its content, while unrelated spans still export. Only original identity,
+timing, routing, and the error marker survive. OTel's sole content marker is
+`braintrust.context_json` containing exactly `{"customizer_error":true}`;
+native records use `context={"customizer_error": True}`. Resource/scope metadata,
+events, links, status descriptions, and partial hook results are not retained.
+Invalid OTel serialization shapes also fail closed. One safe diagnostic is
+logged per process across native and OTel failures, without exception text or
+stacks. `KeyboardInterrupt` and `SystemExit` still propagate. Hooks must be fast
+and synchronous; returning an awaitable or `None` is an error.
+
 ## Documentation
 
 - Python SDK docs: https://www.braintrust.dev/docs/reference/sdks/python

py/src/braintrust/auto.py

@@ -103,12 +103,10 @@ def auto_instrument(
         livekit_agents: Enable LiveKit Agents instrumentation (default: True)
         pipecat: Enable Pipecat AI instrumentation (default: True)
         typesafe: Enable TypeSafe instrumentation (default: True)
-        span_customizers: Ordered synchronous export customizers for all native SDK
-            spans, including manual spans; dataset rows and feedback are excluded.
+        span_customizers: Ordered synchronous export customizers for native SDK and
+            Braintrust OTel spans, including manual spans; dataset rows and feedback are excluded.
             Copies and replaces the global list when provided; None leaves existing
             configuration unchanged. Pass [] to disable.
-            With BRAINTRUST_OTEL_COMPAT enabled, non-empty customizers log an error
-            and leave the existing configuration unchanged; instrumentation continues.
 
     Returns:
         Dict mapping integration name to whether it was successfully instrumented.

py/src/braintrust/logger.py

@@ -102,7 +102,13 @@
 )
 from .queue import DEFAULT_QUEUE_SIZE, LogQueue
 from .serializable_data_class import SerializableDataClass
-from .span_customizer import SpanCustomizer, _customize_span_export, _get_span_customizers, _MaskingCustomizer
+from .span_customizer import (
+    SpanCustomizer,
+    _customize_span_export,
+    _get_span_customizers,
+    _MaskingCustomizer,
+    _StrippedSpanExportData,
+)
 from .span_identifier_v3 import SpanComponentsV3, SpanObjectTypeV3
 from .span_identifier_v4 import SpanComponentsV4
 from .span_origin import SpanOriginEnvironment, detect_environment, merge_span_origin_context
@@ -996,7 +1002,7 @@ def utf8_byte_length(value: str) -> int:
     return len(value) if value.isascii() else len(value.encode("utf-8"))
 
 
-_LazyLogRecord = LazyValue[dict[str, Any]] | LazyValue[dict[str, Any] | None]
+_LazyLogRecord = LazyValue[dict[str, Any]]
 
 
 class _BackgroundLogger(ABC):
@@ -1033,8 +1039,7 @@ def flush(self, batch_size: int | None = None):
             if not self.logs:
                 return
 
-            # Filter dropped records before touching attachments.
-            logs = [record for item in self.logs if (record := item.get()) is not None]
+            logs = [item.get() for item in self.logs]
 
             # Extract attachments from all logs
             attachments: list[BaseAttachment] = []
@@ -1046,7 +1051,7 @@ def flush(self, batch_size: int | None = None):
 
     def pop(self) -> list[dict[str, Any]]:
         with self.lock:
-            logs = [record for item in self.logs if (record := item.get()) is not None]
+            logs = [item.get() for item in self.logs]
             self.logs = []
 
             if not logs:
@@ -1057,11 +1062,7 @@ def pop(self) -> list[dict[str, Any]]:
             batch = merge_row_batch(logs)
 
             if self._export_customizers:
-                batch = [
-                    record
-                    for item in batch
-                    if (record := _customize_span_export(item, self._export_customizers)) is not None
-                ]
+                batch = [_customize_span_export(item, self._export_customizers) for item in batch]
 
             return batch
 
@@ -1210,24 +1211,45 @@ def flush(self, batch_size: int | None = None):
             if len(all_items) == 0:
                 return
 
-            # Construct batches of records to flush in parallel.
-            all_items_with_meta = [stringify_with_overflow_meta(item) for item in all_items]
+            # Failed records are merge barriers and must also be request barriers:
+            # duplicate row IDs in one request cannot preserve update ordering.
+            groups: list[list[dict[str, Any]]] = []
+            group: list[dict[str, Any]] = []
+            for item in all_items:
+                if isinstance(item, _StrippedSpanExportData):
+                    if group:
+                        groups.append(group)
+                        group = []
+                    groups.append([item])
+                else:
+                    group.append(item)
+            if group:
+                groups.append(group)
             max_request_size_result = self._get_max_request_size()
-            batches = batch_items(
-                items=all_items_with_meta,
-                batch_max_num_items=batch_size,
-                batch_max_num_bytes=max(
-                    0, max_request_size_result["max_request_size"] - utf8_byte_length(construct_logs3_data([])) + 1
-                ),
-                get_byte_size=lambda item: item.overflow_meta.byte_size + 1,
-            )
+            batches = [
+                batch
+                for group in groups
+                for batch in batch_items(
+                    items=[stringify_with_overflow_meta(item) for item in group],
+                    batch_max_num_items=batch_size,
+                    batch_max_num_bytes=max(
+                        0, max_request_size_result["max_request_size"] - utf8_byte_length(construct_logs3_data([])) + 1
+                    ),
+                    get_byte_size=lambda item: item.overflow_meta.byte_size + 1,
+                )
+            ]
 
             post_promises = []
             try:
-                post_promises = [
-                    HTTP_REQUEST_THREAD_POOL.submit(self._submit_logs_request, batch, max_request_size_result)
-                    for batch in batches
-                ]
+                if any(isinstance(item, _StrippedSpanExportData) for item in all_items):
+                    # Preserve record order across barriers, including retries.
+                    for batch in batches:
+                        self._submit_logs_request(batch, max_request_size_result)
+                else:
+                    post_promises = [
+                        HTTP_REQUEST_THREAD_POOL.submit(self._submit_logs_request, batch, max_request_size_result)
+                        for batch in batches
+                    ]
             except RuntimeError:
                 # If the thread pool has shut down, e.g. because the process
                 # is terminating, run the requests the old fashioned way.
@@ -1264,16 +1286,12 @@ def _unwrap_lazy_values(
     ) -> tuple[list[dict[str, Any]], list["BaseAttachment"]]:
         for i in range(self.num_tries):
             try:
-                unwrapped_items = [record for item in wrapped_items if (record := item.get()) is not None]
+                unwrapped_items = [item.get() for item in wrapped_items]
                 merged_items = merge_row_batch(unwrapped_items)
 
                 # Logger-local hooks run after span hooks and merging.
                 if self._export_customizers:
-                    merged_items = [
-                        record
-                        for item in merged_items
-                        if (record := _customize_span_export(item, self._export_customizers)) is not None
-                    ]
+                    merged_items = [_customize_span_export(item, self._export_customizers) for item in merged_items]
 
                 attachments: list["BaseAttachment"] = []
                 for item in merged_items:
@@ -4034,7 +4052,7 @@ def parent_ids():
             object_id=parent_object_id.get(),
         ).object_id_fields()
 
-    def compute_record() -> dict[str, Any] | None:
+    def compute_record() -> dict[str, Any]:
         record = dict(
             id=id,
             **update_event,
@@ -4978,11 +4996,11 @@ def write_span_cache(record: dict[str, Any]) -> None:
             self.state.span_cache.queue_write(self.root_span_id, self.span_id, cached_span)
 
         # Customized records are cached after export customization instead, so
-        # local scorers never see content that a customizer redacted or dropped.
+        # local scorers never see content that a customizer redacted or stripped.
         if not customizers:
             write_span_cache(serializable_partial_record)
 
-        def compute_record() -> dict[str, Any] | None:
+        def compute_record() -> dict[str, Any]:
             exporter = _get_exporter()
             record = dict(
                 **serializable_partial_record,
@@ -4997,8 +5015,7 @@ def compute_record() -> dict[str, Any] | None:
             # logger merges, masks, extracts attachments, or retries delivery.
             if customizers:
                 customized_record = _customize_span_export(record, customizers)
-                if customized_record is not None:
-                    write_span_cache(customized_record)
+                write_span_cache(customized_record)
                 if pending_cache_key is not None:
                     self.state.span_cache._forget_pending_record(self.root_span_id, pending_cache_key)
                 return customized_record

py/src/braintrust/merge_row_batch.py

@@ -2,6 +2,7 @@
 from typing import Any, TypeVar
 
 from .db_fields import IS_MERGE_FIELD
+from .span_customizer import _StrippedSpanExportData
 
 
 T = TypeVar("T")
@@ -58,6 +59,9 @@ def merge_row_batch(rows: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
 
     Note that the returned rows will be the same objects as the input `rows`,
     meaning they are mutated in place.
+    SDK-stripped customizer failures are exceptions: they are not mutated and
+    divide the result into independent merge groups. The exporter must send these
+    groups in ordered, separate requests to preserve same-ID update ordering.
 
     There are a few important considerations for the merge procedure:
 
@@ -100,7 +104,15 @@ def merge_row_batch(rows: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
             )
 
     row_groups: dict[_MergedRowKey, dict[str, Any]] = {}
+    merged_rows = []
     for row in rows:
+        if isinstance(row, _StrippedSpanExportData):
+            # Never mutate a memoized failure with neighboring record content.
+            # These barriers also become ordered HTTP request boundaries.
+            merged_rows.extend(row_groups.values())
+            merged_rows.append(row)
+            row_groups.clear()
+            continue
         key = _generate_merged_row_key(row)
         existing_row = row_groups.get(key)
         # If there is an existing row and the new row has the IS_MERGE_FIELD ==
@@ -118,7 +130,8 @@ def merge_row_batch(rows: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
         else:
             row_groups[key] = row
 
-    return list(row_groups.values())
+    merged_rows.extend(row_groups.values())
+    return merged_rows
 
 
 def batch_items(

py/src/braintrust/otel/__init__.py

@@ -8,6 +8,7 @@
 from urllib.parse import urljoin
 
 from braintrust.env import BraintrustEnv
+from braintrust.span_customizer import _get_span_customizers
 from braintrust.span_origin import SpanOriginEnvironment, detect_environment, merge_span_origin_context
 
 
@@ -303,6 +304,21 @@ def initialize(self) -> None:
 
     def export(self, spans):
         self._ensure_api_key()
+        customizers = _get_span_customizers()
+        if customizers:
+            from .span_customizer import customize_otel_span
+
+            # Processor exports are already enriched after AI filtering. Direct
+            # exporter users receive the same provenance before customization.
+            spans = [
+                customize_otel_span(
+                    span
+                    if isinstance(span, _SpanWithAttributes)
+                    else _with_span_origin_attributes(span, detect_environment()),
+                    customizers,
+                )
+                for span in spans
+            ]
         return super().export(spans)
 
     def force_flush(self, timeout_millis=30000):