"""OpenTelemetry tracing for MCP Hangar.

Provides distributed tracing with automatic context propagation
through tool invocations and mcp_server calls.

Configuration via environment variables:
    OTEL_EXPORTER_OTLP_[TRACES_]PROTOCOL: grpc (default) or http/protobuf
    OTEL_EXPORTER_OTLP_[TRACES_]ENDPOINT, _INSECURE, _HEADERS and the other
        standard exporter variables: read by the OpenTelemetry SDK itself
    OTEL_SERVICE_NAME: Service name (default: mcp-hangar)
    OTEL_TRACES_SAMPLER: Sampler type (default: always_on)
    MCP_TRACING_ENABLED: Enable/disable tracing (default: true)
    MCP_SPAN_ATTRIBUTE_LENGTH_LIMIT: Longest attribute value on Hangar's own
        provider, in characters (default: 256). The OTEL_*_LENGTH_LIMIT
        variables win when set: see _span_limits().

OTLP trace exporter precedence, first match wins (resolve_otlp_exporter_settings):
    protocol: OTEL_EXPORTER_OTLP_TRACES_PROTOCOL, OTEL_EXPORTER_OTLP_PROTOCOL,
        then grpc. Any other value adds no OTLP exporter and logs why.
    endpoint: OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, OTEL_EXPORTER_OTLP_ENDPOINT,
        then Hangar's own (``observability.tracing.otlp_endpoint`` in
        config.yaml, or ``init_tracing(otlp_endpoint=...)``), then the SDK
        default: http://localhost:4317 for grpc, http://localhost:4318/v1/traces
        for http/protobuf. The environment beats config.yaml, as everywhere in
        the bootstrap. An empty OTEL_EXPORTER_OTLP_ENDPOINT adds no OTLP exporter.
    TLS: https:// always uses TLS. For grpc otherwise
        OTEL_EXPORTER_OTLP_TRACES_INSECURE, OTEL_EXPORTER_OTLP_INSECURE, then
        the scheme: http:// is plaintext (so both defaults are), a scheme-less
        endpoint uses TLS. For http/protobuf the scheme alone decides.

Example:
    from mcp_hangar.observability.tracing import init_tracing, get_tracer

    # Initialize once at startup
    init_tracing()

    # Get tracer for module
    tracer = get_tracer(__name__)

    # Create spans
    with tracer.start_as_current_span("my_operation") as span:
        span.set_attribute("key", "value")
        do_work()
"""

from collections.abc import Callable, Iterator
from contextlib import contextmanager
from dataclasses import dataclass
import os
import sys
import threading
from typing import Any, TypeVar

from mcp_hangar.errors import bounded_error_type
from mcp_hangar.logging_config import env_length_limit, get_logger
from mcp_hangar.metrics import record_otlp_export_failure
from mcp_hangar.observability.conventions import GenAI, MCP

logger = get_logger(__name__)

# Type variable for generic decorator
F = TypeVar("F", bound=Callable[..., Any])

# Global state. `_initialized` means Hangar's own provider is the registered one.
_tracer_mcp_server: Any = None  # an SDK TracerProvider; the SDK may be absent
_initialized = False
# Set once Hangar has shut its own provider down. The API registers a global
# provider once per process, so the shut-down one stays registered for good.
_shut_down = False

# Upper bound on shutdown_tracing(). The SDK's shutdown waits for an export in
# flight, which against an unreachable collector is the whole OTLP export
# timeout (10 s by default) per processor, and it takes no deadline of its own.
TRACING_SHUTDOWN_TIMEOUT_S = 5.0

#: Hangar's bound on an attribute value on its own provider, in characters; see _span_limits().
SPAN_ATTRIBUTE_LENGTH_LIMIT = 256
SPAN_ATTRIBUTE_LENGTH_LIMIT_ENV = "MCP_SPAN_ATTRIBUTE_LENGTH_LIMIT"

# Check if OpenTelemetry is available
try:
    from opentelemetry.sdk.resources import OTELResourceDetector, Resource, SERVICE_NAME
    from opentelemetry.sdk.trace.export import (
        BatchSpanProcessor,
        ConsoleSpanExporter,
        SpanExporter,
        SpanExportResult,
    )
    from opentelemetry.sdk.trace import SpanLimits, TracerProvider
    from opentelemetry.sdk.trace.sampling import (
        ALWAYS_OFF,
        ALWAYS_ON,
        ParentBased,
        TraceIdRatioBased,
    )
    from opentelemetry import trace
    from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
    from opentelemetry.trace import Status, StatusCode

    OTEL_AVAILABLE = True
except ImportError:
    OTEL_AVAILABLE = False
    trace = None  # type: ignore[assignment]

# The OTLP/gRPC span exporter, the default protocol's. OTLP/HTTP is imported
# only when selected: see _build_otlp_span_exporter().
try:
    from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter as _GrpcSpanExporter

    OTLPSpanExporter: Any = _GrpcSpanExporter
    OTLP_AVAILABLE = True
except ImportError:
    OTLP_AVAILABLE = False
    OTLPSpanExporter = None

# The OTLP protocols Hangar builds a span exporter for, and the package each needs.
_OTLP_SPAN_EXPORTER_PACKAGES = {
    "grpc": "opentelemetry-exporter-otlp-proto-grpc",
    "http/protobuf": "opentelemetry-exporter-otlp-proto-http",
}

# Try to import Jaeger exporter
try:
    from opentelemetry.exporter.jaeger.thrift import JaegerExporter

    JAEGER_AVAILABLE = True
except ImportError:
    JAEGER_AVAILABLE = False
    JaegerExporter = None


if OTEL_AVAILABLE:

    class _MeteredSpanExporter(SpanExporter):
        """Wrap a SpanExporter to make export failures observable.

        ``BatchSpanProcessor`` calls ``export()`` on a background thread and
        swallows failures, so an unreachable collector is silent and the spans
        buffered in that batch are dropped without a metric. This decorator
        increments ``mcp_hangar_otlp_export_failures_total`` when the wrapped
        exporter returns a failure result or raises. It never changes export
        semantics: the inner result (or exception) is propagated unchanged, so
        the SDK's own retry/backoff behaviour is preserved and the MCP path is
        never blocked.
        """

        def __init__(self, inner: SpanExporter) -> None:
            self._inner = inner

        def export(self, spans: Any) -> "SpanExportResult":
            try:
                result = self._inner.export(spans)
            except Exception:
                record_otlp_export_failure()
                raise
            if result is not SpanExportResult.SUCCESS:
                record_otlp_export_failure()
            return result

        def shutdown(self) -> None:
            self._inner.shutdown()

        def force_flush(self, timeout_millis: int = 30000) -> bool:
            return bool(self._inner.force_flush(timeout_millis))


class NoOpSpan:
    """No-op span for when tracing is disabled."""

    def is_recording(self) -> bool:
        return False

    def set_attribute(self, key: str, value: Any) -> None:
        pass

    def set_status(self, status: Any) -> None:
        pass

    def record_exception(self, exception: Exception) -> None:
        pass

    def add_event(self, name: str, attributes: dict | None = None) -> None:
        pass

    def __enter__(self) -> "NoOpSpan":
        return self

    def __exit__(self, *args) -> None:
        pass


class NoOpTracer:
    """No-op tracer for when tracing is disabled."""

    def start_as_current_span(self, name: str, **kwargs) -> NoOpSpan:
        return NoOpSpan()

    @contextmanager
    def start_span(self, name: str, **kwargs):
        yield NoOpSpan()


_noop_tracer = NoOpTracer()


class _TextFreeTracer:
    """The tracer every Hangar span comes from: no exception text reaches a span.

    By default the SDK records an exception that escapes a span as an
    ``exception`` event, message and stacktrace included, and sets the status
    description to ``"<type>: <message>"``. An exception's message can carry
    what an upstream tool returned (GHSA-qwq2-7g49-jxc6), so both are off here,
    whichever way the span is started. A span an exception escapes still ends
    in ERROR, with an ``exception`` event that carries only ``exception.type``.
    Its ``error.type`` is the exception's qualified class name, unless
    something closer to the failure already set one. That is the shape the mcp
    SDK server middleware gives its own spans.
    """

    def __init__(self, tracer: Any) -> None:
        self._tracer = tracer

    @contextmanager
    def start_as_current_span(self, name: str, **kwargs: Any) -> Iterator[Any]:
        kwargs.update(record_exception=False, set_status_on_exception=False)
        with self._tracer.start_as_current_span(name, **kwargs) as span:
            try:
                yield span
            except Exception as error:
                mark_span_error(span)
                _set_error_type_if_absent(span, type(error).__qualname__)
                _record_exception_type(span, error)
                raise

    def start_span(self, name: str, **kwargs: Any) -> Any:
        """A span with the SDK's exception recording off; its caller records the outcome."""
        kwargs.update(record_exception=False, set_status_on_exception=False)
        return self._tracer.start_span(name, **kwargs)


# W3C baggage (GHSA-qwq2-7g49-jxc6).
#
# Baggage is a set of opaque key/value pairs, and nothing on the wire says who
# set an entry. Hangar sets none itself, so no entry in an inbound carrier or in
# the ambient context can be attributed to it. A key prefix is not provenance:
# any caller can write one. Hangar therefore extracts no baggage from inbound
# carriers and forwards none upstream, whoever attached it to the context (host
# auto-instrumentation, an embedding application). It propagates only W3C
# trace context (traceparent/tracestate).
BAGGAGE_HEADER = "baggage"


def _get_propagator() -> Any:
    """The propagator for inbound and outbound carriers: W3C TraceContext only.

    ``traceparent``/``tracestate`` are structural trace identifiers. Baggage is
    deliberately absent; see the note above.
    """
    return TraceContextTextMapPropagator()


# Set by disable_tracing(): the operator turned tracing off in configuration.
_disabled = False


def is_tracing_enabled() -> bool:
    """Check if tracing is enabled."""
    enabled = os.getenv("MCP_TRACING_ENABLED", "true").lower()
    return not _disabled and enabled in ("true", "1", "yes") and OTEL_AVAILABLE


def disable_tracing() -> None:
    """Turn Hangar's tracing off for the process, as configuration asked.

    The env var alone reaches is_tracing_enabled(); a config file does not, so
    the bootstrap calls this. Without it a provider registered by someone else
    would still carry Hangar's spans and trace context.
    """
    global _disabled
    _disabled = True


def _build_sampler() -> Any:
    """Build a sampler from OTEL_TRACES_SAMPLER / OTEL_TRACES_SAMPLER_ARG.

    We construct the TracerProvider by hand, so the SDK's standard env-var
    auto-configuration for sampling never runs. This mirrors that contract so
    OTEL_TRACES_SAMPLER actually takes effect (the docstring long claimed
    support that was never wired). Defaults to parentbased_always_on, matching
    the SDK default. For ratio samplers, OTEL_TRACES_SAMPLER_ARG is the ratio
    in [0, 1]; unset it is 1.0 (sample everything), as in the SDK. Any other
    value logs one warning and is 1.0 too, rather than raising in
    TraceIdRatioBased and so turning tracing off for the whole process.
    """
    name = os.getenv("OTEL_TRACES_SAMPLER", "parentbased_always_on").strip().lower()
    arg = os.getenv("OTEL_TRACES_SAMPLER_ARG", "")

    def _ratio() -> float:
        try:
            ratio = float(arg) if arg.strip() else 1.0
        except ValueError:
            ratio = float("nan")
        if 0.0 <= ratio <= 1.0:  # False for NaN, which the SDK's own range check lets through
            return ratio
        logger.warning(
            "tracing_sampler_arg_invalid",
            variable="OTEL_TRACES_SAMPLER_ARG",
            value=arg,
            sampler=name,
            expected="a number in [0, 1]",
            fallback=1.0,
        )
        return 1.0

    if name == "always_on":
        return ALWAYS_ON
    if name == "always_off":
        return ALWAYS_OFF
    if name == "traceidratio":
        return TraceIdRatioBased(_ratio())
    if name == "parentbased_always_off":
        return ParentBased(ALWAYS_OFF)
    if name == "parentbased_traceidratio":
        return ParentBased(TraceIdRatioBased(_ratio()))
    if name != "parentbased_always_on":
        logger.warning("tracing_unknown_sampler", sampler=name, fallback="parentbased_always_on")
    return ParentBased(ALWAYS_ON)


def _span_limits() -> Any:
    """The SpanLimits of Hangar's own provider: attribute values bounded, the SDK's variables first.

    Per value, first match wins:

    span attributes: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT,
        OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT, MCP_SPAN_ATTRIBUTE_LENGTH_LIMIT, 256.
    span event and link attributes, an exception's message and stack trace
        among them: OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT,
        MCP_SPAN_ATTRIBUTE_LENGTH_LIMIT, 256.

    The SDK reads the OTEL_* variables itself, so Hangar passes its own value
    only while the global one is unset or empty. The SDK cuts a value to the
    limit and has no hook to mark the cut: span attributes carry no marker.
    """
    if os.getenv("OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT", "").strip():
        return SpanLimits()
    return SpanLimits(
        max_attribute_length=env_length_limit(SPAN_ATTRIBUTE_LENGTH_LIMIT_ENV) or SPAN_ATTRIBUTE_LENGTH_LIMIT
    )


@dataclass(frozen=True)
class OtlpExporterSettings:
    """What Hangar passes an SDK OTLP exporter. None leaves a value to the SDK."""

    protocol: str
    endpoint: str | None
    insecure: bool | None


def resolve_otlp_exporter_settings(
    signal: str = "traces",
    endpoint: str | None = None,
    insecure: bool | None = None,
) -> OtlpExporterSettings:
    """Effective settings for one signal's OTLP exporter, env over Hangar's own.

    ``endpoint`` and ``insecure`` are Hangar's configuration (config.yaml or an
    argument); ``signal`` names the ``OTEL_EXPORTER_OTLP_<SIGNAL>_*`` family
    (``traces``, ``logs``). The environment beats Hangar's configuration, and
    the signal-specific variable the generic one. First match wins:

    protocol: ``OTEL_EXPORTER_OTLP_<SIGNAL>_PROTOCOL``,
        ``OTEL_EXPORTER_OTLP_PROTOCOL``, ``grpc``. Lower-cased, not validated:
        the caller rejects what it cannot build.
    endpoint: ``OTEL_EXPORTER_OTLP_<SIGNAL>_ENDPOINT``,
        ``OTEL_EXPORTER_OTLP_ENDPOINT``, ``endpoint``, the SDK default. While
        either variable is set this is None and the SDK reads them itself, in
        that order, adding ``/v1/<signal>`` to the generic one for
        http/protobuf as the spec requires. ``""`` -- an empty ``endpoint``, or
        an empty generic variable and no signal one -- means no OTLP exporter.
    insecure (gRPC only): ``OTEL_EXPORTER_OTLP_<SIGNAL>_INSECURE``,
        ``OTEL_EXPORTER_OTLP_INSECURE``, ``insecure`` (only with the
        ``endpoint`` it came with), then the SDK's rule: ``http://`` is
        plaintext, a scheme-less endpoint uses TLS. The SDK uses TLS for
        ``https://`` whatever this returns.

    An empty variable counts as unset, the generic endpoint aside.
    """
    signal_var = f"OTEL_EXPORTER_OTLP_{signal.upper()}_"
    env = os.environ
    protocol = env.get(signal_var + "PROTOCOL") or env.get("OTEL_EXPORTER_OTLP_PROTOCOL") or "grpc"
    if env.get(signal_var + "ENDPOINT"):
        endpoint = insecure = None
    elif "OTEL_EXPORTER_OTLP_ENDPOINT" in env:
        endpoint = None if env["OTEL_EXPORTER_OTLP_ENDPOINT"].strip() else ""
        insecure = None
    if env.get(signal_var + "INSECURE") or env.get("OTEL_EXPORTER_OTLP_INSECURE"):
        insecure = None
    return OtlpExporterSettings(protocol.strip().lower(), endpoint, insecure)


def _otlp_http_span_exporter_class() -> Any:
    """The SDK's OTLP/HTTP span exporter class, or None when its package is absent."""
    try:
        from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as OTLPHttpSpanExporter
    except ImportError:
        return None
    return OTLPHttpSpanExporter


def _build_otlp_span_exporter(settings: OtlpExporterSettings) -> Any:
    """The SDK span exporter ``settings`` select, or None with the reason logged.

    Logs the protocol, never the endpoint (a URL may carry userinfo) nor any
    header (OTEL_EXPORTER_OTLP_HEADERS carries credentials). Building an
    exporter connects to nothing, so it says nothing about delivery.
    """
    if settings.endpoint == "":
        logger.info("tracing_otlp_exporter_skipped", reason="empty_endpoint")
        return None
    package = _OTLP_SPAN_EXPORTER_PACKAGES.get(settings.protocol)
    if package is None:
        logger.warning(
            "tracing_otlp_exporter_unavailable",
            reason="unsupported_protocol",
            protocol=settings.protocol,
            supported=sorted(_OTLP_SPAN_EXPORTER_PACKAGES),
        )
        return None
    if settings.protocol == "grpc":
        exporter_class = OTLPSpanExporter if OTLP_AVAILABLE else None
        kwargs: dict[str, Any] = {"endpoint": settings.endpoint, "insecure": settings.insecure}
    else:
        exporter_class = _otlp_http_span_exporter_class()
        kwargs = {"endpoint": settings.endpoint}  # TLS follows the URL scheme
    if exporter_class is None:
        logger.warning(
            "tracing_otlp_exporter_unavailable",
            reason="package_missing",
            protocol=settings.protocol,
            package=package,
        )
        return None
    exporter = exporter_class(**kwargs)
    logger.info(
        "tracing_otlp_exporter_added",
        protocol=settings.protocol,
        endpoint_from="hangar_config" if settings.endpoint else "otel_env_or_sdk_default",
    )
    return exporter


def _build_resource(service_name: str, service_instance_id: str | None) -> Any:
    """The provider's resource: Hangar's own attributes, the environment's over them.

    ``Resource.create`` merges its detectors, the one reading
    OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME included, underneath the
    attributes it is given, so on its own Hangar's values beat the operator's.
    The environment is merged again on top. Per attribute, first match wins:

    service.name: OTEL_SERVICE_NAME, ``service.name`` in
        OTEL_RESOURCE_ATTRIBUTES (those two in the SDK's own order),
        ``service_name`` (the bootstrap passes config.yaml's), ``mcp-hangar``.
    deployment.environment: OTEL_RESOURCE_ATTRIBUTES, MCP_ENVIRONMENT,
        ``development``.
    service.instance.id: OTEL_RESOURCE_ATTRIBUTES, ``service_instance_id``,
        then the SDK's own detector (SDK 1.44 mints a random UUID), if any.
    service.version and any other key: OTEL_RESOURCE_ATTRIBUTES, then Hangar's.
    """
    attributes: dict[str, Any] = {
        SERVICE_NAME: service_name,
        "service.version": _get_version(),
        "deployment.environment": os.getenv("MCP_ENVIRONMENT", "development"),
    }
    if service_instance_id:
        attributes["service.instance.id"] = service_instance_id
    return Resource.create(attributes).merge(OTELResourceDetector().detect())


def init_tracing(
    service_name: str = "mcp-hangar",
    otlp_endpoint: str | None = None,
    jaeger_host: str | None = None,
    jaeger_port: int = 6831,
    console_export: bool = False,
    service_instance_id: str | None = None,
) -> bool:
    """Initialize OpenTelemetry tracing.

    Args:
        service_name: Service name for traces. OTEL_SERVICE_NAME and a
            ``service.name`` in OTEL_RESOURCE_ATTRIBUTES beat it; see
            _build_resource() for every resource attribute's precedence.
        otlp_endpoint: Hangar's OTLP collector endpoint. An
            OTEL_EXPORTER_OTLP_[TRACES_]ENDPOINT variable beats it; see the
            module docstring for the protocol and TLS.
        jaeger_host: Jaeger agent host for UDP export.
        jaeger_port: Jaeger agent port.
        console_export: Enable console span export (for debugging).
        service_instance_id: ``service.instance.id``, unless
            OTEL_RESOURCE_ATTRIBUTES sets one. The bootstrap passes
            ``current_instance_id()``, the ``produced_by`` of domain events.
            None leaves it to the SDK.

    Returns:
        True if Hangar registered its own provider with at least one exporter,
        False otherwise. Not proof of delivery: no collector has been contacted.
    """
    global _tracer_mcp_server, _initialized

    if _initialized:
        logger.debug("tracing_already_initialized")
        return True

    if _shut_down:
        logger.warning("tracing_init_refused", reason="already_shut_down")
        return False

    if not OTEL_AVAILABLE:
        logger.info(
            "tracing_disabled_otel_not_available",
            hint="Install opentelemetry-api and opentelemetry-sdk",
        )
        return False

    if not is_tracing_enabled():
        logger.info("tracing_disabled_by_config")
        return False

    if _provider_registered_elsewhere():
        # Someone else owns the global: use it, build nothing, claim nothing.
        logger.info("tracing_external_provider_in_use", provider=type(trace.get_tracer_provider()).__name__)
        return False

    try:
        resource = _build_resource(service_name, service_instance_id)

        # Create tracer mcp_server. Held locally until registered: a provider
        # the API refused to register must not end up in module state.
        sampler = _build_sampler()
        provider = TracerProvider(resource=resource, sampler=sampler, span_limits=_span_limits())
        logger.info("tracing_sampler_configured", sampler=type(sampler).__name__)

        # Add exporters
        exporters: list[str] = []  # by name, as the init log lists them

        # OTLP exporter (preferred): protocol, endpoint and TLS from one effective config.
        otlp_settings = resolve_otlp_exporter_settings("traces", otlp_endpoint)
        try:
            otlp_exporter = _build_otlp_span_exporter(otlp_settings)
        except Exception as e:  # noqa: BLE001 -- fault-barrier: exporter init must not crash tracing setup
            otlp_exporter = None
            logger.warning("tracing_otlp_exporter_failed", protocol=otlp_settings.protocol, error=str(e))
        if otlp_exporter is not None:
            provider.add_span_processor(BatchSpanProcessor(_MeteredSpanExporter(otlp_exporter)))
            exporters.append("otlp_" + otlp_settings.protocol.split("/")[0])  # otlp_grpc, otlp_http

        # Jaeger exporter (fallback)
        if JAEGER_AVAILABLE and jaeger_host:
            try:
                jaeger_exporter = JaegerExporter(
                    agent_host_name=jaeger_host,
                    agent_port=jaeger_port,
                )
                provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
                exporters.append("jaeger")
                logger.info(
                    "tracing_jaeger_exporter_added",
                    host=jaeger_host,
                    port=jaeger_port,
                )
            except Exception as e:  # noqa: BLE001 -- fault-barrier: exporter init must not crash tracing setup
                logger.warning("tracing_jaeger_exporter_failed", error=str(e))

        # Console exporter (debugging). stderr, not the SDK's default stdout: on
        # the stdio transport stdout is the JSON-RPC stream.
        if console_export:
            console_exporter = ConsoleSpanExporter(out=sys.stderr)
            provider.add_span_processor(BatchSpanProcessor(console_exporter))
            exporters.append("console")
            logger.info("tracing_console_exporter_added")

        if not exporters:
            logger.warning("tracing_no_exporters_configured")
            return False

        # Register the global tracer provider (third-party OTel API). A second
        # registration is refused with only a warning, so confirm this one took:
        # another component may have registered since the check above.
        trace.set_tracer_provider(provider)
        if trace.get_tracer_provider() is not provider:
            provider.shutdown()
            logger.warning("tracing_init_refused", reason="provider_registered_concurrently")
            return False
        _tracer_mcp_server = provider
        _initialized = True

        # The one init line, here because only this function knows what it
        # attached. Exporter names only: never an endpoint (a URL may carry
        # userinfo) nor a header (credentials).
        logger.info("tracing_initialized", service_name=service_name, exporters=exporters)
        return True

    except Exception as e:  # noqa: BLE001 -- fault-barrier: tracing init failure must not crash application
        logger.error("tracing_initialization_failed", error=str(e))
        return False


def shutdown_tracing() -> None:
    """Shut down Hangar's own tracer provider, flushing pending spans.

    A provider someone else registered is left to its owner. Safe to call twice.
    Returns within ``TRACING_SHUTDOWN_TIMEOUT_S``: the flush runs on a daemon
    thread, and one still waiting on an unreachable collector is abandoned.
    """
    global _tracer_mcp_server, _initialized, _shut_down

    provider = _tracer_mcp_server
    if provider is None:
        return
    _tracer_mcp_server = None
    _initialized = False
    _shut_down = True

    errors: list[Exception] = []

    def _shutdown() -> None:
        try:
            provider.shutdown()
        except Exception as e:  # noqa: BLE001 -- fault-barrier: tracing shutdown must not crash application
            errors.append(e)

    worker = threading.Thread(target=_shutdown, name="hangar-tracing-shutdown", daemon=True)
    worker.start()
    worker.join(TRACING_SHUTDOWN_TIMEOUT_S)
    if worker.is_alive():
        logger.warning("tracing_shutdown_timed_out", timeout_s=TRACING_SHUTDOWN_TIMEOUT_S)
    elif errors:
        logger.warning("tracing_shutdown_error", error=str(errors[0]))
    else:
        logger.info("tracing_shutdown_complete")


def _provider_registered_elsewhere() -> bool:
    """Whether a provider Hangar did not register owns the OTel global.

    Until anything registers, ``get_tracer_provider()`` returns the API's
    ``ProxyTracerProvider`` singleton -- a public class, though absent from the
    API's ``__all__``; the lifecycle tests pin it. Registration is one-shot, so
    any other type is registered for good. ``OTEL_PYTHON_TRACER_PROVIDER``, when
    set, is loaded and registered by that first lookup: someone else's too.
    Only meaningful while Hangar's own provider is not registered.
    """
    return not isinstance(trace.get_tracer_provider(), trace.ProxyTracerProvider)


def _tracing_active() -> bool:
    """Whether Hangar's spans and trace context go anywhere.

    Yes while Hangar's own provider is registered, or when another was
    registered first -- by the host application or an instrumentation agent --
    which Hangar then uses without claiming. Nothing registered returns without
    allocating. After Hangar shut its provider down, that dead provider is the
    global for the rest of the process, so tracing stays off.
    """
    if not OTEL_AVAILABLE:
        return False
    if _initialized:
        return True
    return not _shut_down and _provider_registered_elsewhere() and is_tracing_enabled()


def get_tracer(name: str = __name__) -> Any:
    """Get a tracer instance.

    Args:
        name: Tracer name (usually __name__).

    Returns:
        A tracer from the registered provider, Hangar's own or one registered
        before it, whose spans never record exception text (see
        ``_TextFreeTracer``); NoOpTracer when there is none or tracing is off.
    """
    if not _tracing_active():
        return _noop_tracer

    return _TextFreeTracer(trace.get_tracer(name))


@contextmanager
def trace_span(
    name: str,
    attributes: dict[str, Any] | None = None,
    kind: str | None = None,
):
    """Context manager for creating trace spans.

    Args:
        name: Span name.
        attributes: Initial span attributes.
        kind: Span kind (client, server, producer, consumer, internal).

    Example:
        with trace_span("my_operation", {"key": "value"}) as span:
            span.add_event("checkpoint_reached")
            do_work()
    """
    tracer = get_tracer(__name__)

    span_kind = None
    if OTEL_AVAILABLE and kind:
        kind_map = {
            "client": trace.SpanKind.CLIENT,
            "server": trace.SpanKind.SERVER,
            "producer": trace.SpanKind.PRODUCER,
            "consumer": trace.SpanKind.CONSUMER,
            "internal": trace.SpanKind.INTERNAL,
        }
        span_kind = kind_map.get(kind.lower())

    with tracer.start_as_current_span(name, kind=span_kind) as span:
        if attributes:
            for key, value in attributes.items():
                span.set_attribute(key, value)
        yield span


@contextmanager
def upstream_call_span(method: str, params: dict[str, Any] | None = None):
    """CLIENT span for an outgoing MCP RPC to an upstream server.

    Wrap the transport call with this so the RPC is a proper SpanKind.CLIENT
    span and the trace context injected downstream (HTTP headers or stdio
    `_meta`) parents the upstream's server span to this one. Names/attributes
    follow OTel GenAI/MCP semconv. No-op when tracing is disabled.

    Keep it open until the answer is classified: an exception leaving it ends
    it in ERROR with the exception's class as ``error.type``, and a failure the
    client returns as data is recorded with :func:`record_upstream_outcome`.

    Args:
        method: JSON-RPC method (e.g. "tools/call", "tools/list", "initialize").
        params: Request params; for tools/call the tool name is read from
            ``params["name"]`` for the span name and gen_ai.tool.name.
    """
    params = params or {}
    attributes: dict[str, Any] = {MCP.METHOD_NAME: method}
    if method == "tools/call":
        tool = params.get("name")
        name = f"execute_tool {tool}" if tool else "execute_tool"
        attributes[GenAI.OPERATION_NAME] = "execute_tool"
        if tool:
            attributes[GenAI.TOOL_NAME] = tool
    else:
        name = method

    with trace_span(name, attributes=attributes, kind="client") as span:
        try:
            yield span
        except Exception as e:
            # ERROR with the exception's class as the bounded type, no message.
            record_upstream_outcome(span, error=e)
            raise


#: OTel ``error.type``, the attribute the mcp SDK's server middleware sets.
#: Its values pass ``mcp_hangar.errors.bounded_error_type``, the one allowlist.
ERROR_TYPE = "error.type"


def _set_error_type_if_absent(span: Any, error_type: str) -> None:
    """Set a bounded ``error.type`` unless the span has one: the failure closest to it names it."""
    try:
        if ERROR_TYPE not in (getattr(span, "attributes", None) or {}):
            span.set_attribute(ERROR_TYPE, bounded_error_type(error_type))
    except Exception:  # noqa: BLE001 -- fault-barrier: tracing must not break the traced path
        pass


#: The OTel exception event's type attribute, the only attribute Hangar gives that event.
EXCEPTION_TYPE = "exception.type"


def _record_exception_type(span: Any, error: BaseException) -> None:
    """Add an ``exception`` event carrying only ``exception.type``: no message, no stacktrace.

    The SDK's ``record_exception`` writes both, and either can carry what an
    upstream returned (GHSA-qwq2-7g49-jxc6). The type is the exception's
    qualified class name, bounded as ``error.type`` is.
    """
    try:
        span.add_event("exception", {EXCEPTION_TYPE: bounded_error_type(type(error).__qualname__)})
    except Exception:  # noqa: BLE001 -- fault-barrier: tracing must not break the traced path
        pass


def mark_span_error(span: Any, error_type: str | None = None) -> None:
    """Set ERROR status on a span, with no description. Safe for NoOp spans and when OTel is absent.

    Use when a failure is handled as data (e.g. converted to a result object)
    rather than raised, so the span would otherwise stay UNSET and the failing
    operation would look successful in the trace UI.

    The status never carries a description: a failure's message can hold what
    an upstream tool returned (GHSA-qwq2-7g49-jxc6). What failed is said by
    ``error.type``, set from ``error_type`` when one is given. It must look like
    a class name or a code (letters, digits and ``_.-<>``, at most 128
    characters); anything else is recorded as ``_OTHER``.
    """
    if not OTEL_AVAILABLE:
        return
    try:
        span.set_status(Status(StatusCode.ERROR))
        if error_type is not None:
            span.set_attribute(ERROR_TYPE, bounded_error_type(error_type))
    except Exception:  # noqa: BLE001 -- fault-barrier: tracing must not break the traced path
        pass


def record_upstream_outcome(
    span: Any, answer: Any = None, *, http_status: int | None = None, error: BaseException | None = None
) -> None:
    """Mark an upstream CLIENT span ERROR when its transport client says the call failed.

    Observes only: the client has already decided on the envelope it returns
    or the exception it raises, and this changes neither. ``error.type`` takes
    the mcp SDK server middleware's values, so both ends of a hop agree:
    ``http_<status>`` for a status failure, an exception's class name, the
    JSON-RPC error code as a string, ``tool_error`` for ``isError: true``.
    Nothing the upstream sent -- message, content, body -- is recorded.

    Args:
        span: The call's CLIENT span; a NoOp span is fine.
        answer: The JSON-RPC envelope the client returns.
        http_status: The status, when the client rejected the answer on it.
        error: The exception the client raised, or turned into an envelope.
    """
    if http_status is not None:
        error_type: str | None = f"http_{http_status}"
    elif error is not None:
        error_type = type(error).__qualname__
    else:
        error_type = _answer_error_type(answer)
    if error_type is None:
        return
    mark_span_error(span, error_type)


def record_handled_failure(span: Any, error: BaseException) -> None:
    """End a span ERROR for a failure its fault barrier caught and did not re-raise.

    Without this a failed append, handler, discovery cycle or cold start read
    as a success. This sets ERROR and sets ``error.type`` to the exception's
    qualified class name. On a span several handlers share, the first failure
    names it and a later success never resets it. Each failure adds an
    ``exception`` event carrying only ``exception.type``: an exception's message
    and stacktrace can carry what an upstream returned (GHSA-qwq2-7g49-jxc6).
    For operational failures only: a policy refusal is a correct answer, not an
    ERROR. No-op without the SDK or on a NoOp span; never raises.
    """
    if not OTEL_AVAILABLE:
        return
    mark_span_error(span)
    _set_error_type_if_absent(span, type(error).__qualname__)
    _record_exception_type(span, error)


def _answer_error_type(answer: Any) -> str | None:
    """``error.type`` of a failed JSON-RPC envelope, None for a success; the aggregate's test."""
    if not isinstance(answer, dict):
        return None
    if "error" in answer:
        code = answer["error"].get("code") if isinstance(answer["error"], dict) else None
        # The spec's integer code; anything else is not copied onto the span.
        return str(code) if type(code) is int else "_OTHER"
    result = answer.get("result")
    return "tool_error" if isinstance(result, dict) and result.get("isError") else None


def inject_trace_context(carrier: dict[str, Any]) -> None:
    """Write the current W3C trace context into an outbound carrier, and no baggage.

    This is the one outbound chokepoint. Both transports call it on every
    carrier they send upstream: HTTP on its headers and on ``params._meta``,
    stdio on ``params._meta``. So the rule cannot differ between them. It
    injects ``traceparent``/``tracestate`` and removes any ``baggage`` entry the
    carrier already holds, such as one a caller put in ``params._meta``. Baggage
    in the ambient context is never injected, whoever attached it: Hangar sets
    none, so none can be attributed to it (GHSA-qwq2-7g49-jxc6). The removal
    also runs with tracing off and without the SDK.

    Args:
        carrier: Outbound HTTP headers or MCP ``_meta``. Mutated in place.

    Example:
        headers = {}
        inject_trace_context(headers)
        # headers now contains traceparent (and tracestate, if any)
    """
    for key in [k for k in carrier if isinstance(k, str) and k.lower() == BAGGAGE_HEADER]:
        del carrier[key]
    if not _tracing_active():
        return

    _get_propagator().inject(carrier)


def extract_trace_context(carrier: dict[str, str]) -> Any:
    """Extract trace context from carrier dict.

    Args:
        carrier: Dict containing trace context.

    Returns:
        OpenTelemetry context or None.

    Extracts W3C TraceContext (traceparent/tracestate) only. A ``baggage``
    entry in the carrier is ignored, so none reaches the returned context.

    Example:
        context = extract_trace_context(request.headers)
        with tracer.start_as_current_span("handle", context=context):
            ...
    """
    if not _tracing_active():
        return None

    return _get_propagator().extract(carrier)


def get_current_trace_id() -> str | None:
    """Get current trace ID as hex string.

    Returns:
        Trace ID or None if not in a trace.
    """
    if not _tracing_active():
        return None

    span = trace.get_current_span()
    if span is None:
        return None

    ctx = span.get_span_context()
    if ctx is None or not ctx.is_valid:
        return None

    return format(ctx.trace_id, "032x")


def get_current_span_id() -> str | None:
    """Get current span ID as hex string.

    Returns:
        Span ID or None if not in a span.
    """
    if not _tracing_active():
        return None

    span = trace.get_current_span()
    if span is None:
        return None

    ctx = span.get_span_context()
    if ctx is None or not ctx.is_valid:
        return None

    return format(ctx.span_id, "016x")


def _get_version() -> str:
    """Get MCP Hangar version.

    Reads the installed distribution directly instead of importing
    ``__version__`` off the package root, which would drag the whole public API
    -- facade included -- into the tracing bootstrap just to label a resource.
    """
    try:
        from importlib.metadata import PackageNotFoundError, version

        return version("mcp-hangar")
    except (ImportError, PackageNotFoundError):
        return "unknown"