"""APM Metrics bridge — drives gen_ai.client.operation.duration histogram.

Mirrors frameworks-py/src/runner/apm_metrics_bridge.py.
"""
from __future__ import annotations

from typing import Any

from opentelemetry.context import Context
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import (
    AggregationTemporality,
    PeriodicExportingMetricReader,
)
from opentelemetry.sdk.metrics._internal.instrument import (
    Counter,
    Histogram,
    ObservableCounter,
    ObservableGauge,
    ObservableUpDownCounter,
    UpDownCounter,
)
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import ReadableSpan, Span
from opentelemetry.sdk.trace.export import SpanProcessor
from opentelemetry.trace.status import StatusCode

from .config import (
    METRIC_EXPORT_INTERVAL_MS,
    METRIC_EXPORT_TIMEOUT_MS,
    MIN_LLM_CALL_DURATION_MS,
)
from .llm_semconv import (
    OI_LLM_KIND_SET,
    OI_LLM_LEAF_KIND_SET,
    oi_kind_to_operation,
)

# 与 span_exporter._INTERRUPT_EXCEPTION_NAMES 保持一致：LangGraph 控制流异常名，
# 外加 interrupt 触发后流拆除 / 取消的 teardown 信号 GeneratorExit / CancelledError。
_INTERRUPT_EXCEPTION_NAMES = frozenset({
    "GraphInterrupt",
    "NodeInterrupt",
    "ParentCommand",
    "GeneratorExit",
    "CancelledError",
})

# 与 span_exporter._CREWAI_INTERRUPT_EXCEPTION_NAMES 保持一致：CrewAI @human_feedback
# 暂停信号 HumanFeedbackPending，非业务错误，仅在 crewai instrumentation scope 内归一，
# 避免把人工反馈暂停计入 LLM 错误指标。
_CREWAI_INTERRUPT_EXCEPTION_NAMES = frozenset({
    "HumanFeedbackPending",
})


def _is_crewai_scope(span: ReadableSpan) -> bool:
    scope = getattr(span, "instrumentation_scope", None)
    name = getattr(scope, "name", None)
    return isinstance(name, str) and "crewai" in name.lower()


def _iter_exception_type_names(span: ReadableSpan):
    attrs = span.attributes or {}
    for key in ("exception.type", "error.type"):
        v = attrs.get(key)
        if isinstance(v, str) and v:
            yield v.rsplit(".", 1)[-1]
    for ev in getattr(span, "events", None) or ():
        if getattr(ev, "name", None) != "exception":
            continue
        ev_attrs = getattr(ev, "attributes", None) or {}
        v = ev_attrs.get("exception.type")
        if isinstance(v, str) and v:
            yield v.rsplit(".", 1)[-1]


def _is_interrupt_span(span: ReadableSpan) -> bool:
    names = list(_iter_exception_type_names(span))
    if any(n in _INTERRUPT_EXCEPTION_NAMES for n in names):
        return True
    if _is_crewai_scope(span) and any(
        n in _CREWAI_INTERRUPT_EXCEPTION_NAMES for n in names
    ):
        return True
    return False


def _pick_str(span: ReadableSpan, *keys: str) -> str | None:
    attrs = span.attributes or {}
    for k in keys:
        v = attrs.get(k)
        if isinstance(v, str) and v:
            return v
    return None


def _shared_attrs(span: ReadableSpan) -> dict[str, Any]:
    model = (
        _pick_str(span, "llm.model_name", "gen_ai.response.model", "gen_ai.request.model")
        or "unknown"
    )
    system = (
        _pick_str(span, "llm.provider", "gen_ai.system", "gen_ai.provider.name")
        or "unknown"
    )
    oi_kind = _pick_str(span, "openinference.span.kind") or ""
    request_type = _pick_str(span, "llm.request.type") or oi_kind_to_operation(oi_kind)
    out: dict[str, Any] = {
        "gen_ai.system": system,
        "gen_ai.response.model": model,
        "gen_ai.operation.name": request_type,
    }
    streaming = (span.attributes or {}).get("llm.is_streaming")
    if streaming is True or streaming == "true":
        out["stream"] = True
    return out


def _is_oi_llm_leaf(span: ReadableSpan) -> bool:
    kind = _pick_str(span, "openinference.span.kind")
    return kind is not None and kind in OI_LLM_LEAF_KIND_SET


def _is_oi_llm_span(span: ReadableSpan) -> bool:
    kind = _pick_str(span, "openinference.span.kind")
    return kind is not None and kind in OI_LLM_KIND_SET


def _duration_seconds(span: ReadableSpan) -> float:
    if span.start_time is None or span.end_time is None:
        return 0.0
    return (span.end_time - span.start_time) / 1e9


_DELTA_PREF = {
    Counter: AggregationTemporality.DELTA,
    UpDownCounter: AggregationTemporality.CUMULATIVE,
    Histogram: AggregationTemporality.DELTA,
    ObservableCounter: AggregationTemporality.DELTA,
    ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
    ObservableGauge: AggregationTemporality.CUMULATIVE,
}


class ApmMetricsBridge:
    def __init__(self, *, endpoint: str, apm_token: str, resource: Resource) -> None:
        self._exporter = OTLPMetricExporter(
            endpoint=f"{endpoint.rstrip('/')}/v1/metrics",
            preferred_temporality=_DELTA_PREF,
        )
        self._reader = PeriodicExportingMetricReader(
            self._exporter,
            export_interval_millis=METRIC_EXPORT_INTERVAL_MS,
            export_timeout_millis=METRIC_EXPORT_TIMEOUT_MS,
        )
        self._meter_provider = MeterProvider(
            resource=resource,
            metric_readers=[self._reader],
        )
        meter = self._meter_provider.get_meter(
            "edgeone-agent-apm-bridge",
            "0.1.0",
        )
        self._duration_histogram = meter.create_histogram(
            name="gen_ai.client.operation.duration",
            unit="s",
            description="GenAI operation duration",
        )
        self._processor = _BridgeSpanProcessor(self)

    @property
    def span_processor(self) -> SpanProcessor:
        return self._processor

    def record(self, span: ReadableSpan) -> None:
        if not _is_oi_llm_span(span):
            return
        if not _is_oi_llm_leaf(span):
            return
        if not _pick_str(
            span,
            "llm.model_name",
            "gen_ai.response.model",
            "gen_ai.request.model",
        ):
            return
        dur = _duration_seconds(span)
        if dur * 1000 < MIN_LLM_CALL_DURATION_MS:
            return
        attrs = _shared_attrs(span)
        if (
            span.status
            and span.status.status_code == StatusCode.ERROR
            and not _is_interrupt_span(span)
        ):
            err = _pick_str(span, "error.type", "exception.type") or "error"
            attrs["error.type"] = err
        self._duration_histogram.record(dur, attributes=attrs)

    def shutdown(self) -> None:
        try:
            self._meter_provider.shutdown()
        except Exception:
            pass


class _BridgeSpanProcessor(SpanProcessor):
    def __init__(self, bridge: ApmMetricsBridge) -> None:
        self._bridge = bridge

    def on_start(self, span: Span, parent_context: Context | None = None) -> None:
        return None

    def on_end(self, span: ReadableSpan) -> None:
        try:
            self._bridge.record(span)
        except Exception:
            pass

    def shutdown(self) -> None:
        return None

    def force_flush(self, timeout_millis: int = 30_000) -> bool:
        return True