honeyhive.tracer

HoneyHive OpenTelemetry tracer module.

This module provides the complete public API for HoneyHive tracing functionality. Users should import from this module rather than internal submodules for stability.

Example

from honeyhive.tracer import HoneyHiveTracer, trace, enrich_span

tracer = HoneyHiveTracer.init(api_key="...", project="...")

@trace(tracer=tracer) def my_function(): return "Hello, World!"

enrich_span module-attribute

enrich_span = UnifiedEnrichSpan()

HoneyHiveTracer

Bases: HoneyHiveTracerBase, TracerOperationsMixin, TracerContextMixin

HoneyHive OpenTelemetry tracer with dynamic multi-instance architecture.

This tracer class is composed from multiple mixins using dynamic inheritance, providing a comprehensive tracing solution with flexible configuration, robust error handling, and multi-instance support.

The class combines: - HoneyHiveTracerBase: Core initialization and configuration - TracerOperationsMixin: Span creation and event management - TracerContextMixin: Context and baggage management

All operations use dynamic logic for flexible parameter handling, automatic error recovery, and graceful degradation.

Example

New Pydantic config approach (recommended)

config = TracerConfig(api_key="...", project="...", verbose=True) tracer = HoneyHiveTracer(config=config)

Backwards compatible approach

tracer = HoneyHiveTracer(api_key="...", project="...", verbose=True)

Dynamic span creation

with tracer.start_span("operation") as span: ... span.set_attribute("key", "value")

Dynamic event creation

event_id = tracer.create_event( ... event_name="my_event", ... event_type="tool", ... inputs={"input": "data"}, ... outputs={"output": "result"} ... )

Source code in src/honeyhive/tracer/core/tracer.py

class HoneyHiveTracer(HoneyHiveTracerBase, TracerOperationsMixin, TracerContextMixin):
    """HoneyHive OpenTelemetry tracer with dynamic multi-instance architecture.

    This tracer class is composed from multiple mixins using dynamic inheritance,
    providing a comprehensive tracing solution with flexible configuration,
    robust error handling, and multi-instance support.

    The class combines:
    - HoneyHiveTracerBase: Core initialization and configuration
    - TracerOperationsMixin: Span creation and event management
    - TracerContextMixin: Context and baggage management

    All operations use dynamic logic for flexible parameter handling,
    automatic error recovery, and graceful degradation.

    Example:
        >>> # New Pydantic config approach (recommended)
        >>> config = TracerConfig(api_key="...", project="...", verbose=True)
        >>> tracer = HoneyHiveTracer(config=config)
        >>>
        >>> # Backwards compatible approach
        >>> tracer = HoneyHiveTracer(api_key="...", project="...", verbose=True)

        >>> # Dynamic span creation
        >>> with tracer.start_span("operation") as span:
        ...     span.set_attribute("key", "value")

        >>> # Dynamic event creation
        >>> event_id = tracer.create_event(
        ...     event_name="my_event",
        ...     event_type="tool",
        ...     inputs={"input": "data"},
        ...     outputs={"output": "result"}
        ... )
    """

    # Explicit implementation to satisfy ABC requirements
    # The TracerContextMixin provides the actual implementation
    def get_baggage(self, key: str) -> Optional[str]:
        """Get baggage value by key.

        Delegates to TracerContextMixin implementation.

        Args:
            key: The baggage key to retrieve

        Returns:
            Baggage value or None if not found
        """
        return TracerContextMixin.get_baggage(self, key)

    def flush(self, timeout_millis: float = 30000) -> bool:
        """Flush tracer data. Alias for force_flush().

        Args:
            timeout_millis: Timeout in milliseconds (default 30000)

        Returns:
            True if flush successful, False otherwise
        """
        return self.force_flush(timeout_millis)

    @classmethod
    def flush_all(cls, timeout_millis: float = 30000) -> bool:
        """Flush all active tracer instances.

        This is a convenience class method that flushes all active tracers.

        Args:
            timeout_millis: Timeout in milliseconds (default 30000)

        Returns:
            True if all flushes successful, False otherwise
        """
        from honeyhive.tracer.lifecycle.flush import force_flush_tracer

        # Get current tracer from context if available
        try:
            from honeyhive.tracer.processing.context import get_current_tracer

            current = get_current_tracer()
            if current:
                return force_flush_tracer(current, timeout_millis)
        except Exception:
            pass
        return True

    def __repr__(self) -> str:
        """Dynamic string representation of tracer instance."""
        return (
            f"HoneyHiveTracer("
            f"project={self.project_name!r}, "
            f"source={self.source_environment!r}, "
            f"initialized={self.is_initialized}, "
            f"test_mode={self.is_test_mode})"
        )

get_baggage

get_baggage(key: str) -> Optional[str]

Get baggage value by key.

Delegates to TracerContextMixin implementation.

Parameters:

Name	Type	Description	Default
key	str	The baggage key to retrieve	required

Returns:

Type	Description
Optional[str]	Baggage value or None if not found

Source code in src/honeyhive/tracer/core/tracer.py

def get_baggage(self, key: str) -> Optional[str]:
    """Get baggage value by key.

    Delegates to TracerContextMixin implementation.

    Args:
        key: The baggage key to retrieve

    Returns:
        Baggage value or None if not found
    """
    return TracerContextMixin.get_baggage(self, key)

flush

flush(timeout_millis: float = 30000) -> bool

Flush tracer data. Alias for force_flush().

Parameters:

Name	Type	Description	Default
timeout_millis	float	Timeout in milliseconds (default 30000)	30000

Returns:

Type	Description
bool	True if flush successful, False otherwise

Source code in src/honeyhive/tracer/core/tracer.py

def flush(self, timeout_millis: float = 30000) -> bool:
    """Flush tracer data. Alias for force_flush().

    Args:
        timeout_millis: Timeout in milliseconds (default 30000)

    Returns:
        True if flush successful, False otherwise
    """
    return self.force_flush(timeout_millis)

flush_all classmethod

flush_all(timeout_millis: float = 30000) -> bool

Flush all active tracer instances.

This is a convenience class method that flushes all active tracers.

Parameters:

Name	Type	Description	Default
timeout_millis	float	Timeout in milliseconds (default 30000)	30000

Returns:

Type	Description
bool	True if all flushes successful, False otherwise

Source code in src/honeyhive/tracer/core/tracer.py

@classmethod
def flush_all(cls, timeout_millis: float = 30000) -> bool:
    """Flush all active tracer instances.

    This is a convenience class method that flushes all active tracers.

    Args:
        timeout_millis: Timeout in milliseconds (default 30000)

    Returns:
        True if all flushes successful, False otherwise
    """
    from honeyhive.tracer.lifecycle.flush import force_flush_tracer

    # Get current tracer from context if available
    try:
        from honeyhive.tracer.processing.context import get_current_tracer

        current = get_current_tracer()
        if current:
            return force_flush_tracer(current, timeout_millis)
    except Exception:
        pass
    return True

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

atrace

atrace(
    event_type: Optional[str] = None,
    event_name: Optional[str] = None,
    **kwargs: Any
) -> Union[
    Callable[[Callable[..., Any]], Callable[..., Any]],
    Callable[..., Any],
]

Legacy async-specific trace decorator (deprecated).

Note

This decorator is maintained for backwards compatibility. Use the unified :func:trace decorator instead, which auto-detects sync/async functions.

Parameters:

Name	Type	Description	Default
event_type	Optional[str]	Type of event being traced (e.g., "model", "tool", "chain")	None
event_name	Optional[str]	Name of the event (defaults to function name)	None
**kwargs	Any	Additional tracing parameters (source, project, session_id, etc.)	{}

Returns:

Type	Description
Union[Callable[[Callable[..., Any]], Callable[..., Any]], Callable[..., Any]]	Decorated async function with tracing capabilities

See Also

:func:trace: Unified decorator that handles both sync and async functions

Source code in src/honeyhive/tracer/instrumentation/decorators.py

def atrace(
    event_type: Optional[str] = None,
    event_name: Optional[str] = None,
    **kwargs: Any,
) -> Union[Callable[[Callable[..., Any]], Callable[..., Any]], Callable[..., Any]]:
    """Legacy async-specific trace decorator (deprecated).

    Note:
        This decorator is maintained for backwards compatibility.
        Use the unified :func:`trace` decorator instead, which auto-detects
        sync/async functions.

    Args:
        event_type: Type of event being traced (e.g., "model", "tool", "chain")
        event_name: Name of the event (defaults to function name)
        **kwargs: Additional tracing parameters (source, project, session_id, etc.)

    Returns:
        Decorated async function with tracing capabilities

    See Also:
        :func:`trace`: Unified decorator that handles both sync and async functions
    """

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        params = _create_tracing_params(
            event_type=event_type, event_name=event_name, **kwargs
        )
        return _create_wrapper(func, params, is_async=True, **kwargs)

    # Handle both @atrace and @atrace() usage patterns
    if event_type is not None and callable(event_type):
        # Used as @atrace (without parentheses)
        func = event_type
        params = _create_tracing_params(event_type="tool")
        return _create_wrapper(func, params, is_async=True)

    # Used as @atrace(...) (with parentheses)
    return decorator

trace

trace(
    event_type: Optional[str] = None,
    event_name: Optional[str] = None,
    **kwargs: Any
) -> Union[
    Callable[[Callable[..., T]], Callable[..., T]],
    Callable[..., T],
]

Unified trace decorator that auto-detects sync/async functions.

Automatically detects whether the decorated function is synchronous or asynchronous and applies the appropriate wrapper. This decorator can be used on both sync and async functions without needing separate decorators.

Parameters:

Name	Type	Description	Default
event_type	Optional[str]	Type of event being traced (e.g., "model", "tool", "chain")	None
event_name	Optional[str]	Name of the event (defaults to function name)	None
**kwargs	Any	Additional tracing parameters (source, project, session_id, etc.)	{}

Returns:

Type	Description
Union[Callable[[Callable[..., T]], Callable[..., T]], Callable[..., T]]	Decorated function with tracing capabilities

Example

@trace(event_type="model", event_name="gpt_call") ... def sync_function(): ... return "result"

@trace(event_type="model", event_name="async_gpt_call") ... async def async_function(): ... return "async result"

Source code in src/honeyhive/tracer/instrumentation/decorators.py

def trace(
    event_type: Optional[str] = None,
    event_name: Optional[str] = None,
    **kwargs: Any,
) -> Union[Callable[[Callable[..., T]], Callable[..., T]], Callable[..., T]]:
    """Unified trace decorator that auto-detects sync/async functions.

    Automatically detects whether the decorated function is synchronous or
    asynchronous and applies the appropriate wrapper. This decorator can be
    used on both sync and async functions without needing separate decorators.

    Args:
        event_type: Type of event being traced (e.g., "model", "tool", "chain")
        event_name: Name of the event (defaults to function name)
        **kwargs: Additional tracing parameters (source, project, session_id, etc.)

    Returns:
        Decorated function with tracing capabilities

    Example:
        >>> @trace(event_type="model", event_name="gpt_call")
        ... def sync_function():
        ...     return "result"

        >>> @trace(event_type="model", event_name="async_gpt_call")
        ... async def async_function():
        ...     return "async result"
    """

    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        # Auto-detect if function is async
        is_async = inspect.iscoroutinefunction(func)

        # Filter out tracer argument for _create_tracing_params
        tracing_kwargs = {k: v for k, v in kwargs.items() if k != "tracer"}
        params = _create_tracing_params(
            event_type=event_type, event_name=event_name, **tracing_kwargs
        )
        return _create_wrapper(func, params, is_async=is_async, **kwargs)

    # Handle both @trace and @trace() usage patterns
    if event_type is not None and callable(event_type):
        # Used as @trace (without parentheses)
        func = event_type
        is_async = inspect.iscoroutinefunction(func)
        params = _create_tracing_params(event_type="tool")
        return _create_wrapper(func, params, is_async=is_async)

    # Used as @trace(...) (with parentheses)
    return decorator

trace_class

trace_class(cls: type) -> type

Class decorator to automatically trace all public methods.

Uses dynamic reflection to discover and wrap all public methods of a class. Automatically detects sync/async methods and applies appropriate tracing.

Parameters:

Name	Type	Description	Default
cls	type	The class to be decorated	required

Returns:

Type	Description
type	The decorated class with all public methods traced

Example

@trace_class ... class MyService: ... def process_data(self, data): ... return data.upper() ... ... async def async_process(self, data): ... return await some_async_operation(data)

Source code in src/honeyhive/tracer/instrumentation/decorators.py

def trace_class(cls: type) -> type:
    """Class decorator to automatically trace all public methods.

    Uses dynamic reflection to discover and wrap all public methods of a class.
    Automatically detects sync/async methods and applies appropriate tracing.

    Args:
        cls: The class to be decorated

    Returns:
        The decorated class with all public methods traced

    Example:
        >>> @trace_class
        ... class MyService:
        ...     def process_data(self, data):
        ...         return data.upper()
        ...
        ...     async def async_process(self, data):
        ...         return await some_async_operation(data)
    """
    # Dynamically discover and wrap methods
    for attr_name in dir(cls):
        attr_value = getattr(cls, attr_name)

        # Dynamic method detection
        if (
            callable(attr_value)
            and not attr_name.startswith("_")
            and not isinstance(attr_value, (classmethod, staticmethod))
        ):
            # Determine if method is async
            is_async_method = inspect.iscoroutinefunction(attr_value)

            # Create tracing params for method
            params = _create_tracing_params(
                event_type="tool",
                event_name=f"{cls.__name__}.{attr_name}",
            )

            # Wrap method with appropriate wrapper
            wrapped_method = _create_wrapper(
                attr_value, params, is_async=is_async_method
            )
            setattr(cls, attr_name, wrapped_method)

    return cls

enrich_session

enrich_session(
    session_id: str,
    metadata: Optional[Dict[str, Any]] = None,
    tracer: Optional[Any] = None,
    tracer_instance: Optional[Any] = None,
) -> None

LEGACY (v1.0+): Dynamically enrich session with metadata.

.. deprecated:: 1.0 This free function pattern is provided for backward compatibility only. Use instance methods instead: tracer.enrich_session() This pattern will be removed in v2.0.

Recommended Pattern (v1.0+): Use the tracer instance method for explicit tracer reference::

tracer = HoneyHiveTracer.init(api_key="...", project="...")
tracer.enrich_session(
    metadata={"user_id": "user-456"},
    user_properties={"plan": "premium"}
)

This function provides backward compatibility for the global enrich_session function using dynamic tracer discovery and flexible metadata handling.

Parameters:

Name	Type	Description	Default
session_id	str	The session ID to enrich	required
metadata	Optional[Dict[str, Any]]	Metadata dictionary to add to the session	None
tracer	Optional[Any]	Optional tracer instance to use	None
tracer_instance	Optional[Any]	Optional tracer instance for logging context	None

Legacy Example

Using default tracer (backward compatibility)

enrich_session("session-123", {"user_id": "user-456"})

Using specific tracer (backward compatibility)

enrich_session("session-123", {"user_id": "user-456"}, tracer=my_tracer)

See Also

• :meth:HoneyHiveTracer.enrich_session - Primary pattern (v1.0+)
• :meth:HoneyHiveTracer.enrich_span - Span enrichment

Source code in src/honeyhive/tracer/integration/compatibility.py

def enrich_session(
    session_id: str,
    metadata: Optional[Dict[str, Any]] = None,
    tracer: Optional[Any] = None,
    tracer_instance: Optional[Any] = None,
) -> None:
    """**LEGACY (v1.0+):** Dynamically enrich session with metadata.

    .. deprecated:: 1.0
       This free function pattern is provided for backward compatibility only.
       **Use instance methods instead:** ``tracer.enrich_session()``
       This pattern will be removed in v2.0.

    **Recommended Pattern (v1.0+):**
    Use the tracer instance method for explicit tracer reference::

        tracer = HoneyHiveTracer.init(api_key="...", project="...")
        tracer.enrich_session(
            metadata={"user_id": "user-456"},
            user_properties={"plan": "premium"}
        )

    This function provides backward compatibility for the global enrich_session
    function using dynamic tracer discovery and flexible metadata handling.

    Args:
        session_id: The session ID to enrich
        metadata: Metadata dictionary to add to the session
        tracer: Optional tracer instance to use
        tracer_instance: Optional tracer instance for logging context

    Legacy Example:
        >>> # Using default tracer (backward compatibility)
        >>> enrich_session("session-123", {"user_id": "user-456"})
        >>>
        >>> # Using specific tracer (backward compatibility)
        >>> enrich_session("session-123", {"user_id": "user-456"}, tracer=my_tracer)

    See Also:
        - :meth:`HoneyHiveTracer.enrich_session` - Primary pattern (v1.0+)
        - :meth:`HoneyHiveTracer.enrich_span` - Span enrichment
    """
    # Dynamic tracer discovery
    active_tracer = _discover_tracer_dynamically(tracer, tracer_instance)

    if active_tracer is None:
        safe_log(
            tracer_instance,
            "warning",
            "No tracer available for session enrichment",
            honeyhive_data={
                "session_id": session_id,
                "metadata_keys": list(metadata.keys()) if metadata else [],
            },
        )
        return

    # Dynamic session enrichment
    try:
        _enrich_session_dynamically(
            active_tracer, session_id, metadata, tracer_instance
        )

        safe_log(
            tracer_instance,
            "debug",
            "Session enriched successfully",
            honeyhive_data={
                "session_id": session_id,
                "tracer_type": type(active_tracer).__name__,
                "metadata_count": len(metadata) if metadata else 0,
            },
        )

    except Exception as e:
        safe_log(
            tracer_instance,
            "error",
            "Failed to enrich session",
            honeyhive_data={
                "session_id": session_id,
                "error": str(e),
                "error_type": type(e).__name__,
            },
        )
    return

get_global_provider

get_global_provider(tracer_instance: Any = None) -> Any

Dynamically get the current global OpenTelemetry tracer provider.

Returns:

Type	Description
Any	The current global TracerProvider instance

Example

from honeyhive.tracer.integration import get_global_provider provider = get_global_provider()

Source code in src/honeyhive/tracer/integration/detection.py

def get_global_provider(tracer_instance: Any = None) -> Any:
    """Dynamically get the current global OpenTelemetry tracer provider.

    Returns:
        The current global TracerProvider instance

    Example:
        >>> from honeyhive.tracer.integration import get_global_provider
        >>> provider = get_global_provider()
    """
    provider = trace.get_tracer_provider()

    safe_log(
        tracer_instance,
        "debug",
        "Retrieved global provider",
        honeyhive_data={
            "provider_class": type(provider).__name__,
        },
    )

    return provider

set_global_provider

set_global_provider(
    provider: Any,
    force_override: bool = False,
    tracer_instance: Any = None,
) -> None

Dynamically set the global OpenTelemetry tracer provider.

This function properly handles OpenTelemetry's internal warnings when setting a tracer provider, using dynamic provider management techniques.

Parameters:

Name	Type	Description	Default
provider	Any	The TracerProvider instance to set as global	required
force_override	bool	If True, allows overriding existing real providers (intended for test utilities and clean state management)	False
tracer_instance	Any	Optional tracer instance for logging context	None

Example

from opentelemetry.sdk.trace import TracerProvider from honeyhive.tracer.integration import set_global_provider provider = TracerProvider() set_global_provider(provider)

For test utilities - force override existing provider

set_global_provider(NoOpTracerProvider(), force_override=True)

Source code in src/honeyhive/tracer/integration/detection.py

def set_global_provider(
    provider: Any, force_override: bool = False, tracer_instance: Any = None
) -> None:
    """Dynamically set the global OpenTelemetry tracer provider.

    This function properly handles OpenTelemetry's internal warnings when
    setting a tracer provider, using dynamic provider management techniques.

    Args:
        provider: The TracerProvider instance to set as global
        force_override: If True, allows overriding existing real providers
                       (intended for test utilities and clean state management)
        tracer_instance: Optional tracer instance for logging context

    Example:
        >>> from opentelemetry.sdk.trace import TracerProvider
        >>> from honeyhive.tracer.integration import set_global_provider
        >>> provider = TracerProvider()
        >>> set_global_provider(provider)

        # For test utilities - force override existing provider
        >>> set_global_provider(NoOpTracerProvider(), force_override=True)
    """
    try:
        # Check if a provider is already set to avoid the override warning
        current_provider = trace.get_tracer_provider()
        provider_type = type(current_provider).__name__

        # Determine if we should set the provider
        should_set = False
        reason = ""

        if provider_type in ["NoOpTracerProvider", "ProxyTracerProvider"]:
            # Safe to set as global provider - no real provider exists
            should_set = True
            reason = "no_real_provider_exists"
            # Reset SET_ONCE flag for both NoOp and Proxy providers
            # Both set the flag but should be replaceable
            _reset_provider_flag_dynamically(tracer_instance)
        elif force_override:
            # Force override requested (for test utilities)
            should_set = True
            reason = "force_override_requested"
            # Reset the SET_ONCE flag to allow clean override
            _reset_provider_flag_dynamically(tracer_instance)
        else:
            # Another real provider exists and no force override
            should_set = False
            reason = "real_provider_exists_no_force"

        if should_set:
            _set_tracer_provider(provider, log=False)
            safe_log(
                tracer_instance,
                "debug",
                "Global provider set successfully",
                honeyhive_data={
                    "provider_class": type(provider).__name__,
                    "replaced_provider": provider_type,
                    "reason": reason,
                    "force_override": force_override,
                },
            )
        else:
            # Another real provider is already set, don't override
            safe_log(
                tracer_instance,
                "debug",
                "Real TracerProvider already exists, skipping global provider set",
                honeyhive_data={
                    "existing_provider": provider_type,
                    "requested_provider": type(provider).__name__,
                    "reason": reason,
                    "force_override": force_override,
                },
            )

    except Exception as e:
        safe_log(
            tracer_instance,
            "warning",
            "Failed to set global provider",
            honeyhive_data={
                "provider_class": type(provider).__name__,
                "error": str(e),
            },
        )
        raise

graceful_shutdown_all

graceful_shutdown_all() -> None

Gracefully shutdown all registered tracer instances.

This function attempts to find and shutdown all active HoneyHive tracer instances. It's useful for application shutdown or cleanup scenarios where multiple tracers might be active.

Example:

.. code-block:: python

# Application shutdown
import atexit
atexit.register(graceful_shutdown_all)

# Or explicit cleanup
graceful_shutdown_all()

Note:

This function uses the tracer registry to find active instances. It attempts graceful shutdown but continues even if some instances fail to shutdown properly.

Source code in src/honeyhive/tracer/lifecycle/shutdown.py

def graceful_shutdown_all() -> None:
    """Gracefully shutdown all registered tracer instances.

    This function attempts to find and shutdown all active HoneyHive
    tracer instances. It's useful for application shutdown or cleanup
    scenarios where multiple tracers might be active.

    **Example:**

    .. code-block:: python

        # Application shutdown
        import atexit
        atexit.register(graceful_shutdown_all)

        # Or explicit cleanup
        graceful_shutdown_all()

    **Note:**

    This function uses the tracer registry to find active instances.
    It attempts graceful shutdown but continues even if some instances
    fail to shutdown properly.
    """
    try:
        active_tracers = registry.get_all_tracers()

        if not active_tracers:
            safe_log(None, "debug", "No active tracers found for shutdown")
            return

        safe_log(
            None,
            "info",
            "Starting graceful shutdown of all tracers",
            honeyhive_data={"tracer_count": len(active_tracers)},
        )

        shutdown_results = []

        for tracer_instance in active_tracers:
            try:
                shutdown_tracer(tracer_instance)
                shutdown_results.append(
                    (getattr(tracer_instance, "_tracer_id", "unknown"), True)
                )
                safe_log(
                    tracer_instance,
                    "debug",
                    "Tracer shutdown successful",
                    honeyhive_data={
                        "tracer_id": getattr(tracer_instance, "_tracer_id", "unknown")
                    },
                )
            except Exception as e:
                shutdown_results.append(
                    (getattr(tracer_instance, "_tracer_id", "unknown"), False)
                )
                # Graceful degradation - never crash host
                safe_log(
                    tracer_instance,
                    "error",
                    "Tracer shutdown failed",
                    honeyhive_data={
                        "tracer_id": getattr(tracer_instance, "_tracer_id", "unknown"),
                        "error": str(e),
                        "error_type": type(e).__name__,
                        "operation": "graceful_shutdown_single_tracer",
                    },
                )

        # Log summary
        successful_shutdowns = sum(1 for _, success in shutdown_results if success)
        safe_log(
            None,
            "info",
            "Graceful shutdown completed",
            honeyhive_data={
                "total_tracers": len(active_tracers),
                "successful_shutdowns": successful_shutdowns,
                "failed_shutdowns": len(active_tracers) - successful_shutdowns,
            },
        )

    except Exception as e:
        # Graceful degradation - never crash host
        safe_log(
            None,
            "error",
            "Error during graceful shutdown of all tracers",
            honeyhive_data={
                "error": str(e),
                "error_type": type(e).__name__,
                "operation": "graceful_shutdown_all",
            },
        )

shutdown_tracer

shutdown_tracer(tracer_instance: Any) -> None

Shutdown a tracer instance and clean up its resources.

This function performs a graceful shutdown of a tracer instance, including flushing pending data, shutting down providers, and cleaning up resources. It handles both main and secondary providers appropriately.

:param tracer_instance: The tracer instance to shutdown :type tracer_instance: HoneyHiveTracer

Example:

.. code-block:: python

# Graceful shutdown
shutdown_tracer(tracer)

# In a try/finally block
try:
    # Use tracer
    with tracer.start_span("operation") as span:
        pass
finally:
    shutdown_tracer(tracer)

Note:

This function only shuts down the OpenTelemetry provider if the tracer instance is the main provider. Secondary providers are left running to avoid disrupting other tracer instances.

Source code in src/honeyhive/tracer/lifecycle/shutdown.py

def shutdown_tracer(tracer_instance: Any) -> None:
    """Shutdown a tracer instance and clean up its resources.

    This function performs a graceful shutdown of a tracer instance,
    including flushing pending data, shutting down providers, and
    cleaning up resources. It handles both main and secondary providers
    appropriately.

    :param tracer_instance: The tracer instance to shutdown
    :type tracer_instance: HoneyHiveTracer

    **Example:**

    .. code-block:: python

        # Graceful shutdown
        shutdown_tracer(tracer)

        # In a try/finally block
        try:
            # Use tracer
            with tracer.start_span("operation") as span:
                pass
        finally:
            shutdown_tracer(tracer)

    **Note:**

    This function only shuts down the OpenTelemetry provider if the
    tracer instance is the main provider. Secondary providers are
    left running to avoid disrupting other tracer instances.
    """
    # Check if logging is still available (pytest-xdist workers may have closed streams)
    safe_log(
        tracer_instance, "debug", "shutdown_tracer: Starting data loss prevention phase"
    )

    # Phase 1: Data loss prevention - optimized for parallel execution
    # This ensures we attempt to preserve data even if locking fails
    test_mode = getattr(tracer_instance, "test_mode", False)

    # Skip data loss prevention in test mode to prevent worker conflicts
    # In production, this is critical for data preservation
    if not test_mode:
        # Graceful drain phase (production only)
        # For multi-instance architecture: only disable globally if main provider
        if getattr(tracer_instance, "is_main_provider", False):
            disable_new_span_creation()

        # Always set instance-specific shutdown flag for this tracer
        # Protected access required for multi-instance lifecycle management
        tracer_instance._instance_shutdown = True  # pylint: disable=protected-access

        # Brief grace period for existing spans to complete naturally
        time.sleep(0.1)

        # Force flush with extended timeout and retry logic (before lock acquisition)
        timeout_ms = 5000  # Extended timeout for production

        safe_log(
            tracer_instance,
            "debug",
            "Starting pre-lock force flush for data loss prevention",
            honeyhive_data={
                "timeout_ms": timeout_ms,
                "test_mode": test_mode,
                "phase": "pre_lock_data_preservation",
            },
        )

        flush_success = force_flush_tracer(tracer_instance, timeout_millis=timeout_ms)

        # Retry logic for critical data preservation (production only)
        if not flush_success:
            safe_log(
                tracer_instance,
                "warning",
                f"Pre-lock flush failed (timeout: {timeout_ms}ms), retrying",
            )

            retry_timeout_ms = timeout_ms * 2
            flush_success = force_flush_tracer(
                tracer_instance, timeout_millis=retry_timeout_ms
            )

            if flush_success:
                safe_log(
                    tracer_instance,
                    "info",
                    f"Pre-lock flush succeeded on retry ({retry_timeout_ms}ms)",
                )
            else:
                safe_log(
                    tracer_instance,
                    "error",
                    f"Pre-lock flush failed after retry ({retry_timeout_ms}ms), "
                    "continuing with shutdown - potential data loss",
                )
    else:
        # Test mode: skip pre-lock flush to prevent pytest-xdist worker conflicts
        safe_log(
            tracer_instance,
            "debug",
            "Skipping pre-lock flush in test mode to prevent conflicts",
        )
        flush_success = True  # Assume success for test mode

    safe_log(
        tracer_instance,
        "debug",
        "shutdown_tracer: Acquiring _lifecycle_lock for shutdown",
    )

    # Use environment-optimized lock timeout for better performance
    # Automatically detects Lambda, K8s, high-concurrency environments
    with acquire_lifecycle_lock_optimized("lifecycle") as lock_acquired:
        if not lock_acquired:
            # Graceful degradation: Try to log timeout but don't crash
            config = get_lock_config()
            timeout_used = config.get("lifecycle_timeout", 1.0)
            safe_log(
                tracer_instance,
                "warning",
                f"Failed to acquire _lifecycle_lock within {timeout_used}s, "
                "proceeding without lock",
                honeyhive_data={
                    "lock_timeout": timeout_used,
                    "lock_strategy": config.get("description", "unknown"),
                    "degradation_reason": "lock_acquisition_timeout",
                    "data_flush_completed": flush_success,
                },
            )
            # Continue without the lock - better than hanging indefinitely
            _shutdown_without_lock(tracer_instance)
            return

        try:
            safe_log(
                tracer_instance,
                "debug",
                "Starting tracer shutdown",
                honeyhive_data={
                    "is_main_provider": tracer_instance.is_main_provider,
                    "has_provider": bool(tracer_instance.provider),
                },
            )

            # Skip force_flush during shutdown to prevent recursive deadlock
            # The force_flush_tracer also tries to acquire _lifecycle_lock,
            # causing deadlock
            safe_log(
                tracer_instance,
                "debug",
                "Skipping force_flush during shutdown to prevent recursive deadlock",
            )

            # Only shutdown if we're the main provider
            if (
                tracer_instance.is_main_provider
                and tracer_instance.provider
                and hasattr(tracer_instance.provider, "shutdown")
            ):
                _shutdown_main_provider(tracer_instance)
            else:
                _cleanup_secondary_provider(tracer_instance)

            # Clean up instance state
            _cleanup_tracer_state(tracer_instance)

        except Exception as e:
            # Graceful degradation - never crash host
            safe_log(
                tracer_instance,
                "error",
                "Error during tracer shutdown",
                honeyhive_data={
                    "error": str(e),
                    "error_type": type(e).__name__,
                    "operation": "tracer_shutdown",
                },
            )

flush

flush(
    tracer_instance: Any, timeout_millis: float = 30000
) -> bool

Force flush any pending spans and data for a tracer instance.

This function ensures that all pending spans and telemetry data are immediately sent to their destinations, rather than waiting for automatic batching/flushing. It handles multiple flush targets and provides comprehensive error handling.

:param tracer_instance: The tracer instance to flush :type tracer_instance: HoneyHiveTracer :param timeout_millis: Maximum time to wait for flush completion in milliseconds :type timeout_millis: float :return: True if flush completed successfully within timeout, False otherwise :rtype: bool

Example:

.. code-block:: python

# Flush with default timeout (30 seconds)
success = force_flush_tracer(tracer)

# Flush with custom timeout (5 seconds)
success = force_flush_tracer(tracer, timeout_millis=5000)

# Use before critical sections
if force_flush_tracer(tracer):
    print("All spans flushed successfully")
else:
    print("Flush timeout or error occurred")

Note:

This function attempts to flush multiple components in sequence: the tracer provider, custom span processors, and batch processors. It returns True only if all components flush successfully.

Source code in src/honeyhive/tracer/lifecycle/flush.py

def force_flush_tracer(tracer_instance: Any, timeout_millis: float = 30000) -> bool:
    """Force flush any pending spans and data for a tracer instance.

    This function ensures that all pending spans and telemetry data are
    immediately sent to their destinations, rather than waiting for
    automatic batching/flushing. It handles multiple flush targets and
    provides comprehensive error handling.

    :param tracer_instance: The tracer instance to flush
    :type tracer_instance: HoneyHiveTracer
    :param timeout_millis: Maximum time to wait for flush completion in milliseconds
    :type timeout_millis: float
    :return: True if flush completed successfully within timeout, False otherwise
    :rtype: bool

    **Example:**

    .. code-block:: python

        # Flush with default timeout (30 seconds)
        success = force_flush_tracer(tracer)

        # Flush with custom timeout (5 seconds)
        success = force_flush_tracer(tracer, timeout_millis=5000)

        # Use before critical sections
        if force_flush_tracer(tracer):
            print("All spans flushed successfully")
        else:
            print("Flush timeout or error occurred")

    **Note:**

    This function attempts to flush multiple components in sequence:
    the tracer provider, custom span processors, and batch processors.
    It returns True only if all components flush successfully.
    """
    safe_log(tracer_instance, "debug", "Force flush requested")

    flush_results: List[Tuple[str, bool]] = []

    safe_log(
        tracer_instance,
        "debug",
        f"force_flush_tracer: Acquiring _lifecycle_lock (timeout: {timeout_millis}ms)",
    )

    try:
        safe_log(
            tracer_instance,
            "debug",
            "force_flush_tracer: Attempting to acquire _lifecycle_lock...",
        )
        # Use environment-optimized flush timeout
        flush_timeout_seconds = timeout_millis / 1000.0
        with acquire_lifecycle_lock_optimized(
            "flush", custom_timeout=flush_timeout_seconds
        ) as acquired:
            if not acquired:
                safe_log(
                    tracer_instance,
                    "warning",
                    f"Failed to acquire _lifecycle_lock ({flush_timeout_seconds}s)",
                )
                return False
            safe_log(
                tracer_instance,
                "debug",
                "force_flush_tracer: Successfully acquired _lifecycle_lock",
            )
            # 1. Flush the tracer provider if available and supports it
            _flush_tracer_provider(tracer_instance, timeout_millis, flush_results)

            # 2. Flush our custom span processor if available
            _flush_span_processor(tracer_instance, timeout_millis, flush_results)

            # 3. Flush any batch span processors attached to the provider
            _flush_batch_processors(tracer_instance, timeout_millis, flush_results)

        # Calculate overall result
        overall_success = all(result for _, result in flush_results)

        _log_flush_results(tracer_instance, overall_success, flush_results)

        return overall_success

    except Exception as e:
        # Graceful degradation - never crash host
        safe_log(
            tracer_instance,
            "error",
            "Force flush failed",
            honeyhive_data={
                "error": str(e),
                "error_type": type(e).__name__,
                "operation": "force_flush_tracer",
            },
        )
        return False

clear_registry

clear_registry() -> None

Clear all registered tracers and default tracer.

This is primarily useful for testing and cleanup scenarios.

Warning

This will break any ongoing tracing operations that depend on auto-discovery. Use with caution.

Example

In test teardown:

clear_registry()

Source code in src/honeyhive/tracer/registry.py

def clear_registry() -> None:
    """Clear all registered tracers and default tracer.

    This is primarily useful for testing and cleanup scenarios.

    Warning:
        This will break any ongoing tracing operations that depend
        on auto-discovery. Use with caution.

    Example:
        >>> # In test teardown:
        >>> clear_registry()
    """
    global _DEFAULT_TRACER
    _TRACER_REGISTRY.clear()
    _DEFAULT_TRACER = None

get_default_tracer

get_default_tracer() -> Optional[HoneyHiveTracer]

Get the global default tracer if set and still alive.

Returns:

Type	Description
Optional[HoneyHiveTracer]	Default HoneyHiveTracer instance if set and not garbage collected,
Optional[HoneyHiveTracer]	None otherwise

Example

tracer = get_default_tracer() if tracer: ... print(f"Default tracer project: {tracer.project}")

Source code in src/honeyhive/tracer/registry.py

def get_default_tracer() -> "Optional[HoneyHiveTracer]":
    """Get the global default tracer if set and still alive.

    Returns:
        Default HoneyHiveTracer instance if set and not garbage collected,
        None otherwise

    Example:
        >>> tracer = get_default_tracer()
        >>> if tracer:
        ...     print(f"Default tracer project: {tracer.project}")
    """
    global _DEFAULT_TRACER

    # PYTEST-XDIST COMPATIBLE: No cross-process locks needed
    if _DEFAULT_TRACER is not None:
        # Weak reference - check if still alive
        tracer = _DEFAULT_TRACER()
        if tracer is not None:
            return tracer
        # Tracer was garbage collected, clear the reference
        _DEFAULT_TRACER = None

    return None

set_default_tracer

set_default_tracer(
    tracer: Optional[HoneyHiveTracer],
) -> None

Set a global default tracer for backward compatibility.

This tracer will be used as a fallback when no tracer is found in baggage context and no explicit tracer is provided.

Parameters:

Name	Type	Description	Default
tracer	Optional[HoneyHiveTracer]	HoneyHiveTracer instance to use as default, or None to clear the default	required

Example

default_tracer = HoneyHiveTracer(api_key="your-api-key") # auto project set_default_tracer(default_tracer)

Now @trace will use default_tracer when no context available

@trace ... def my_function(): ... pass

Source code in src/honeyhive/tracer/registry.py

def set_default_tracer(tracer: "Optional[HoneyHiveTracer]") -> None:
    """Set a global default tracer for backward compatibility.

    This tracer will be used as a fallback when no tracer is found
    in baggage context and no explicit tracer is provided.

    Args:
        tracer: HoneyHiveTracer instance to use as default,
                or None to clear the default

    Example:
        >>> default_tracer = HoneyHiveTracer(api_key="your-api-key")  # auto project
        >>> set_default_tracer(default_tracer)
        >>>
        >>> # Now @trace will use default_tracer when no context available
        >>> @trace
        ... def my_function():
        ...     pass
    """
    global _DEFAULT_TRACER

    # PYTEST-XDIST COMPATIBLE: No cross-process locks needed
    if tracer is None:
        _DEFAULT_TRACER = None
    else:
        # Register the tracer to ensure it's in the registry
        register_tracer(tracer)
        _DEFAULT_TRACER = weakref.ref(tracer)