honeyhive.tracer.processing

Processing pipeline for HoneyHive tracer spans and data transformation.

This module provides the complete data processing pipeline from span creation to export, including context management, span processing, and OTLP export. All components use dynamic logic patterns for flexible, extensible processing.

HoneyHiveOTLPExporter

Bases: SpanExporter

HoneyHive OTLP exporter with optimized connection pooling.

This exporter is an enhanced wrapper around the standard OpenTelemetry OTLP exporter that includes optimized HTTP session with connection pooling for better performance and reliability. All span processing should have been completed by the HoneyHiveSpanProcessor before spans reach this exporter.

Features: - Optimized HTTP session with connection pooling - Enhanced retry strategies for reliable span delivery - Session statistics and monitoring capabilities - Graceful fallback to standard sessions if optimization fails

Source code in src/honeyhive/tracer/processing/otlp_exporter.py

class HoneyHiveOTLPExporter(SpanExporter):
    """HoneyHive OTLP exporter with optimized connection pooling.

    This exporter is an enhanced wrapper around the standard OpenTelemetry OTLP
    exporter that includes optimized HTTP session with connection pooling for
    better performance and reliability. All span processing should have been
    completed by the HoneyHiveSpanProcessor before spans reach this exporter.

    Features:
    - Optimized HTTP session with connection pooling
    - Enhanced retry strategies for reliable span delivery
    - Session statistics and monitoring capabilities
    - Graceful fallback to standard sessions if optimization fails
    """

    def __init__(
        self,
        tracer_instance: Any = None,
        session_config: Optional[OTLPSessionConfig] = None,
        use_optimized_session: bool = True,
        protocol: str = "http/json",
        **kwargs: Any,
    ) -> None:
        """Initialize the HoneyHive OTLP exporter with optional connection pooling.

        Args:
            tracer_instance: Optional tracer instance for logging context
            session_config: Optional configuration for optimized HTTP session
            use_optimized_session: Whether to use optimized session (default: True)
            protocol: OTLP protocol format
                - "http/json" (default) or "http/protobuf"
            **kwargs: Arguments passed to underlying OTLPSpanExporter or
                OTLPJSONExporter
        """
        self.tracer_instance = tracer_instance
        self.session_config = session_config or get_default_otlp_config(tracer_instance)
        self.use_optimized_session = use_optimized_session
        self.protocol = protocol.lower()
        self._session: Optional[requests.Session] = None
        self._is_shutdown = False
        self._use_json = self.protocol == "http/json"
        self._otlp_exporter: Union[OTLPSpanExporter, OTLPJSONExporter]

        # Create optimized session if requested and not already provided
        if use_optimized_session and "session" not in kwargs:
            try:
                self._session = create_optimized_otlp_session(
                    config=self.session_config, tracer_instance=tracer_instance
                )
                kwargs["session"] = self._session

                safe_log(
                    tracer_instance,
                    "info",
                    "HoneyHiveOTLPExporter initialized with optimized pooling",
                    honeyhive_data=self.session_config.to_dict(),
                )

            except Exception as e:
                safe_log(
                    tracer_instance,
                    "warning",
                    f"Failed to create optimized session, using default: {e}",
                    honeyhive_data={"error_type": type(e).__name__},
                )
                # Continue with default session
        else:
            # Store reference to provided session or None
            self._session = kwargs.get("session")

        # Initialize the appropriate exporter based on protocol
        if self._use_json:
            # Use JSON exporter
            endpoint = kwargs.get("endpoint")
            if not endpoint:
                raise ValueError("endpoint is required for OTLP exporter")
            headers = kwargs.get("headers", {})
            timeout = kwargs.get("timeout")
            self._otlp_exporter = OTLPJSONExporter(
                endpoint=endpoint,
                headers=headers,
                session=self._session,
                timeout=timeout,
                tracer_instance=tracer_instance,
            )
            safe_log(
                tracer_instance,
                "info",
                "HoneyHiveOTLPExporter initialized with JSON format",
                honeyhive_data={"protocol": "http/json", "endpoint": endpoint},
            )
        else:
            # Use standard Protobuf exporter
            self._otlp_exporter = OTLPSpanExporter(**kwargs)

        # Log initialization details
        session_type = (
            "optimized" if self._session and use_optimized_session else "default"
        )
        safe_log(
            tracer_instance,
            "debug",
            f"HoneyHiveOTLPExporter initialized with {session_type} session",
            honeyhive_data={
                "session_type": session_type,
                "use_optimized_session": use_optimized_session,
                "has_custom_session": "session" in kwargs,
            },
        )

    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
        """Export spans to HoneyHive via OTLP.

        This method exports spans that have already been processed by the
        HoneyHiveSpanProcessor. All attribute processing should have been
        completed before reaching this exporter.

        Args:
            spans: Sequence of ReadableSpan objects to export

        Returns:
            SpanExportResult indicating success or failure
        """
        if self._is_shutdown:
            safe_log(
                self.tracer_instance,
                "debug",
                "Exporter already shutdown, skipping export",
            )
            return SpanExportResult.FAILURE

        safe_log(
            self.tracer_instance,
            "debug",
            f"Exporting {len(spans)} processed spans to HoneyHive",
            honeyhive_data={"span_count": len(spans)},
        )

        try:
            # All span processing completed by HoneyHiveSpanProcessor
            # This exporter simply passes the spans to the underlying OTLP exporter
            return self._otlp_exporter.export(spans)

        except Exception as e:
            safe_log(
                self.tracer_instance,
                "error",
                f"Error in OTLP export: {e}",
                honeyhive_data={"error_type": type(e).__name__},
            )
            return SpanExportResult.FAILURE

    def force_flush(self, timeout_millis: int = 30000) -> bool:
        """Force flush any buffered spans."""
        if self._is_shutdown:
            safe_log(
                self.tracer_instance,
                "debug",
                "Exporter already shutdown, skipping force_flush",
            )
            return True
        return self._otlp_exporter.force_flush(timeout_millis)

    def get_session_stats(self) -> Dict[str, Any]:
        """Get connection pool statistics from the HTTP session.

        Returns:
            Dictionary containing session and connection pool statistics
        """
        if not self._session:
            return {"error": "No session available", "session_type": "default"}

        try:
            stats = get_session_stats(self._session)
            stats.update(
                {
                    "session_type": (
                        "optimized" if self.use_optimized_session else "custom"
                    ),
                    "session_config": (
                        self.session_config.to_dict() if self.session_config else None
                    ),
                }
            )
            return stats
        except Exception as e:
            return {
                "error": f"Failed to get session stats: {e}",
                "session_type": "optimized" if self.use_optimized_session else "custom",
            }

    def log_session_stats(self) -> None:
        """Log current session statistics for monitoring."""
        stats = self.get_session_stats()
        safe_log(
            self.tracer_instance,
            "debug",
            "OTLP exporter session statistics",
            honeyhive_data={"session_stats": stats},
        )

    def shutdown(self) -> None:
        """Shutdown the exporter and log final statistics."""
        if self._is_shutdown:
            safe_log(
                self.tracer_instance,
                "debug",
                "Exporter already shutdown, ignoring call",
            )
            return

        # Log final session statistics before shutdown
        if self._session and self.tracer_instance:
            try:
                final_stats = self.get_session_stats()
                safe_log(
                    self.tracer_instance,
                    "info",
                    "OTLP exporter final session statistics",
                    honeyhive_data={"final_session_stats": final_stats},
                )
            except Exception as e:
                safe_log(
                    self.tracer_instance,
                    "debug",
                    f"Could not get final session stats: {e}",
                )

        self._is_shutdown = True
        self._otlp_exporter.shutdown()
        safe_log(
            self.tracer_instance, "debug", "HoneyHiveOTLPExporter shutdown completed"
        )

tracer_instance instance-attribute

tracer_instance = tracer_instance

session_config instance-attribute

session_config = session_config or get_default_otlp_config(
    tracer_instance
)

use_optimized_session instance-attribute

use_optimized_session = use_optimized_session

protocol instance-attribute

protocol = lower()

export

export(spans: Sequence[ReadableSpan]) -> SpanExportResult

Export spans to HoneyHive via OTLP.

This method exports spans that have already been processed by the HoneyHiveSpanProcessor. All attribute processing should have been completed before reaching this exporter.

Parameters:

Name	Type	Description	Default
spans	Sequence[ReadableSpan]	Sequence of ReadableSpan objects to export	required

Returns:

Type	Description
SpanExportResult	SpanExportResult indicating success or failure

Source code in src/honeyhive/tracer/processing/otlp_exporter.py

def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
    """Export spans to HoneyHive via OTLP.

    This method exports spans that have already been processed by the
    HoneyHiveSpanProcessor. All attribute processing should have been
    completed before reaching this exporter.

    Args:
        spans: Sequence of ReadableSpan objects to export

    Returns:
        SpanExportResult indicating success or failure
    """
    if self._is_shutdown:
        safe_log(
            self.tracer_instance,
            "debug",
            "Exporter already shutdown, skipping export",
        )
        return SpanExportResult.FAILURE

    safe_log(
        self.tracer_instance,
        "debug",
        f"Exporting {len(spans)} processed spans to HoneyHive",
        honeyhive_data={"span_count": len(spans)},
    )

    try:
        # All span processing completed by HoneyHiveSpanProcessor
        # This exporter simply passes the spans to the underlying OTLP exporter
        return self._otlp_exporter.export(spans)

    except Exception as e:
        safe_log(
            self.tracer_instance,
            "error",
            f"Error in OTLP export: {e}",
            honeyhive_data={"error_type": type(e).__name__},
        )
        return SpanExportResult.FAILURE

force_flush

force_flush(timeout_millis: int = 30000) -> bool

Force flush any buffered spans.

Source code in src/honeyhive/tracer/processing/otlp_exporter.py

def force_flush(self, timeout_millis: int = 30000) -> bool:
    """Force flush any buffered spans."""
    if self._is_shutdown:
        safe_log(
            self.tracer_instance,
            "debug",
            "Exporter already shutdown, skipping force_flush",
        )
        return True
    return self._otlp_exporter.force_flush(timeout_millis)

get_session_stats

get_session_stats() -> Dict[str, Any]

Get connection pool statistics from the HTTP session.

Returns:

Type	Description
Dict[str, Any]	Dictionary containing session and connection pool statistics

Source code in src/honeyhive/tracer/processing/otlp_exporter.py

def get_session_stats(self) -> Dict[str, Any]:
    """Get connection pool statistics from the HTTP session.

    Returns:
        Dictionary containing session and connection pool statistics
    """
    if not self._session:
        return {"error": "No session available", "session_type": "default"}

    try:
        stats = get_session_stats(self._session)
        stats.update(
            {
                "session_type": (
                    "optimized" if self.use_optimized_session else "custom"
                ),
                "session_config": (
                    self.session_config.to_dict() if self.session_config else None
                ),
            }
        )
        return stats
    except Exception as e:
        return {
            "error": f"Failed to get session stats: {e}",
            "session_type": "optimized" if self.use_optimized_session else "custom",
        }

log_session_stats

log_session_stats() -> None

Log current session statistics for monitoring.

Source code in src/honeyhive/tracer/processing/otlp_exporter.py

def log_session_stats(self) -> None:
    """Log current session statistics for monitoring."""
    stats = self.get_session_stats()
    safe_log(
        self.tracer_instance,
        "debug",
        "OTLP exporter session statistics",
        honeyhive_data={"session_stats": stats},
    )

shutdown

shutdown() -> None

Shutdown the exporter and log final statistics.

Source code in src/honeyhive/tracer/processing/otlp_exporter.py

def shutdown(self) -> None:
    """Shutdown the exporter and log final statistics."""
    if self._is_shutdown:
        safe_log(
            self.tracer_instance,
            "debug",
            "Exporter already shutdown, ignoring call",
        )
        return

    # Log final session statistics before shutdown
    if self._session and self.tracer_instance:
        try:
            final_stats = self.get_session_stats()
            safe_log(
                self.tracer_instance,
                "info",
                "OTLP exporter final session statistics",
                honeyhive_data={"final_session_stats": final_stats},
            )
        except Exception as e:
            safe_log(
                self.tracer_instance,
                "debug",
                f"Could not get final session stats: {e}",
            )

    self._is_shutdown = True
    self._otlp_exporter.shutdown()
    safe_log(
        self.tracer_instance, "debug", "HoneyHiveOTLPExporter shutdown completed"
    )

HoneyHiveSpanProcessor

Bases: SpanProcessor

HoneyHive span processor for OpenTelemetry span export.

Use OTLP exporter for both immediate and batch processing

• disable_batch=True: OTLP exporter sends spans immediately
• disable_batch=False: OTLP exporter batches spans before sending

.. deprecated:: Client mode (direct Events API) is deprecated and will be removed in a future release. OTLP is the only supported export path.

Source code in src/honeyhive/tracer/processing/span_processor.py

class HoneyHiveSpanProcessor(SpanProcessor):
    """HoneyHive span processor for OpenTelemetry span export.

    OTLP mode: Use OTLP exporter for both immediate and batch processing
       - disable_batch=True: OTLP exporter sends spans immediately
       - disable_batch=False: OTLP exporter batches spans before sending

    .. deprecated::
        Client mode (direct Events API) is deprecated and will be removed
        in a future release. OTLP is the only supported export path.
    """

    def __init__(
        self,
        client: Optional[Any] = None,
        disable_batch: bool = False,
        otlp_exporter: Optional[Any] = None,
        tracer_instance: Optional[Any] = None,
    ) -> None:
        """Initialize the span processor.

        :param client: HoneyHive API client for direct Events API usage
        :type client: Optional[Any]
        :param disable_batch: If True, process spans immediately; if False, use batch
        :type disable_batch: bool
        :param otlp_exporter: OTLP exporter for batch mode (when disable_batch=False)
        :type otlp_exporter: Optional[Any]
        :param tracer_instance: HoneyHive tracer instance for session isolation
        :type tracer_instance: Optional[Any]
        """
        self.client = client
        self.disable_batch = disable_batch
        self.otlp_exporter = otlp_exporter
        self.tracer_instance = tracer_instance
        self._batch_processor: Optional[BatchSpanProcessor] = None

        # Multi-instance logging architecture uses safe_log utility
        # No need to store logger reference directly

        if client is not None:
            warnings.warn(
                "HoneyHiveSpanProcessor 'client' mode is deprecated and will be "
                "removed in a future release. Use OTLP mode instead. "
                "The client parameter is ignored; OTLP is the only supported "
                "export path.",
                DeprecationWarning,
                stacklevel=2,
            )

        # OTLP is the only supported export mode
        self.mode = "otlp"
        batch_mode = "immediate" if disable_batch else "batched"

        # When batching is enabled and we have an OTLP exporter, wrap it in
        # OTel's BatchSpanProcessor for async background export.
        # Uses OTel's upstream defaults (queue=2048, delay=5000ms, batch=512, timeout=30s).
        if not disable_batch and otlp_exporter is not None:
            self._batch_processor = BatchSpanProcessor(
                span_exporter=otlp_exporter,
            )
            self._safe_log(
                "debug",
                "🔧 BatchSpanProcessor created with OTel defaults",
            )

        self._safe_log(
            "debug",
            "🚀 HoneyHiveSpanProcessor initialized in OTLP mode (%s)",
            batch_mode,
        )

        self._safe_log(
            "debug",
            "🔧 Span processor mode: %s, disable_batch: %s",
            self.mode,
            disable_batch,
        )

        # Parse span name filters from tracer config (cached for hot-path performance)
        self._span_name_include_prefixes: List[str] = []
        self._span_name_exclude_prefixes: List[str] = []
        self._parse_span_name_filters()

    def _parse_span_name_filters(self) -> None:
        """Parse and cache span_name_filters from tracer config.

        Reads span_name_filters from the tracer instance config and caches
        the prefix values as flat lists for fast matching in on_start/on_end.
        """
        try:
            if not self.tracer_instance:
                return

            config = getattr(self.tracer_instance, "config", None)
            if not config:
                return

            filters = config.get("span_name_filters")
            if not filters:
                return

            self._do_parse_span_name_filters(filters)
        except Exception:
            self._safe_log(
                "warning",
                "Failed to parse span_name_filters config, filters disabled",
            )
            self._span_name_include_prefixes = []
            self._span_name_exclude_prefixes = []

    def _do_parse_span_name_filters(self, filters: Any) -> None:
        """Inner parsing logic for span_name_filters, separated for graceful degradation."""

        def _get(obj: Any, key: str, default: Any = None) -> Any:
            """Get value from dict or object with attributes."""
            if isinstance(obj, dict):
                return obj.get(key, default)
            return getattr(obj, key, default)

        # Extract include/exclude filter lists from either Pydantic model or dict
        for raw_list, target, _label in [
            (_get(filters, "include"), self._span_name_include_prefixes, "include"),
            (_get(filters, "exclude"), self._span_name_exclude_prefixes, "exclude"),
        ]:
            if not raw_list:
                continue
            for entry in raw_list:
                filter_type = _get(entry, "type")
                filter_value = _get(entry, "value")
                if filter_type == "prefix" and filter_value:
                    target.append(filter_value)
                elif filter_type:
                    self._safe_log(
                        "warning",
                        "Unsupported span_name_filter type: %s "
                        "(only 'prefix' is supported)",
                        filter_type,
                    )

        if self._span_name_include_prefixes or self._span_name_exclude_prefixes:
            self._safe_log(
                "debug",
                "Span name filters configured: %d include prefixes, "
                "%d exclude prefixes",
                len(self._span_name_include_prefixes),
                len(self._span_name_exclude_prefixes),
            )

    def _is_span_excluded(self, span_name: str) -> bool:
        """Check if a span should be excluded (dropped) based on span_name_filters.

        Returns True if the span should be DROPPED.

        Logic:
        - If include filters exist, span must match at least one include prefix.
        - If exclude filters exist, span must NOT match any exclude prefix.
        - If both, span must match include AND not match exclude.

        :param span_name: The name of the span to check
        :type span_name: str
        :return: True if the span should be excluded
        :rtype: bool
        """
        # No filters configured - keep all spans (fast path)
        if (
            not self._span_name_include_prefixes
            and not self._span_name_exclude_prefixes
        ):
            return False

        # Check include filters: span must match at least one
        if self._span_name_include_prefixes:
            if not any(
                span_name.startswith(p) for p in self._span_name_include_prefixes
            ):
                return True  # Drop: doesn't match any include prefix

        # Check exclude filters: span must NOT match any
        if self._span_name_exclude_prefixes:
            if any(span_name.startswith(p) for p in self._span_name_exclude_prefixes):
                return True  # Drop: matches an exclude prefix

        return False

    def _safe_log(self, level: str, message: str, *args: Any, **kwargs: Any) -> None:
        """Safely log using the centralized safe_log utility."""
        # pylint: disable=import-outside-toplevel  # Avoids circular
        # import: processor -> logger
        from ...utils.logger import safe_log

        # Format message with args if provided (maintain backward compatibility)
        if args:
            formatted_message = message % args
        else:
            formatted_message = message
        safe_log(self.tracer_instance, level, formatted_message, **kwargs)

    def _dump_raw_span_data(self, span: ReadableSpan) -> str:
        """Dump all raw span data for debugging.

        :param span: The span to dump
        :type span: ReadableSpan
        :return: Formatted string with all span properties
        :rtype: str
        """
        try:
            # Get span context
            span_context = span.context if hasattr(span, "context") else None

            # Build comprehensive span data dictionary
            span_data = {
                "name": span.name if hasattr(span, "name") else None,
                "context": (
                    {
                        "trace_id": (
                            f"{span_context.trace_id:032x}" if span_context else None
                        ),
                        "span_id": (
                            f"{span_context.span_id:016x}" if span_context else None
                        ),
                        "trace_flags": (
                            str(span_context.trace_flags) if span_context else None
                        ),
                        "trace_state": (
                            str(span_context.trace_state) if span_context else None
                        ),
                        "is_remote": span_context.is_remote if span_context else None,
                    }
                    if span_context
                    else None
                ),
                "parent": (
                    {
                        "trace_id": (
                            f"{span.parent.trace_id:032x}" if span.parent else None
                        ),
                        "span_id": (
                            f"{span.parent.span_id:016x}" if span.parent else None
                        ),
                    }
                    if hasattr(span, "parent") and span.parent
                    else None
                ),
                "kind": str(span.kind) if hasattr(span, "kind") else None,
                "start_time": span.start_time if hasattr(span, "start_time") else None,
                "end_time": span.end_time if hasattr(span, "end_time") else None,
                "status": (
                    {
                        "status_code": (
                            str(span.status.status_code)
                            if hasattr(span, "status") and span.status
                            else None
                        ),
                        "description": (
                            span.status.description
                            if hasattr(span, "status") and span.status
                            else None
                        ),
                    }
                    if hasattr(span, "status")
                    else None
                ),
                "attributes": (
                    dict(span.attributes)
                    if hasattr(span, "attributes") and span.attributes
                    else {}
                ),
                "events": [
                    {
                        "name": event.name if hasattr(event, "name") else None,
                        "timestamp": (
                            event.timestamp if hasattr(event, "timestamp") else None
                        ),
                        "attributes": (
                            dict(event.attributes)
                            if hasattr(event, "attributes") and event.attributes
                            else {}
                        ),
                    }
                    for event in (
                        span.events if hasattr(span, "events") and span.events else []
                    )
                ],
                "links": [
                    {
                        "context": {
                            "trace_id": (
                                f"{link.context.trace_id:032x}"
                                if hasattr(link, "context") and link.context
                                else None
                            ),
                            "span_id": (
                                f"{link.context.span_id:016x}"
                                if hasattr(link, "context") and link.context
                                else None
                            ),
                        },
                        "attributes": (
                            dict(link.attributes)
                            if hasattr(link, "attributes") and link.attributes
                            else {}
                        ),
                    }
                    for link in (
                        span.links if hasattr(span, "links") and span.links else []
                    )
                ],
                "resource": (
                    {
                        "attributes": (
                            dict(span.resource.attributes)
                            if hasattr(span, "resource")
                            and hasattr(span.resource, "attributes")
                            and span.resource.attributes
                            else {}
                        ),
                        "schema_url": (
                            span.resource.schema_url
                            if hasattr(span, "resource")
                            and hasattr(span.resource, "schema_url")
                            else None
                        ),
                    }
                    if hasattr(span, "resource") and span.resource
                    else None
                ),
                "instrumentation_info": (
                    {
                        "name": (
                            span.instrumentation_info.name
                            if hasattr(span, "instrumentation_info")
                            and hasattr(span.instrumentation_info, "name")
                            else None
                        ),
                        "version": (
                            span.instrumentation_info.version
                            if hasattr(span, "instrumentation_info")
                            and hasattr(span.instrumentation_info, "version")
                            else None
                        ),
                        "schema_url": (
                            span.instrumentation_info.schema_url
                            if hasattr(span, "instrumentation_info")
                            and hasattr(span.instrumentation_info, "schema_url")
                            else None
                        ),
                    }
                    if hasattr(span, "instrumentation_info")
                    and span.instrumentation_info
                    else None
                ),
            }

            # Return formatted JSON with proper indentation
            return json.dumps(span_data, indent=2, default=str)

        except Exception as e:
            return f"Error dumping span data: {e}"

    def _get_context(self, parent_context: Optional[Context]) -> Optional[Context]:
        """Get the appropriate context for baggage operations.

        :param parent_context: Parent context to use, or None to use current context
        :type parent_context: Optional[Context]
        :return: Context to use for baggage operations
        :rtype: Optional[Context]
        """
        return parent_context if parent_context is not None else context.get_current()

    def _get_basic_baggage_attributes(self, ctx: Context) -> dict:
        """Get basic baggage attributes (session_id, project, source, parent_id).

        :param ctx: OpenTelemetry context to extract baggage from
        :type ctx: Context
        :return: Dictionary of baggage attributes
        :rtype: dict
        """
        attributes = {}

        # Priority: baggage session_id (for distributed tracing),
        # then tracer instance
        # This ensures distributed traces use the propagated session_id from the client
        session_id = baggage.get_baggage("session_id", ctx)

        # Fallback to tracer instance session_id if baggage doesn't have it
        # (for local tracing scenarios)
        if not session_id:
            if self.tracer_instance and hasattr(self.tracer_instance, "session_id"):
                session_id = self.tracer_instance.session_id

        if session_id:
            attributes["honeyhive.session_id"] = session_id
            # Backend compatibility: also set Traceloop-style attribute
            attributes["traceloop.association.properties.session_id"] = session_id

        # Priority: baggage project (for distributed tracing), then tracer instance
        project = baggage.get_baggage("project", ctx)

        # Fallback to tracer instance project if baggage doesn't have it
        if not project:
            if self.tracer_instance and hasattr(self.tracer_instance, "project_name"):
                project = self.tracer_instance.project_name

        if project:
            attributes["honeyhive.project"] = project
            # Backend compatibility: also set Traceloop-style attribute
            attributes["traceloop.association.properties.project"] = project

        # Priority: baggage source (for distributed tracing), then tracer instance
        source = baggage.get_baggage("source", ctx)

        # Fallback to tracer instance source if baggage doesn't have it
        if not source:
            if self.tracer_instance and hasattr(
                self.tracer_instance, "source_environment"
            ):
                source = self.tracer_instance.source_environment

        if source:
            attributes["honeyhive.source"] = source
            # Backend compatibility: also set Traceloop-style attribute
            attributes["traceloop.association.properties.source"] = source

        parent_id = baggage.get_baggage("parent_id", ctx)
        if parent_id:
            attributes["honeyhive.parent_id"] = parent_id

        return attributes

    def _get_experiment_attributes(self) -> dict:
        """Get experiment configuration attributes.

        :return: Dictionary of experiment attributes
        :rtype: dict
        """
        attributes = {}

        try:
            # Use dynamic configuration extraction (config object and legacy attributes)
            experiment_attrs = [
                "experiment_id",
                "experiment_name",
                "experiment_variant",
                "experiment_group",
            ]

            for attr_name in experiment_attrs:
                if self.tracer_instance is not None:
                    value = self.tracer_instance.config.get(attr_name)
                    if value:
                        attributes[f"honeyhive.{attr_name}"] = value

            # Handle experiment metadata using nested config access
            experiment_metadata = None
            if self.tracer_instance is not None:
                experiment_metadata = getattr(
                    self.tracer_instance.config.experiment, "experiment_metadata", None
                )
            if experiment_metadata and isinstance(experiment_metadata, dict):
                # Add experiment metadata as individual attributes
                # for better observability
                for key, value in experiment_metadata.items():
                    attr_key = f"honeyhive.experiment_metadata.{key}"
                    attributes[attr_key] = str(value)

        except Exception as e:
            # Graceful degradation - never crash host
            self._safe_log(
                "debug",
                "Error adding experiment attributes",
                honeyhive_data={"error_type": type(e).__name__},
            )

        return attributes

    def _process_association_properties(self, ctx: Context) -> dict:
        """Process legacy association_properties from context.

        :param ctx: OpenTelemetry context to extract association properties from
        :type ctx: Context
        :return: Dictionary of association properties attributes
        :rtype: dict
        """
        attributes = {}

        try:
            # Check if context has association_properties (legacy support)
            if hasattr(ctx, "get") and callable(getattr(ctx, "get", None)):
                association_properties = ctx.get("association_properties")
                if association_properties and isinstance(association_properties, dict):
                    # Found association_properties
                    for key, value in association_properties.items():
                        if value is not None and not baggage.get_baggage(key, ctx):
                            # Set traceloop.association.properties.* format
                            # for backend compatibility
                            attr_key = f"traceloop.association.properties.{key}"
                            attributes[attr_key] = str(value)
        except Exception as e:
            # Graceful degradation - never crash host
            self._safe_log(
                "debug",
                "Error checking association_properties",
                honeyhive_data={"error_type": type(e).__name__},
            )

        return attributes

    def _get_traceloop_compatibility_attributes(self, ctx: Context) -> dict:
        """Get traceloop.association.properties.* attributes for backend compatibility.

        :param ctx: OpenTelemetry context to extract baggage from
        :type ctx: Context
        :return: Dictionary of traceloop compatibility attributes
        :rtype: dict
        """
        attributes = {}

        session_id = baggage.get_baggage("session_id", ctx)
        if session_id:
            attributes["traceloop.association.properties.session_id"] = session_id

        project = baggage.get_baggage("project", ctx)
        if project:
            attributes["traceloop.association.properties.project"] = project

        source = baggage.get_baggage("source", ctx)
        if source:
            attributes["traceloop.association.properties.source"] = source

        parent_id = baggage.get_baggage("parent_id", ctx)
        if parent_id:
            attributes["traceloop.association.properties.parent_id"] = parent_id

        return attributes

    def _get_evaluation_attributes_from_baggage(self, ctx: Context) -> dict:
        """Get evaluation metadata from baggage (run_id, dataset_id, datapoint_id).

        This method reads evaluation context that was set during evaluate() execution
        and ensures it propagates to all child spans created during
        # datapoint processing.

        :param ctx: OpenTelemetry context to extract baggage from
        :type ctx: Context
        :return: Dictionary of evaluation attributes
        :rtype: dict
        """
        attributes = {}

        # Read evaluation metadata from baggage
        run_id = baggage.get_baggage("run_id", ctx)
        if run_id:
            attributes["honeyhive_metadata.run_id"] = run_id

        dataset_id = baggage.get_baggage("dataset_id", ctx)
        if dataset_id:
            attributes["honeyhive_metadata.dataset_id"] = dataset_id

        datapoint_id = baggage.get_baggage("datapoint_id", ctx)
        if datapoint_id:
            attributes["honeyhive_metadata.datapoint_id"] = datapoint_id

        # Log if evaluation attributes were found
        if attributes:
            self._safe_log(
                "debug",
                "📊 Evaluation metadata from baggage",
                honeyhive_data={
                    "attributes": attributes,
                    "span_name": "will_be_set_on_span",
                },
            )

        return attributes

    def _get_all_baggage_attributes(self, ctx: Context) -> dict:
        """Get all baggage attributes from context, excluding already-processed keys.

        This method extracts ALL baggage items from the OpenTelemetry context and
        adds them as span attributes with a "baggage." prefix. This ensures that
        custom baggage items set by users are automatically propagated to spans.

        Excludes keys that are already handled by:
        - _get_basic_baggage_attributes (session_id, project, source, parent_id)
        - _get_evaluation_attributes_from_baggage (run_id, dataset_id, datapoint_id)

        :param ctx: OpenTelemetry context to extract baggage from
        :type ctx: Context
        :return: Dictionary of baggage attributes with "baggage." prefix
        :rtype: dict
        """
        attributes: dict[str, Any] = {}

        try:
            # Get all baggage items from context
            all_baggage = baggage.get_all(ctx)
            if not all_baggage:
                return attributes

            # Keys that are already processed by other methods
            excluded_keys = {
                "session_id",
                "project",
                "source",
                "parent_id",
                "run_id",
                "dataset_id",
                "datapoint_id",
                "honeyhive_tracer_id",  # Internal tracer discovery key
            }

            # Extract all other baggage items
            for key, value in all_baggage.items():
                if key not in excluded_keys and value is not None:
                    # Add baggage items with "baggage." prefix for clarity
                    attributes[f"baggage.{key}"] = str(value)

            if attributes:
                self._safe_log(
                    "debug",
                    "📦 Extracted custom baggage attributes",
                    honeyhive_data={
                        "baggage_keys": list(attributes.keys()),
                        "count": len(attributes),
                    },
                )

        except Exception as e:
            self._safe_log(
                "warning",
                "Failed to extract all baggage attributes: %s",
                e,
                honeyhive_data={"error_type": type(e).__name__},
            )

        return attributes

    def on_start(self, span: Span, parent_context: Optional[Context] = None) -> None:
        """Called when a span starts - enriches spans with HoneyHive attributes.

        :param span: The span that is starting
        :type span: Span
        :param parent_context: Parent context for baggage operations
        :type parent_context: Optional[Context]
        """
        self._safe_log(
            "debug",
            "🚀 SPAN PROCESSOR on_start called",
            honeyhive_data={
                "span_name": span.name,
                "span_id": span.get_span_context().span_id,
                "trace_id": span.get_span_context().trace_id,
                "tracer_instance_id": (
                    id(self.tracer_instance) if self.tracer_instance else None
                ),
                "tracer_instance_type": (
                    type(self.tracer_instance).__name__
                    if self.tracer_instance
                    else None
                ),
                "has_parent_context": parent_context is not None,
            },
        )

        try:
            # Check span name filters — skip enrichment for excluded spans
            if self._is_span_excluded(span.name):
                self._safe_log(
                    "debug",
                    "Dropping span from on_start (excluded by span_name_filters): %s",
                    span.name,
                )
                return

            ctx = self._get_context(parent_context)
            if ctx is None:
                self._safe_log(
                    "debug",
                    "⚠️ DEBUG: Context is None, exiting on_start early",
                    honeyhive_data={
                        "span_name": span.name,
                        "parent_context": parent_context,
                    },
                )
                return

            # Get session_id to determine if this span should be enriched
            # Priority: baggage session_id (distributed tracing), then
            # tracer instance. This ensures distributed traces use the
            # propagated session_id from the client
            session_id = baggage.get_baggage("session_id", ctx)

            if session_id:
                self._safe_log(
                    "debug",
                    "🔍 DEBUG: Using baggage session_id (distributed tracing)",
                    honeyhive_data={
                        "span_name": span.name,
                        "session_id": session_id,
                        "source": "baggage",
                    },
                )

            # Fallback to tracer instance session_id if baggage doesn't have it
            # (for local tracing scenarios)
            if not session_id:
                if self.tracer_instance and hasattr(self.tracer_instance, "session_id"):
                    session_id = self.tracer_instance.session_id
                    self._safe_log(
                        "debug",
                        "🔍 DEBUG: Using tracer instance session_id (local tracing)",
                        honeyhive_data={
                            "span_name": span.name,
                            "session_id": session_id,
                            "tracer_instance_id": id(self.tracer_instance),
                            "source": "tracer_instance",
                        },
                    )
                else:
                    self._safe_log(
                        "debug",
                        ("⚠️ DEBUG: No session_id found in tracer instance or baggage"),
                        honeyhive_data={
                            "span_name": span.name,
                            "tracer_instance_id": (
                                id(self.tracer_instance)
                                if self.tracer_instance
                                else None
                            ),
                            "has_tracer_instance": self.tracer_instance is not None,
                            "baggage_keys": (
                                list(baggage.get_all(ctx).keys()) if ctx else []
                            ),
                        },
                    )

            # Collect all attributes to set
            attributes_to_set = {}

            # Always process association_properties for legacy support
            attributes_to_set.update(self._process_association_properties(ctx))

            # Always add experiment attributes (they don't require session_id)
            attributes_to_set.update(self._get_experiment_attributes())

            if session_id:
                # Set session_id attributes directly (multi-instance isolation)
                attributes_to_set["honeyhive.session_id"] = session_id
                attributes_to_set["traceloop.association.properties.session_id"] = (
                    session_id
                )

                # Signal ingestion to auto-create the Session row if it doesn't
                # exist yet. Stamped on every span; ingestion is idempotent on
                # session_id.
                if getattr(self.tracer_instance, "_session_auto_create", False):
                    attributes_to_set["honeyhive.session_auto_create"] = True
                    # Prefer session_name from baggage (per-request) over the
                    # tracer-instance value (init-time), matching how
                    # session_id is resolved.
                    session_name = baggage.get_baggage("session_name", ctx)
                    if not session_name:
                        session_name = getattr(
                            self.tracer_instance, "session_name", None
                        )
                    if session_name:
                        attributes_to_set["honeyhive.session_name"] = session_name

                # Get other baggage attributes (project, source, etc.)
                other_baggage_attrs = self._get_basic_baggage_attributes(ctx)
                # Remove session_id from baggage attrs since we're setting it directly
                other_baggage_attrs.pop("honeyhive.session_id", None)
                other_baggage_attrs.pop(
                    "traceloop.association.properties.session_id", None
                )
                attributes_to_set.update(other_baggage_attrs)

                # Add traceloop compatibility attributes for backend
                attributes_to_set.update(
                    self._get_traceloop_compatibility_attributes(ctx)
                )

                # Add evaluation metadata from baggage (run_id, dataset_id,
                # datapoint_id)
                attributes_to_set.update(
                    self._get_evaluation_attributes_from_baggage(ctx)
                )

            # Add all custom baggage attributes (generalized baggage extraction)
            # This extracts ALL baggage items not already processed above
            attributes_to_set.update(self._get_all_baggage_attributes(ctx))

            # Apply all attributes to the span
            for key, value in attributes_to_set.items():
                if value is not None:
                    span.set_attribute(key, value)

            # Process all honeyhive attributes and map them to backend format
            self._process_honeyhive_attributes(span)

            # Detect and set event type using priority-based logic
            detected_event_type = self._detect_event_type(span)
            if detected_event_type:
                span.set_attribute("honeyhive_event_type", detected_event_type)
                span_context = span.get_span_context()
                self._safe_log(
                    "debug",
                    "🎯 Event type set on span: %s",
                    detected_event_type,
                    honeyhive_data={
                        "span_name": span.name,
                        "detected_event_type": detected_event_type,
                        "span_id": (
                            span_context.span_id
                            if span_context is not None
                            else "unknown"
                        ),
                    },
                )

        except Exception as e:
            # Graceful degradation - never crash host
            self._safe_log(
                "debug",
                "Error in span enrichment",
                honeyhive_data={"error_type": type(e).__name__},
            )

    def on_end(self, span: ReadableSpan) -> None:
        """Called when a span ends - send span data based on processor mode.

        :param span: The span that is ending
        :type span: ReadableSpan
        """
        try:
            self._safe_log("debug", f"🟦 ON_END CALLED for span: {span.name}")

            # Check span name filters — skip export for excluded spans
            if self._is_span_excluded(span.name):
                self._safe_log(
                    "debug",
                    "Dropping span from on_end (excluded by span_name_filters): %s",
                    span.name,
                )
                return

            # Get span duration for performance metrics
            span_context = span.get_span_context()
            if span_context is None or span_context.span_id == 0:
                self._safe_log(
                    "warning",
                    f"❌ ON_END: Invalid span context for {span.name}, skipping",
                )
                return  # Skip invalid spans

            # Extract span attributes
            attributes = {}
            if hasattr(span, "attributes") and span.attributes:
                attributes = dict(span.attributes)

            # Get session information from span attributes (set in on_start)
            session_id_raw = attributes.get("honeyhive.session_id") or attributes.get(
                "traceloop.association.properties.session_id"
            )

            if not session_id_raw:
                # Span has no session_id, skipping HoneyHive export
                self._safe_log(
                    "warning",
                    (
                        f"⚠️ ON_END: Span {span.name} has no session_id, "
                        f"skipping HoneyHive export. Attributes: "
                        f"{list(attributes.keys())}"
                    ),
                )
                return

            # Convert session_id to string
            session_id = str(session_id_raw)

            # Dump raw span data for debugging
            raw_span_data = self._dump_raw_span_data(span)
            instrumentation_scope = getattr(span, "instrumentation_scope", None)
            instrumentation_scope_name = (
                instrumentation_scope.name if instrumentation_scope else None
            )
            instrumentation_scope_version = (
                instrumentation_scope.version if instrumentation_scope else None
            )
            self._safe_log(
                "debug",
                "🔎 ON_END instrumentation scope - span: %s, scope_name: %s, scope_version: %s",
                span.name,
                instrumentation_scope_name or "unknown",
                instrumentation_scope_version or "unknown",
            )

            self._safe_log(
                "debug",
                "🚀 SPAN PROCESSOR on_end - mode: %s, span: %s\n📊 RAW DATA:\n%s",
                self.mode,
                span.name,
                raw_span_data,
            )

            if self.otlp_exporter:
                self._send_via_otlp(span, attributes, session_id)
            else:
                self._safe_log(
                    "warning",
                    "⚠️ No valid export method: OTLP exporter is %s",
                    self.otlp_exporter is not None,
                )

        except Exception as e:
            # Error processing span end - continue without disrupting application
            self._safe_log("debug", "❌ Error in span processor on_end: %s", e)

    def _send_via_client(
        self, span: ReadableSpan, attributes: dict, session_id: str
    ) -> None:
        """Send span via HoneyHive SDK client (Events API).

        .. deprecated::
            Client mode is deprecated. Use OTLP export instead.
            This method will be removed in a future release.

        :param span: The span to send
        :type span: ReadableSpan
        :param attributes: Span attributes dictionary
        :type attributes: dict
        :param session_id: HoneyHive session ID
        :type session_id: str
        :raises NotImplementedError: Always. Client mode is deprecated.
        """
        warnings.warn(
            "_send_via_client is deprecated and will be removed in a future "
            "release. Use OTLP export mode instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        raise NotImplementedError(
            "Client mode is deprecated. Use OTLP export mode instead."
        )

    def _send_via_otlp(
        self, span: ReadableSpan, _attributes: dict, _session_id: str
    ) -> None:
        """Send span via OTLP exporter - ALWAYS exports spans to ensure delivery.

        :param span: The span to send
        :type span: ReadableSpan
        :param attributes: Span attributes dictionary
        :type attributes: dict
        :param session_id: HoneyHive session ID
        :type session_id: str
        """
        try:
            batch_mode = "immediate" if self.disable_batch else "batched"
            self._safe_log(
                "debug", "🚀 OTLP EXPORT CALLED - %s MODE", batch_mode.upper()
            )

            if self._batch_processor is not None:
                # Batched async mode: delegate to OTel BatchSpanProcessor
                # which queues the span and exports in a background thread.
                self._batch_processor.on_end(span)
                self._safe_log(
                    "debug",
                    "📦 Span enqueued to BatchSpanProcessor (async batch mode)",
                )
            elif self.otlp_exporter:
                # Immediate sync mode (disable_batch=True): export inline
                result = self.otlp_exporter.export([span])
                self._safe_log(
                    "debug",
                    "✅ Span exported via OTLP exporter (immediate sync mode)",
                )

                # Log export result for debugging
                if hasattr(result, "name"):
                    self._safe_log("debug", "📊 OTLP export result: %s", result.name)
            else:
                self._safe_log("warning", "⚠️ No OTLP exporter available")

        except Exception as e:
            self._safe_log("error", "❌ Error sending via OTLP: %s", e)

    def _process_honeyhive_attributes(self, span: Span) -> None:
        """Process all honeyhive_* attributes and map them to backend-expected format.

        This method handles:
        1. Converting honeyhive_* attributes to backend format
        2. Processing _raw attributes if they exist
        3. Converting enums to strings
        4. Ensuring proper attribute naming for backend compatibility

        :param span: The span to process attributes for
        :type span: Span
        """
        try:
            # Get current span attributes
            attributes = (
                dict(span.attributes)
                if hasattr(span, "attributes") and span.attributes
                else {}
            )

            self._safe_log(
                "debug",
                "🔧 Processing honeyhive attributes for span: %s",
                span.name,
                honeyhive_data={
                    "span_name": span.name,
                    "total_attributes": len(attributes),
                    "honeyhive_attributes": [
                        k for k in attributes.keys() if k.startswith("honeyhive")
                    ],
                    "attribute_types": {
                        k: type(v).__name__
                        for k, v in attributes.items()
                        if k.startswith("honeyhive")
                    },
                },
            )

            # Define all honeyhive attributes that need processing
            honeyhive_basic_attrs = [
                "honeyhive_event_type",
                "honeyhive_event_name",
                "honeyhive_event_id",
                "honeyhive_source",
                "honeyhive_project",
                "honeyhive_session_id",
                "honeyhive_user_id",
                "honeyhive_session_name",
            ]

            honeyhive_complex_attrs = [
                "honeyhive_inputs",
                "honeyhive_config",
                "honeyhive_metadata",
                "honeyhive_metrics",
                "honeyhive_feedback",
                "honeyhive_outputs",
            ]

            # Process basic attributes
            for attr_name in honeyhive_basic_attrs:
                if attr_name in attributes:
                    value = attributes[attr_name]
                    # Convert enum to string if needed
                    processed_value = convert_enum_to_string(value)
                    if processed_value is not None:
                        # Set the processed value back to the span
                        span.set_attribute(attr_name, processed_value)
                        self._safe_log(
                            "debug",
                            "Processed basic attribute: %s = %s",
                            attr_name,
                            processed_value,
                        )

            # Process complex attributes (these might have nested structures)
            for attr_name in honeyhive_complex_attrs:
                if attr_name in attributes:
                    value = attributes[attr_name]
                    # Complex attributes processed by _set_span_attributes
                    # Just ensure they're properly formatted
                    self._safe_log("debug", "Found complex attribute: %s", attr_name)

            # Process attributes using centralized dynamic logic
            self._safe_log(
                "debug", "🔍 Processing attributes using dynamic extraction logic"
            )

            # Use the centralized dynamic logic from event_type utility
            processed_attributes = extract_raw_attributes(
                attributes, self.tracer_instance
            )

            # Set processed attributes on the span
            for attr_name, attr_value in processed_attributes.items():
                if attr_name not in attributes:  # Don't override existing attributes
                    span.set_attribute(attr_name, attr_value)
                    self._safe_log(
                        "debug",
                        "Set processed attribute: %s = %s",
                        attr_name,
                        attr_value,
                    )

        except Exception as e:
            self._safe_log("debug", "Error processing honeyhive attributes: %s", e)

    def _detect_event_type(self, span: Span) -> Optional[str]:
        """Dynamically detect event type using priority-based patterns.

        Priority Order:
        1. honeyhive_event_type_raw - Set by @trace decorator (highest priority)
        2. honeyhive_event_type - Alternative explicit format
        3. openinference.span.kind - Standard instrumentor convention
           (LLM/CHAIN/TOOL/AGENT)
        4. Span name inference - Pattern matching fallback
        5. Default to "tool" - Final fallback

        OpenInference span.kind mappings:
        - LLM → model (actual LLM invocations)
        - CHAIN → chain (multi-step workflows)
        - TOOL → tool (function/tool calls)
        - AGENT → chain (agent operations)
        - RETRIEVER → tool (retrieval operations)
        - EMBEDDING → tool (embedding generation)
        - RERANKER → tool (reranking operations)
        - GUARDRAIL → tool (guardrail checks)

        :param span: The span to analyze for event type
        :type span: Span
        :return: Detected event type or None if no detection possible
        :rtype: Optional[str]
        """
        try:
            attributes = (
                dict(span.attributes)
                if hasattr(span, "attributes") and span.attributes
                else {}
            )

            span_context = span.get_span_context()
            self._safe_log(
                "debug",
                "🔍 Starting event type detection for span: %s",
                span.name,
                honeyhive_data={
                    "span_name": span.name,
                    "available_attributes": list(attributes.keys()),
                    "span_id": (
                        span_context.span_id if span_context is not None else "unknown"
                    ),
                },
            )

            # Priority 1: Check if event type is already set
            existing_type = attributes.get("honeyhive_event_type")
            if (
                existing_type and existing_type != "tool"
            ):  # Don't return if it's just the default
                self._safe_log(
                    "debug", "✅ Event type already processed: %s", existing_type
                )
                return None  # Don't override existing processed value

            # Priority 2: Explicit _raw decorator attributes
            raw_type = attributes.get("honeyhive_event_type_raw")
            if raw_type:
                self._safe_log(
                    "debug", "✅ Event type from _raw decorator: %s", raw_type
                )
                return str(raw_type)

            # Priority 3: Direct decorator attributes
            direct_type = attributes.get("honeyhive_event_type")
            if direct_type and direct_type != "tool":
                (
                    self._safe_log(
                        "debug", "✅ Event type from decorator: %s", direct_type
                    )
                )
                return str(direct_type)

            # Priority 4: gen_ai.operation.name (GenAI semantic conventions)
            op_name = attributes.get("gen_ai.operation.name")
            if op_name:
                op_name_str = str(op_name).lower()
                event_type = _GENAI_OP_TO_EVENT_TYPE.get(op_name_str)
                if event_type:
                    self._safe_log(
                        "debug",
                        "✅ Event type from gen_ai.operation.name: %s (%s)",
                        event_type,
                        op_name,
                    )
                    return event_type
                self._safe_log(
                    "debug",
                    "Unknown gen_ai.operation.name value: %s, falling through",
                    op_name,
                )

            # Priority 5: OpenInference span.kind attribute (standard
            # instrumentor convention)
            span_kind = attributes.get("openinference.span.kind")
            if span_kind:
                # Map OpenInference span kinds to HoneyHive event types
                # Complete OpenInference span.kind mapping
                span_kind_upper = str(span_kind).upper()

                # Deterministic mapping table
                OPENINFERENCE_TO_HONEYHIVE = {
                    "LLM": "model",  # LLM invocations
                    "CHAIN": "chain",  # Multi-step workflows
                    "TOOL": "tool",  # Tool/function calls
                    "AGENT": "chain",  # Agent operations (map to chain)
                    "RETRIEVER": "tool",  # Retrieval operations
                    "EMBEDDING": "tool",  # Embedding generation (map to tool)
                    "RERANKER": "tool",  # Reranking operations
                    "GUARDRAIL": "tool",  # Guardrail checks
                }

                event_type = OPENINFERENCE_TO_HONEYHIVE.get(span_kind_upper)
                if event_type:
                    self._safe_log(
                        "debug",
                        (
                            f"✅ Event type from openinference.span.kind: "
                            f"{event_type} ({span_kind_upper})"
                        ),
                    )
                    return event_type
                else:
                    # Unknown span.kind - log warning and default to tool
                    self._safe_log(
                        "warning",
                        (
                            f"⚠️ Unknown openinference.span.kind: "
                            f"{span_kind_upper}, defaulting to tool"
                        ),
                    )
                    return "tool"

            # Priority 5: Dynamic pattern matching using utility function
            self._safe_log(
                "debug", "🔍 Using dynamic pattern matching for span: '%s'", span.name
            )

            # Use the centralized dynamic logic from event_type utility
            detected_type = detect_event_type_from_patterns(
                span.name, attributes, self.tracer_instance
            )

            if detected_type:
                self._safe_log(
                    "debug",
                    "✅ Event type detected via dynamic patterns: '%s' for span '%s'",
                    detected_type,
                    span.name,
                )
                return detected_type

            # Priority 6: Default fallback
            self._safe_log(
                "debug",
                "⚠️ No event type pattern matched for '%s', defaulting to 'tool'",
                span.name,
            )
            return "tool"

        except Exception as e:
            self._safe_log("debug", "Error in event type detection: %s", e)
            return "tool"  # Safe fallback

    def _convert_span_to_event(
        self, span: ReadableSpan, attributes: dict, session_id: str
    ) -> dict:
        """Convert OpenTelemetry span to HoneyHive event format.

        .. deprecated::
            Client mode is deprecated. This conversion is no longer needed
            since OTLP export handles span serialization natively.
            This method will be removed in a future release.

        :param span: The span to convert
        :type span: ReadableSpan
        :param attributes: Span attributes dictionary
        :type attributes: dict
        :param session_id: HoneyHive session ID
        :type session_id: str
        :raises NotImplementedError: Always. Client mode is deprecated.
        """
        warnings.warn(
            "_convert_span_to_event is deprecated and will be removed in a "
            "future release. OTLP export handles span serialization natively.",
            DeprecationWarning,
            stacklevel=2,
        )
        raise NotImplementedError(
            "Client mode is deprecated. Use OTLP export mode instead."
        )

    def shutdown(self) -> None:
        """Shutdown the span processor.

        Performs graceful shutdown of the internal BatchSpanProcessor (which
        drains its queue and shuts down the exporter), or the OTLP exporter
        directly in immediate mode.
        """
        try:
            # Shutdown the internal batch processor first — this drains the
            # queue and calls exporter.shutdown() internally.
            if self._batch_processor is not None:
                self._safe_log(
                    "debug",
                    "🛑 Shutting down internal BatchSpanProcessor",
                )
                self._batch_processor.shutdown()
                self._safe_log(
                    "debug",
                    "✅ Internal BatchSpanProcessor shutdown complete",
                )
            elif hasattr(self, "otlp_exporter") and self.otlp_exporter:
                # Immediate mode: shutdown exporter directly
                if hasattr(self.otlp_exporter, "shutdown"):
                    self.otlp_exporter.shutdown()
        except Exception as e:
            self._safe_log("debug", "Error during shutdown: %s", e)
            # Graceful degradation - continue shutdown process

    def force_flush(self, timeout_millis: float = 30000) -> bool:
        """Force flush any pending spans.

        In batched mode, this drains the internal BatchSpanProcessor queue
        and blocks until the current batch is exported (or timeout).
        In immediate mode, delegates to the OTLP exporter's force_flush.

        :param timeout_millis: Maximum time to wait for flush completion (ms).
        :type timeout_millis: float
        :return: True if flush operations completed successfully, False otherwise.
        :rtype: bool
        """
        try:
            # Batched mode: flush the internal BatchSpanProcessor
            if self._batch_processor is not None:
                self._safe_log(
                    "debug",
                    "🔄 Force flushing internal BatchSpanProcessor (timeout=%dms)",
                    int(timeout_millis),
                )
                result = self._batch_processor.force_flush(
                    timeout_millis=int(timeout_millis)
                )
                self._safe_log(
                    "debug",
                    "✅ BatchSpanProcessor force_flush result: %s",
                    result,
                )
                return bool(result)

            # Immediate mode: flush the exporter directly
            if hasattr(self, "otlp_exporter") and self.otlp_exporter:
                if hasattr(self.otlp_exporter, "force_flush"):
                    result = self.otlp_exporter.force_flush(timeout_millis)
                    return bool(result)

            return True

        except Exception as e:
            # Graceful degradation - never crash host
            self._safe_log(
                "debug",
                "HoneyHive span processor flush error",
                honeyhive_data={"error_type": type(e).__name__},
            )
            return False

client instance-attribute

client = client

disable_batch instance-attribute

disable_batch = disable_batch

otlp_exporter instance-attribute

otlp_exporter = otlp_exporter

tracer_instance instance-attribute

tracer_instance = tracer_instance

mode instance-attribute

mode = 'otlp'

on_start

on_start(
    span: Span, parent_context: Optional[Context] = None
) -> None

Called when a span starts - enriches spans with HoneyHive attributes.

:param span: The span that is starting :type span: Span :param parent_context: Parent context for baggage operations :type parent_context: Optional[Context]

Source code in src/honeyhive/tracer/processing/span_processor.py

def on_start(self, span: Span, parent_context: Optional[Context] = None) -> None:
    """Called when a span starts - enriches spans with HoneyHive attributes.

    :param span: The span that is starting
    :type span: Span
    :param parent_context: Parent context for baggage operations
    :type parent_context: Optional[Context]
    """
    self._safe_log(
        "debug",
        "🚀 SPAN PROCESSOR on_start called",
        honeyhive_data={
            "span_name": span.name,
            "span_id": span.get_span_context().span_id,
            "trace_id": span.get_span_context().trace_id,
            "tracer_instance_id": (
                id(self.tracer_instance) if self.tracer_instance else None
            ),
            "tracer_instance_type": (
                type(self.tracer_instance).__name__
                if self.tracer_instance
                else None
            ),
            "has_parent_context": parent_context is not None,
        },
    )

    try:
        # Check span name filters — skip enrichment for excluded spans
        if self._is_span_excluded(span.name):
            self._safe_log(
                "debug",
                "Dropping span from on_start (excluded by span_name_filters): %s",
                span.name,
            )
            return

        ctx = self._get_context(parent_context)
        if ctx is None:
            self._safe_log(
                "debug",
                "⚠️ DEBUG: Context is None, exiting on_start early",
                honeyhive_data={
                    "span_name": span.name,
                    "parent_context": parent_context,
                },
            )
            return

        # Get session_id to determine if this span should be enriched
        # Priority: baggage session_id (distributed tracing), then
        # tracer instance. This ensures distributed traces use the
        # propagated session_id from the client
        session_id = baggage.get_baggage("session_id", ctx)

        if session_id:
            self._safe_log(
                "debug",
                "🔍 DEBUG: Using baggage session_id (distributed tracing)",
                honeyhive_data={
                    "span_name": span.name,
                    "session_id": session_id,
                    "source": "baggage",
                },
            )

        # Fallback to tracer instance session_id if baggage doesn't have it
        # (for local tracing scenarios)
        if not session_id:
            if self.tracer_instance and hasattr(self.tracer_instance, "session_id"):
                session_id = self.tracer_instance.session_id
                self._safe_log(
                    "debug",
                    "🔍 DEBUG: Using tracer instance session_id (local tracing)",
                    honeyhive_data={
                        "span_name": span.name,
                        "session_id": session_id,
                        "tracer_instance_id": id(self.tracer_instance),
                        "source": "tracer_instance",
                    },
                )
            else:
                self._safe_log(
                    "debug",
                    ("⚠️ DEBUG: No session_id found in tracer instance or baggage"),
                    honeyhive_data={
                        "span_name": span.name,
                        "tracer_instance_id": (
                            id(self.tracer_instance)
                            if self.tracer_instance
                            else None
                        ),
                        "has_tracer_instance": self.tracer_instance is not None,
                        "baggage_keys": (
                            list(baggage.get_all(ctx).keys()) if ctx else []
                        ),
                    },
                )

        # Collect all attributes to set
        attributes_to_set = {}

        # Always process association_properties for legacy support
        attributes_to_set.update(self._process_association_properties(ctx))

        # Always add experiment attributes (they don't require session_id)
        attributes_to_set.update(self._get_experiment_attributes())

        if session_id:
            # Set session_id attributes directly (multi-instance isolation)
            attributes_to_set["honeyhive.session_id"] = session_id
            attributes_to_set["traceloop.association.properties.session_id"] = (
                session_id
            )

            # Signal ingestion to auto-create the Session row if it doesn't
            # exist yet. Stamped on every span; ingestion is idempotent on
            # session_id.
            if getattr(self.tracer_instance, "_session_auto_create", False):
                attributes_to_set["honeyhive.session_auto_create"] = True
                # Prefer session_name from baggage (per-request) over the
                # tracer-instance value (init-time), matching how
                # session_id is resolved.
                session_name = baggage.get_baggage("session_name", ctx)
                if not session_name:
                    session_name = getattr(
                        self.tracer_instance, "session_name", None
                    )
                if session_name:
                    attributes_to_set["honeyhive.session_name"] = session_name

            # Get other baggage attributes (project, source, etc.)
            other_baggage_attrs = self._get_basic_baggage_attributes(ctx)
            # Remove session_id from baggage attrs since we're setting it directly
            other_baggage_attrs.pop("honeyhive.session_id", None)
            other_baggage_attrs.pop(
                "traceloop.association.properties.session_id", None
            )
            attributes_to_set.update(other_baggage_attrs)

            # Add traceloop compatibility attributes for backend
            attributes_to_set.update(
                self._get_traceloop_compatibility_attributes(ctx)
            )

            # Add evaluation metadata from baggage (run_id, dataset_id,
            # datapoint_id)
            attributes_to_set.update(
                self._get_evaluation_attributes_from_baggage(ctx)
            )

        # Add all custom baggage attributes (generalized baggage extraction)
        # This extracts ALL baggage items not already processed above
        attributes_to_set.update(self._get_all_baggage_attributes(ctx))

        # Apply all attributes to the span
        for key, value in attributes_to_set.items():
            if value is not None:
                span.set_attribute(key, value)

        # Process all honeyhive attributes and map them to backend format
        self._process_honeyhive_attributes(span)

        # Detect and set event type using priority-based logic
        detected_event_type = self._detect_event_type(span)
        if detected_event_type:
            span.set_attribute("honeyhive_event_type", detected_event_type)
            span_context = span.get_span_context()
            self._safe_log(
                "debug",
                "🎯 Event type set on span: %s",
                detected_event_type,
                honeyhive_data={
                    "span_name": span.name,
                    "detected_event_type": detected_event_type,
                    "span_id": (
                        span_context.span_id
                        if span_context is not None
                        else "unknown"
                    ),
                },
            )

    except Exception as e:
        # Graceful degradation - never crash host
        self._safe_log(
            "debug",
            "Error in span enrichment",
            honeyhive_data={"error_type": type(e).__name__},
        )

on_end

on_end(span: ReadableSpan) -> None

Called when a span ends - send span data based on processor mode.

:param span: The span that is ending :type span: ReadableSpan

Source code in src/honeyhive/tracer/processing/span_processor.py

def on_end(self, span: ReadableSpan) -> None:
    """Called when a span ends - send span data based on processor mode.

    :param span: The span that is ending
    :type span: ReadableSpan
    """
    try:
        self._safe_log("debug", f"🟦 ON_END CALLED for span: {span.name}")

        # Check span name filters — skip export for excluded spans
        if self._is_span_excluded(span.name):
            self._safe_log(
                "debug",
                "Dropping span from on_end (excluded by span_name_filters): %s",
                span.name,
            )
            return

        # Get span duration for performance metrics
        span_context = span.get_span_context()
        if span_context is None or span_context.span_id == 0:
            self._safe_log(
                "warning",
                f"❌ ON_END: Invalid span context for {span.name}, skipping",
            )
            return  # Skip invalid spans

        # Extract span attributes
        attributes = {}
        if hasattr(span, "attributes") and span.attributes:
            attributes = dict(span.attributes)

        # Get session information from span attributes (set in on_start)
        session_id_raw = attributes.get("honeyhive.session_id") or attributes.get(
            "traceloop.association.properties.session_id"
        )

        if not session_id_raw:
            # Span has no session_id, skipping HoneyHive export
            self._safe_log(
                "warning",
                (
                    f"⚠️ ON_END: Span {span.name} has no session_id, "
                    f"skipping HoneyHive export. Attributes: "
                    f"{list(attributes.keys())}"
                ),
            )
            return

        # Convert session_id to string
        session_id = str(session_id_raw)

        # Dump raw span data for debugging
        raw_span_data = self._dump_raw_span_data(span)
        instrumentation_scope = getattr(span, "instrumentation_scope", None)
        instrumentation_scope_name = (
            instrumentation_scope.name if instrumentation_scope else None
        )
        instrumentation_scope_version = (
            instrumentation_scope.version if instrumentation_scope else None
        )
        self._safe_log(
            "debug",
            "🔎 ON_END instrumentation scope - span: %s, scope_name: %s, scope_version: %s",
            span.name,
            instrumentation_scope_name or "unknown",
            instrumentation_scope_version or "unknown",
        )

        self._safe_log(
            "debug",
            "🚀 SPAN PROCESSOR on_end - mode: %s, span: %s\n📊 RAW DATA:\n%s",
            self.mode,
            span.name,
            raw_span_data,
        )

        if self.otlp_exporter:
            self._send_via_otlp(span, attributes, session_id)
        else:
            self._safe_log(
                "warning",
                "⚠️ No valid export method: OTLP exporter is %s",
                self.otlp_exporter is not None,
            )

    except Exception as e:
        # Error processing span end - continue without disrupting application
        self._safe_log("debug", "❌ Error in span processor on_end: %s", e)

shutdown

shutdown() -> None

Shutdown the span processor.

Performs graceful shutdown of the internal BatchSpanProcessor (which drains its queue and shuts down the exporter), or the OTLP exporter directly in immediate mode.

Source code in src/honeyhive/tracer/processing/span_processor.py

def shutdown(self) -> None:
    """Shutdown the span processor.

    Performs graceful shutdown of the internal BatchSpanProcessor (which
    drains its queue and shuts down the exporter), or the OTLP exporter
    directly in immediate mode.
    """
    try:
        # Shutdown the internal batch processor first — this drains the
        # queue and calls exporter.shutdown() internally.
        if self._batch_processor is not None:
            self._safe_log(
                "debug",
                "🛑 Shutting down internal BatchSpanProcessor",
            )
            self._batch_processor.shutdown()
            self._safe_log(
                "debug",
                "✅ Internal BatchSpanProcessor shutdown complete",
            )
        elif hasattr(self, "otlp_exporter") and self.otlp_exporter:
            # Immediate mode: shutdown exporter directly
            if hasattr(self.otlp_exporter, "shutdown"):
                self.otlp_exporter.shutdown()
    except Exception as e:
        self._safe_log("debug", "Error during shutdown: %s", e)

force_flush

force_flush(timeout_millis: float = 30000) -> bool

Force flush any pending spans.

In batched mode, this drains the internal BatchSpanProcessor queue and blocks until the current batch is exported (or timeout). In immediate mode, delegates to the OTLP exporter's force_flush.

:param timeout_millis: Maximum time to wait for flush completion (ms). :type timeout_millis: float :return: True if flush operations completed successfully, False otherwise. :rtype: bool

Source code in src/honeyhive/tracer/processing/span_processor.py

def force_flush(self, timeout_millis: float = 30000) -> bool:
    """Force flush any pending spans.

    In batched mode, this drains the internal BatchSpanProcessor queue
    and blocks until the current batch is exported (or timeout).
    In immediate mode, delegates to the OTLP exporter's force_flush.

    :param timeout_millis: Maximum time to wait for flush completion (ms).
    :type timeout_millis: float
    :return: True if flush operations completed successfully, False otherwise.
    :rtype: bool
    """
    try:
        # Batched mode: flush the internal BatchSpanProcessor
        if self._batch_processor is not None:
            self._safe_log(
                "debug",
                "🔄 Force flushing internal BatchSpanProcessor (timeout=%dms)",
                int(timeout_millis),
            )
            result = self._batch_processor.force_flush(
                timeout_millis=int(timeout_millis)
            )
            self._safe_log(
                "debug",
                "✅ BatchSpanProcessor force_flush result: %s",
                result,
            )
            return bool(result)

        # Immediate mode: flush the exporter directly
        if hasattr(self, "otlp_exporter") and self.otlp_exporter:
            if hasattr(self.otlp_exporter, "force_flush"):
                result = self.otlp_exporter.force_flush(timeout_millis)
                return bool(result)

        return True

    except Exception as e:
        # Graceful degradation - never crash host
        self._safe_log(
            "debug",
            "HoneyHive span processor flush error",
            honeyhive_data={"error_type": type(e).__name__},
        )
        return False

extract_context_from_carrier

extract_context_from_carrier(
    carrier: Dict[str, str],
    tracer_instance: HoneyHiveTracer,
) -> Optional[Context]

Extract OpenTelemetry context from a carrier dictionary.

This function extracts OpenTelemetry context (including trace context and baggage) from a carrier dictionary, typically received from another service or process.

:param carrier: Dictionary containing context information :type carrier: Dict[str, str] :param tracer_instance: The tracer instance for propagator access :type tracer_instance: HoneyHiveTracer :return: Extracted OpenTelemetry context or None if extraction fails :rtype: Optional[Context]

Example:

.. code-block:: python

# Extract context from HTTP headers
extracted_context = extract_context_from_carrier(request.headers, tracer)

# Use extracted context as parent for new spans
with tracer.start_span("operation", context=extracted_context) as span:
    # This span will be a child of the remote span
    pass

Note:

This function is typically used in service endpoints to continue distributed traces from upstream services. The extracted context can be used as a parent context for new spans.

Source code in src/honeyhive/tracer/processing/context.py

def extract_context_from_carrier(
    carrier: Dict[str, str], tracer_instance: "HoneyHiveTracer"
) -> Optional["Context"]:
    """Extract OpenTelemetry context from a carrier dictionary.

    This function extracts OpenTelemetry context (including trace context
    and baggage) from a carrier dictionary, typically received from another
    service or process.

    :param carrier: Dictionary containing context information
    :type carrier: Dict[str, str]
    :param tracer_instance: The tracer instance for propagator access
    :type tracer_instance: HoneyHiveTracer
    :return: Extracted OpenTelemetry context or None if extraction fails
    :rtype: Optional[Context]

    **Example:**

    .. code-block:: python

        # Extract context from HTTP headers
        extracted_context = extract_context_from_carrier(request.headers, tracer)

        # Use extracted context as parent for new spans
        with tracer.start_span("operation", context=extracted_context) as span:
            # This span will be a child of the remote span
            pass

    **Note:**

    This function is typically used in service endpoints to continue
    distributed traces from upstream services. The extracted context
    can be used as a parent context for new spans.
    """

    try:
        if not tracer_instance.propagator:
            safe_log(
                tracer_instance,
                "warning",
                "No propagator available for context extraction",
            )
            return None

        # Extract context from carrier
        extracted_context: Optional["Context"] = tracer_instance.propagator.extract(
            carrier
        )

        safe_log(
            tracer_instance,
            "debug",
            "Context extracted from carrier",
            honeyhive_data={
                "carrier_keys": list(carrier.keys()),
                "has_context": extracted_context is not None,
            },
        )

        return extracted_context

    except Exception as e:
        safe_log(
            tracer_instance,
            "error",
            "Failed to extract context from carrier: %s",
            e,
            honeyhive_data={"carrier_keys": list(carrier.keys())},
        )
        return None

get_current_baggage

get_current_baggage() -> Dict[str, str]

Get all baggage items from the current OpenTelemetry context.

This function dynamically discovers and returns all baggage items from the current context, providing a way to inspect what context information is available to spans.

:return: Dictionary of all baggage key-value pairs :rtype: Dict[str, str]

Example:

.. code-block:: python

current_baggage = get_current_baggage()
print(f"Session ID: {current_baggage.get('session_id')}")
print(f"Project: {current_baggage.get('project')}")

Note:

This function provides read-only access to the current baggage. To modify baggage, use the tracer's baggage management methods.

Source code in src/honeyhive/tracer/processing/context.py

def get_current_baggage() -> Dict[str, str]:
    """Get all baggage items from the current OpenTelemetry context.

    This function dynamically discovers and returns all baggage items
    from the current context, providing a way to inspect what context
    information is available to spans.

    :return: Dictionary of all baggage key-value pairs
    :rtype: Dict[str, str]

    **Example:**

    .. code-block:: python

        current_baggage = get_current_baggage()
        print(f"Session ID: {current_baggage.get('session_id')}")
        print(f"Project: {current_baggage.get('project')}")

    **Note:**

    This function provides read-only access to the current baggage.
    To modify baggage, use the tracer's baggage management methods.
    """
    try:
        current_baggage = {}
        ctx = context.get_current()

        # Get all baggage items dynamically
        baggage_dict = baggage.get_all(ctx)

        for key, value in baggage_dict.items():
            current_baggage[key] = str(value)

        # Baggage retrieved successfully (removed logging from utility function)

        return current_baggage

    except Exception:
        # Error getting baggage (removed logging from utility function)
        return {}

inject_context_into_carrier

inject_context_into_carrier(
    carrier: Dict[str, str],
    tracer_instance: HoneyHiveTracer,
) -> None

Inject OpenTelemetry context into a carrier dictionary.

This function injects the current OpenTelemetry context (including trace context and baggage) into a carrier dictionary for cross-service or cross-process propagation.

:param carrier: Dictionary to inject context into :type carrier: Dict[str, str] :param tracer_instance: The tracer instance for propagator access :type tracer_instance: HoneyHiveTracer

Example:

.. code-block:: python

headers = {}
inject_context_into_carrier(headers, tracer)
# headers now contains trace context and baggage

# Use headers in HTTP request
response = requests.get(url, headers=headers)

Note:

The carrier dictionary will be modified in-place with context information. This is typically used for HTTP headers or message metadata in distributed systems.

Source code in src/honeyhive/tracer/processing/context.py

def inject_context_into_carrier(
    carrier: Dict[str, str], tracer_instance: "HoneyHiveTracer"
) -> None:
    """Inject OpenTelemetry context into a carrier dictionary.

    This function injects the current OpenTelemetry context (including
    trace context and baggage) into a carrier dictionary for cross-service
    or cross-process propagation.

    :param carrier: Dictionary to inject context into
    :type carrier: Dict[str, str]
    :param tracer_instance: The tracer instance for propagator access
    :type tracer_instance: HoneyHiveTracer

    **Example:**

    .. code-block:: python

        headers = {}
        inject_context_into_carrier(headers, tracer)
        # headers now contains trace context and baggage

        # Use headers in HTTP request
        response = requests.get(url, headers=headers)

    **Note:**

    The carrier dictionary will be modified in-place with context
    information. This is typically used for HTTP headers or message
    metadata in distributed systems.
    """

    try:
        if not tracer_instance.propagator:
            safe_log(
                tracer_instance,
                "warning",
                "No propagator available for context injection",
            )
            return

        # Inject current context into carrier
        tracer_instance.propagator.inject(carrier)

        safe_log(
            tracer_instance,
            "debug",
            "Context injected into carrier",
            honeyhive_data={
                "carrier_keys": list(carrier.keys()),
                "injected_items": len(carrier),
            },
        )

    except Exception as e:
        safe_log(
            tracer_instance,
            "error",
            "Failed to inject context into carrier: %s",
            e,
            honeyhive_data={"carrier_keys": list(carrier.keys())},
        )

setup_baggage_context

setup_baggage_context(
    tracer_instance: HoneyHiveTracer,
) -> None

Set up baggage with session context for OpenInference integration.

This function dynamically discovers and sets up baggage items from the tracer configuration and environment. It supports experiment harness integration and evaluation context propagation.

:param tracer_instance: The tracer instance to setup baggage for :type tracer_instance: HoneyHiveTracer

Example:

.. code-block:: python

tracer = HoneyHiveTracer(api_key="key", project="project")
setup_baggage_context(tracer)
# Baggage is now available to all spans created by this tracer

Note:

This function uses dynamic discovery to find all relevant context information and automatically sets up baggage for downstream spans. It gracefully handles missing or invalid configuration.

Source code in src/honeyhive/tracer/processing/context.py

def setup_baggage_context(tracer_instance: "HoneyHiveTracer") -> None:
    """Set up baggage with session context for OpenInference integration.

    This function dynamically discovers and sets up baggage items from the
    tracer configuration and environment. It supports experiment harness
    integration and evaluation context propagation.

    :param tracer_instance: The tracer instance to setup baggage for
    :type tracer_instance: HoneyHiveTracer

    **Example:**

    .. code-block:: python

        tracer = HoneyHiveTracer(api_key="key", project="project")
        setup_baggage_context(tracer)
        # Baggage is now available to all spans created by this tracer

    **Note:**

    This function uses dynamic discovery to find all relevant context
    information and automatically sets up baggage for downstream spans.
    It gracefully handles missing or invalid configuration.
    """
    try:
        # Dynamically discover baggage items
        baggage_items = _discover_baggage_items(tracer_instance)

        # Set up baggage context
        _apply_baggage_context(baggage_items, tracer_instance)

        safe_log(
            tracer_instance,
            "debug",
            "Baggage context set up successfully",
            honeyhive_data={
                "baggage_items": list(baggage_items.keys()),
                "item_count": len(baggage_items),
            },
        )

    except Exception as e:
        safe_log(
            tracer_instance,
            "warning",
            "Failed to set up baggage context",
            honeyhive_data={"error": str(e)},
        )

with_distributed_trace_context

with_distributed_trace_context(
    carrier: Dict[str, str],
    tracer_instance: HoneyHiveTracer,
    *,
    session_id: Optional[str] = None
) -> Iterator[Context]

Context manager for distributed tracing that extracts and sets up context.

This function extracts OpenTelemetry context from a carrier (e.g., HTTP headers), extracts session_id from baggage if available, and attaches the context with session_id in baggage. This is the recommended way to handle distributed tracing on the server side.

:param carrier: Dictionary containing trace context (e.g., HTTP headers) :type carrier: Dict[str, str] :param tracer_instance: The tracer instance for propagator access :type tracer_instance: HoneyHiveTracer :param session_id: Optional explicit session_id to use (overrides baggage) :type session_id: Optional[str] :return: Context manager that yields the extracted context

Example:

.. code-block:: python

@app.route("/api/endpoint", methods=["POST"])
def my_endpoint():
    with with_distributed_trace_context(dict(request.headers), tracer) as ctx:
        # All spans created here will use the propagated session_id
        with tracer.start_span("operation"):
            pass

Note for async functions:

If you need to use this with asyncio.run(), you'll need to re-attach the context inside the async function since asyncio.run() creates a new event loop:

.. code-block:: python

with with_distributed_trace_context(dict(request.headers), tracer) as ctx:
    async def my_async_function():
        # Re-attach context in new event loop
        token = context.attach(ctx)
        try:
            # Your async code here
            pass
        finally:
            context.detach(token)

    asyncio.run(my_async_function())

Source code in src/honeyhive/tracer/processing/context.py

@contextmanager
def with_distributed_trace_context(
    carrier: Dict[str, str],
    tracer_instance: "HoneyHiveTracer",
    *,
    session_id: Optional[str] = None,
) -> Iterator["Context"]:
    """Context manager for distributed tracing that extracts and sets up context.

    This function extracts OpenTelemetry context from a carrier (e.g., HTTP headers),
    extracts session_id from baggage if available, and attaches the context with
    session_id in baggage. This is the recommended way to handle distributed tracing
    on the server side.

    :param carrier: Dictionary containing trace context (e.g., HTTP headers)
    :type carrier: Dict[str, str]
    :param tracer_instance: The tracer instance for propagator access
    :type tracer_instance: HoneyHiveTracer
    :param session_id: Optional explicit session_id to use (overrides baggage)
    :type session_id: Optional[str]
    :return: Context manager that yields the extracted context

    **Example:**

    .. code-block:: python

        @app.route("/api/endpoint", methods=["POST"])
        def my_endpoint():
            with with_distributed_trace_context(dict(request.headers), tracer) as ctx:
                # All spans created here will use the propagated session_id
                with tracer.start_span("operation"):
                    pass

    **Note for async functions:**

    If you need to use this with `asyncio.run()`, you'll need to re-attach the context
    inside the async function since `asyncio.run()` creates a new event loop:

    .. code-block:: python

        with with_distributed_trace_context(dict(request.headers), tracer) as ctx:
            async def my_async_function():
                # Re-attach context in new event loop
                token = context.attach(ctx)
                try:
                    # Your async code here
                    pass
                finally:
                    context.detach(token)

            asyncio.run(my_async_function())
    """
    # Extract trace context from carrier
    incoming_context = extract_context_from_carrier(carrier, tracer_instance)

    # Extract session_id, project, source from baggage header if not explicit
    propagated_session_id = session_id
    propagated_project = None
    propagated_source = None

    if not propagated_session_id:
        baggage_header = carrier.get("baggage") or carrier.get("Baggage")
        if baggage_header:
            # Parse baggage manually (fallback if extract doesn't populate)
            for item in baggage_header.split(","):
                if "=" in item:
                    key, value = item.split("=", 1)
                    key = key.strip()
                    value = value.strip()

                    # Extract session_id
                    if key in (
                        "session_id",
                        "honeyhive_session_id",
                        "honeyhive.session_id",
                    ):
                        propagated_session_id = value
                    # Extract project
                    elif key in ("project", "honeyhive_project", "honeyhive.project"):
                        propagated_project = value
                    # Extract source
                    elif key in ("source", "honeyhive_source", "honeyhive.source"):
                        propagated_source = value

    # Set up context with session_id, project, and source in baggage
    context_to_use = incoming_context if incoming_context else context.get_current()
    if propagated_session_id:
        context_to_use = baggage.set_baggage(
            "session_id", propagated_session_id, context_to_use
        )
    if propagated_project:
        context_to_use = baggage.set_baggage(
            "project", propagated_project, context_to_use
        )
    if propagated_source:
        context_to_use = baggage.set_baggage(
            "source", propagated_source, context_to_use
        )

    # Attach context
    token = context.attach(context_to_use)
    try:
        yield context_to_use
    finally:
        context.detach(token)