feat(logger): add OpenTelemetry-compatible log emission

Add `Logger.emit_log()` so applications can emit independent `type="log"`
rows without constructing spans manually. Correlate rows with the active
Braintrust or OpenTelemetry context when available.

Seed each logger with a baseline trace ID for unscoped logs. This keeps
consecutive logs from one logger together while preserving unique row and span
IDs, and avoids grouping logs emitted by separate logger instances.

Map the six base OpenTelemetry severities into `context.otel.log` and add
`trace()`, `debug()`, `info()`, `warn()`, `error()`, and `fatal()` helpers.

    logger.error("Payment failed", metadata={"payment_id": "pay_123"})

    helper -> emit_log -> `type="log"` row
                         |-- active span: reuse trace/span IDs
                         `-- no span: reuse logger trace, generate span ID

Author: AbhiPrasad

py/src/braintrust/logger.py

@@ -131,6 +131,16 @@
 # 6 MB for the AWS lambda gateway (from our own testing).
 DEFAULT_MAX_REQUEST_SIZE = 6 * 1024 * 1024
 
+LogLevel = Literal["trace", "debug", "info", "warn", "error", "fatal"]
+_OTEL_LOG_LEVELS: dict[LogLevel, int] = {
+    "trace": 1,
+    "debug": 5,
+    "info": 9,
+    "warn": 13,
+    "error": 17,
+    "fatal": 21,
+}
+
 
 @dataclasses.dataclass
 class Logs3OverflowInputRow:
@@ -5854,6 +5864,7 @@ def __init__(
         # fallbacks when generating links
         self._link_args = link_args
         self.state = state or _state
+        self._baseline_trace_id = self.state.id_generator.get_trace_id()
 
     @property
     def org_id(self) -> str:
@@ -5932,6 +5943,84 @@ def log(
 
         return span.id
 
+    def emit_log(
+        self,
+        body: Any,
+        level: LogLevel,
+        metadata: Metadata | None = None,
+    ) -> str:
+        """Capture a log record, associating it with the active span when one exists.
+
+        The log is stored as an independent row. If a Braintrust or OpenTelemetry
+        span is active, the row reuses its span and trace IDs for correlation.
+        Otherwise, the row uses this logger's baseline trace ID.
+
+        :param body: The log body. May be any JSON-serializable value.
+        :param level: The OpenTelemetry log severity: ``trace``, ``debug``,
+            ``info``, ``warn``, ``error``, or ``fatal``.
+        :param metadata: Optional JSON-serializable attributes for the log.
+        :returns: The unique ID of the captured log row.
+        """
+        if level not in _OTEL_LOG_LEVELS:
+            valid_levels = ", ".join(_OTEL_LOG_LEVELS)
+            raise ValueError(f"Invalid log level {level!r}. Expected one of: {valid_levels}")
+
+        captured_at = time.time()
+        span_info = self.state.context_manager.get_current_span_info()
+        severity_number = _OTEL_LOG_LEVELS[level]
+        span = self._start_span_impl(
+            name="Log",
+            type=SpanTypeAttribute.LOG,
+            start_time=captured_at,
+            set_current=False,
+            span_id=span_info.span_id if span_info else None,
+            root_span_id=span_info.trace_id if span_info else self._baseline_trace_id,
+            lookup_span_parent=False,
+            output=body,
+            error=body if severity_number >= _OTEL_LOG_LEVELS["error"] and isinstance(body, str) else None,
+            metadata=metadata,
+            context={
+                "otel": {
+                    "signal": "logs",
+                    "log": {
+                        "time_unix_nano": str(round(captured_at * 1_000_000_000)),
+                        "severity_number": severity_number,
+                        "severity_text": level.upper(),
+                    },
+                }
+            },
+        )
+        span.end(end_time=captured_at)
+
+        if not self.async_flush:
+            self.flush()
+
+        return span.id
+
+    def trace(self, body: Any, metadata: Metadata | None = None) -> str:
+        """Capture a log at OpenTelemetry TRACE severity."""
+        return self.emit_log(body=body, level="trace", metadata=metadata)
+
+    def debug(self, body: Any, metadata: Metadata | None = None) -> str:
+        """Capture a log at OpenTelemetry DEBUG severity."""
+        return self.emit_log(body=body, level="debug", metadata=metadata)
+
+    def info(self, body: Any, metadata: Metadata | None = None) -> str:
+        """Capture a log at OpenTelemetry INFO severity."""
+        return self.emit_log(body=body, level="info", metadata=metadata)
+
+    def warn(self, body: Any, metadata: Metadata | None = None) -> str:
+        """Capture a log at OpenTelemetry WARN severity."""
+        return self.emit_log(body=body, level="warn", metadata=metadata)
+
+    def error(self, body: Any, metadata: Metadata | None = None) -> str:
+        """Capture a log at OpenTelemetry ERROR severity."""
+        return self.emit_log(body=body, level="error", metadata=metadata)
+
+    def fatal(self, body: Any, metadata: Metadata | None = None) -> str:
+        """Capture a log at OpenTelemetry FATAL severity."""
+        return self.emit_log(body=body, level="fatal", metadata=metadata)
+
     def log_feedback(
         self,
         id: str,

py/src/braintrust/otel/test_otel_bt_integration.py

@@ -122,6 +122,22 @@ def test_mixed_otel_bt_tracing_with_bt_logger_first(otel_fixture):
     assert s2_span_id in s3["span_parents"]
 
 
+def test_emit_log_uses_active_otel_span(otel_fixture):
+    logger = init_test_logger(__name__)
+    tracer = otel_fixture.tracer
+    memory_logger = otel_fixture.memory_logger
+
+    with tracer.start_as_current_span("owner") as owner:
+        log_id = logger.emit_log(body="Inside OTel span", level="info")
+        owner_context = owner.get_span_context()
+
+    [log_row] = memory_logger.pop()
+    assert log_row["id"] == log_id
+    assert log_row["span_id"] == format(owner_context.span_id, "016x")
+    assert log_row["root_span_id"] == format(owner_context.trace_id, "032x")
+    assert not log_row.get("span_parents")
+
+
 def test_mixed_otel_bt_tracing_with_experiment_parent(otel_fixture):
     experiment = init_test_exp("otel-bt-mixed", "test-mixed-tracing-experiment")
     tracer = otel_fixture.tracer

py/src/braintrust/span_types.py

@@ -18,6 +18,7 @@ class SpanTypeAttribute(str, Enum):
     PREPROCESSOR = "preprocessor"
     CLASSIFIER = "classifier"
     REVIEW = "review"
+    LOG = "log"
 
 
 class SpanPurpose(str, Enum):

py/src/braintrust/test_logger.py

@@ -1396,6 +1396,107 @@ def test_logger_log_accepts_model_dump_metadata(with_memory_logger):
     assert logs[0]["metadata"] == {"foo": "bar"}
 
 
+def test_logger_emit_log_without_active_span(with_memory_logger):
+    test_logger = init_test_logger(__name__)
+
+    first_id = test_logger.emit_log(
+        body="Payment failed",
+        level="error",
+        metadata={"payment_id": "pay_123"},
+    )
+    second_id = test_logger.emit_log(body="Retrying payment", level="info")
+
+    logs = with_memory_logger.pop()
+    assert len(logs) == 2
+    first, second = logs
+    assert first_id == first["id"]
+    assert second_id == second["id"]
+    assert first["id"] != second["id"]
+    assert first["span_id"] != second["span_id"]
+    assert first["root_span_id"] == second["root_span_id"]
+    assert not first.get("span_parents")
+    assert first["output"] == "Payment failed"
+    assert first["error"] == "Payment failed"
+    assert first["metadata"] == {"payment_id": "pay_123"}
+    assert first["span_attributes"]["name"] == "Log"
+    assert first["span_attributes"]["type"] == "log"
+    assert first["metrics"]["start"] == first["metrics"]["end"]
+    assert first["context"]["otel"]["signal"] == "logs"
+    assert first["context"]["otel"]["log"] == {
+        "time_unix_nano": str(round(first["metrics"]["start"] * 1_000_000_000)),
+        "severity_number": 17,
+        "severity_text": "ERROR",
+    }
+    assert "error" not in second
+    assert second["context"]["otel"]["log"]["severity_number"] == 9
+
+
+def test_logger_emit_log_uses_distinct_baseline_trace_per_logger(with_memory_logger):
+    first_logger = init_test_logger(f"{__name__}-first")
+    second_logger = init_test_logger(f"{__name__}-second")
+
+    first_logger.info("first")
+    second_logger.info("second")
+
+    first, second = with_memory_logger.pop()
+    assert first["root_span_id"] != second["root_span_id"]
+
+
+def test_logger_emit_log_uses_active_span(with_memory_logger):
+    test_logger = init_test_logger(__name__)
+
+    with test_logger.start_span(name="owner") as owner:
+        log_id = test_logger.emit_log(body="Inside span", level="debug", metadata={"attempt": 1})
+
+    rows = with_memory_logger.pop()
+    log_row = next(row for row in rows if row["id"] == log_id)
+    owner_row = next(row for row in rows if row["span_attributes"]["name"] == "owner")
+    assert log_row["id"] != owner_row["id"]
+    assert log_row["span_id"] == owner_row["span_id"]
+    assert log_row["root_span_id"] == owner_row["root_span_id"]
+    assert not log_row.get("span_parents")
+    assert log_row["context"]["otel"]["log"]["severity_number"] == 5
+
+
+@pytest.mark.parametrize(
+    ("level", "severity_number"),
+    [("trace", 1), ("debug", 5), ("info", 9), ("warn", 13), ("error", 17), ("fatal", 21)],
+)
+def test_logger_emit_log_maps_otel_log_levels(with_memory_logger, level, severity_number):
+    test_logger = init_test_logger(__name__)
+
+    test_logger.emit_log(body="message", level=level)
+
+    [row] = with_memory_logger.pop()
+    assert row["context"]["otel"]["log"]["severity_number"] == severity_number
+    assert row["context"]["otel"]["log"]["severity_text"] == level.upper()
+
+
+@pytest.mark.parametrize(
+    ("method_name", "severity_number"),
+    [("trace", 1), ("debug", 5), ("info", 9), ("warn", 13), ("error", 17), ("fatal", 21)],
+)
+def test_logger_log_level_helpers(with_memory_logger, method_name, severity_number):
+    test_logger = init_test_logger(__name__)
+
+    log_id = getattr(test_logger, method_name)("message", metadata={"source": method_name})
+
+    [row] = with_memory_logger.pop()
+    assert row["id"] == log_id
+    assert row["output"] == "message"
+    assert row["metadata"] == {"source": method_name}
+    assert row["context"]["otel"]["log"]["severity_number"] == severity_number
+
+
+def test_logger_emit_log_rejects_invalid_level(with_memory_logger):
+    test_logger = init_test_logger(__name__)
+
+    with pytest.raises(ValueError, match="Invalid log level"):
+        test_logger.emit_log(body="message", level="warning")
+
+    assert with_memory_logger.pop() == []
+
+
 def test_experiment_log_accepts_model_dump_metadata(with_memory_logger):
     experiment = init_test_exp("test-experiment", "test-project")
 

py/src/braintrust/type_tests/test_metadata_types.py

@@ -23,6 +23,16 @@ def accepts_logger_metadata(logger: Logger) -> None:
     logger.log(metadata=PydanticV2Metadata())
     logger.log(metadata=PydanticV1Metadata())
 
+    logger.emit_log(body="message", level="info", metadata=mapping_metadata)
+    logger.emit_log(body="message", level="info", metadata=PydanticV2Metadata())
+    logger.emit_log(body="message", level="info", metadata=PydanticV1Metadata())
+    logger.trace("message", metadata=mapping_metadata)
+    logger.debug("message", metadata=PydanticV2Metadata())
+    logger.info("message", metadata=PydanticV1Metadata())
+    logger.warn("message", metadata=mapping_metadata)
+    logger.error("message", metadata=PydanticV2Metadata())
+    logger.fatal("message", metadata=PydanticV1Metadata())
+
     logger.log_feedback(id="event-id", metadata=mapping_metadata)
     logger.log_feedback(id="event-id", metadata=PydanticV2Metadata())
     logger.log_feedback(id="event-id", metadata=PydanticV1Metadata())