feat(logger): add new logging apis

py/src/braintrust/__init__.py

@@ -81,6 +81,7 @@ def is_equal(expected, output):
     _internal_reset_global_state,  # noqa: F401 # type: ignore[reportUnusedImport]
     _internal_with_custom_background_logger,  # noqa: F401 # type: ignore[reportUnusedImport]
 )
+from .logs import BraintrustLogHandler as BraintrustLogHandler
 from .sandbox import RegisteredSandboxFunction as RegisteredSandboxFunction
 from .sandbox import RegisterSandboxResult as RegisterSandboxResult
 from .sandbox import SandboxConfig as SandboxConfig

py/src/braintrust/logger.py

@@ -8,6 +8,7 @@
 import dataclasses
 import datetime
 import hashlib
+import importlib
 import inspect
 import io
 import json
@@ -131,6 +132,58 @@
 # 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"]
+_LOG_LEVELS: tuple[LogLevel, ...] = ("trace", "debug", "info", "warn", "error", "fatal")
+
+_TEMPLATELIB = importlib.import_module("string.templatelib") if sys.version_info >= (3, 14) else None
+
+
+class _LogTemplateParameters(dict[str, object]):
+    """Preserve placeholders whose values were not provided."""
+
+    def __missing__(self, key: str) -> str:
+        return "{" + key + "}"
+
+
+def _is_t_string(value: Any) -> bool:
+    return _TEMPLATELIB is not None and isinstance(value, _TEMPLATELIB.Template)
+
+
+def _render_t_string(template: Any) -> tuple[str, str, dict[str, object]]:
+    """Render a Python 3.14 t-string and retain its template structure."""
+    assert _TEMPLATELIB is not None
+
+    rendered_parts: list[str] = []
+    template_parts: list[str] = []
+    parameters: dict[str, object] = {}
+
+    for index, (literal, interpolation) in enumerate(zip(template.strings, template.interpolations)):
+        rendered_parts.append(literal)
+        template_parts.append(literal.replace("{", "{{").replace("}", "}}"))
+
+        placeholder = "{" + interpolation.expression
+        if interpolation.conversion is not None:
+            placeholder += "!" + interpolation.conversion
+        if interpolation.format_spec:
+            placeholder += ":" + interpolation.format_spec
+        placeholder += "}"
+        template_parts.append(placeholder)
+
+        parameter_name = interpolation.expression.strip() or str(index)
+        parameters[parameter_name] = interpolation.value
+        try:
+            converted = _TEMPLATELIB.convert(interpolation.value, interpolation.conversion)
+            rendered_parts.append(format(converted, interpolation.format_spec))
+        except Exception:
+            # Logging should not disrupt the application because an interpolation
+            # uses an unsupported conversion or format specifier.
+            rendered_parts.append(placeholder)
+
+    final_literal = template.strings[-1]
+    rendered_parts.append(final_literal)
+    template_parts.append(final_literal.replace("{", "{{").replace("}", "}}"))
+    return "".join(rendered_parts), "".join(template_parts), parameters
+
 
 @dataclasses.dataclass
 class Logs3OverflowInputRow:
@@ -5896,6 +5949,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:
@@ -5974,6 +6028,122 @@ def log(
 
         return span.id
 
+    def emit_log(
+        self,
+        body: Any,
+        level: LogLevel,
+        metadata: dict[str, Any] | None = None,
+        **parameters: object,
+    ) -> 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.
+
+        String bodies may contain ``str.format``-style placeholders. Keyword
+        parameters are interpolated into the body and retained in metadata along
+        with the original template. Missing parameters remain as placeholders.
+        On Python 3.14 and newer, ``string.templatelib.Template`` bodies are
+        rendered using their embedded interpolation values, which are also
+        retained in metadata.
+
+        :param body: The log body. May be a Python 3.14+ t-string or any
+            JSON-serializable value when no template parameters are provided.
+        :param level: The OpenTelemetry log severity: ``trace``, ``debug``,
+            ``info``, ``warn``, ``error``, or ``fatal``.
+        :param metadata: Optional JSON-serializable attributes for the log.
+        :param parameters: Values for named placeholders in a string body.
+        :returns: The unique ID of the captured log row.
+        """
+        rendered_body = body
+        rendered_metadata = metadata
+        if _is_t_string(body):
+            if parameters:
+                raise TypeError("T-string bodies already contain their interpolation values")
+            rendered_body, template, t_string_parameters = _render_t_string(body)
+            rendered_metadata = dict(metadata) if metadata is not None else {}
+            rendered_metadata.update(
+                {f"braintrust.template.parameter.{key}": value for key, value in t_string_parameters.items()}
+            )
+            rendered_metadata["braintrust.template"] = template
+        elif parameters:
+            if not isinstance(body, str):
+                raise TypeError("Log body must be a string when template parameters are provided")
+            rendered_metadata = dict(metadata) if metadata is not None else {}
+            rendered_metadata.update(
+                {f"braintrust.template.parameter.{key}": value for key, value in parameters.items()}
+            )
+            rendered_metadata["braintrust.template"] = body
+            try:
+                rendered_body = body.format_map(_LogTemplateParameters(parameters))
+            except Exception:
+                # Logging should not disrupt the application because a template
+                # contains malformed braces or an unsupported format specifier.
+                rendered_body = body
+
+        return self._emit_log_record(
+            body=rendered_body,
+            level=level,
+            metadata=rendered_metadata,
+            captured_at=time.time(),
+        )
+
+    def _emit_log_record(
+        self,
+        body: Any,
+        level: LogLevel,
+        metadata: dict[str, Any] | None,
+        captured_at: float,
+    ) -> str:
+        if level not in _LOG_LEVELS:
+            valid_levels = ", ".join(_LOG_LEVELS)
+            raise ValueError(f"Invalid log level {level!r}. Expected one of: {valid_levels}")
+
+        span_info = self.state.context_manager.get_current_span_info()
+        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,
+            metadata={**(metadata or {}), "braintrust.log_level": level},
+            metrics={"end": captured_at},
+            created=datetime.datetime.fromtimestamp(captured_at, datetime.timezone.utc).isoformat(),
+        )
+
+        if not self.async_flush:
+            self.flush()
+
+        return span.id
+
+    def trace(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
+        """Capture a trace-level log."""
+        return self.emit_log(body=body, level="trace", metadata=metadata, **parameters)
+
+    def debug(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
+        """Capture a debug-level log."""
+        return self.emit_log(body=body, level="debug", metadata=metadata, **parameters)
+
+    def info(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
+        """Capture an info-level log."""
+        return self.emit_log(body=body, level="info", metadata=metadata, **parameters)
+
+    def warn(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
+        """Capture a warn-level log."""
+        return self.emit_log(body=body, level="warn", metadata=metadata, **parameters)
+
+    def error(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
+        """Capture an error-level log."""
+        return self.emit_log(body=body, level="error", metadata=metadata, **parameters)
+
+    def fatal(self, body: Any, metadata: dict[str, Any] | None = None, **parameters: object) -> str:
+        """Capture a fatal-level log."""
+        return self.emit_log(body=body, level="fatal", metadata=metadata, **parameters)
+
     def log_feedback(
         self,
         id: str,

py/src/braintrust/logs.py

@@ -0,0 +1,88 @@
+"""Adapters for forwarding standard-library log records to Braintrust."""
+
+import logging
+from typing import Any
+
+from .logger import Logger, LogLevel
+
+
+_STANDARD_LOG_RECORD_ATTRIBUTES = frozenset(vars(logging.LogRecord("", logging.NOTSET, "", 0, "", (), None))) | {
+    "asctime",
+    "message",
+}
+_IGNORED_LOGGER_PREFIXES = ("braintrust", "urllib3")
+
+
+def _log_level(level: int) -> LogLevel:
+    if level >= logging.CRITICAL:
+        return "fatal"
+    if level >= logging.ERROR:
+        return "error"
+    if level >= logging.WARNING:
+        return "warn"
+    if level >= logging.INFO:
+        return "info"
+    if level >= logging.DEBUG:
+        return "debug"
+    return "trace"
+
+
+def _is_ignored_logger(name: str) -> bool:
+    return any(name == prefix or name.startswith(f"{prefix}.") for prefix in _IGNORED_LOGGER_PREFIXES)
+
+
+def _record_metadata(record: logging.LogRecord) -> dict[str, Any]:
+    metadata = {
+        key: value
+        for key, value in vars(record).items()
+        if key not in _STANDARD_LOG_RECORD_ATTRIBUTES and not key.startswith("_")
+    }
+
+    if record.args and isinstance(record.msg, str):
+        metadata["braintrust.template"] = record.msg
+        parameters = record.args.items() if isinstance(record.args, dict) else enumerate(record.args)
+        metadata.update({f"braintrust.template.parameter.{key}": value for key, value in parameters})
+
+    metadata.update(
+        {
+            "logger.name": record.name,
+            "code.file.path": record.pathname,
+            "code.function.name": record.funcName,
+            "code.line.number": record.lineno,
+        }
+    )
+
+    return metadata
+
+
+class BraintrustLogHandler(logging.Handler):
+    """Forward Python ``logging`` records to a Braintrust logger.
+
+    Attach this handler explicitly with ``logging.Logger.addHandler``. Records
+    emitted by Braintrust and its HTTP transport are ignored to prevent logging
+    recursion.
+    """
+
+    def __init__(self, logger: Logger, level: int | str = logging.NOTSET):
+        super().__init__(level=level)
+        self._logger = logger
+
+    def emit(self, record: logging.LogRecord) -> None:
+        if _is_ignored_logger(record.name):
+            return
+
+        try:
+            self._logger._emit_log_record(
+                body=self.format(record),
+                level=_log_level(record.levelno),
+                metadata=_record_metadata(record),
+                captured_at=record.created,
+            )
+        except Exception:
+            self.handleError(record)
+
+    def flush(self) -> None:
+        self._logger.flush()
+
+
+__all__ = ["BraintrustLogHandler"]

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

@@ -19,6 +19,7 @@ class SpanTypeAttribute(str, Enum):
     CLASSIFIER = "classifier"
     REVIEW = "review"
     QUESTION = "question"
+    LOG = "log"
 
 
 class SpanPurpose(str, Enum):