"""Typed span-data inputs: frozen dataclasses the engine and mappers consume."""

from __future__ import annotations

import json
from collections.abc import Mapping
from dataclasses import dataclass, field
from enum import Enum
from types import MappingProxyType
from typing import TYPE_CHECKING, ClassVar, Final, cast
from urllib.parse import urlsplit

from litellm.integrations.otel.model.metadata import (
    RequestContext,
    RequestIdentity,
)
from litellm.integrations.otel.model.semconv import (
    GenAIOperation,
    GenAIOutputType,
    MCPMethod,
    resolve_operation,
    resolve_output_type,
    resolve_provider,
)
from litellm.integrations.otel.model.utils import (
    as_bool,
    as_float,
    as_int,
    as_str,
    as_str_tuple,
)

# ``RequestIdentity`` and the request-metadata translation now live in
# :mod:`metadata`; re-exported here so existing ``model.payloads`` imports keep
# resolving it.
__all__ = [
    "GuardrailSpanData",
    "LLMCallSpanData",
    "LLMCost",
    "LLMRequestParams",
    "LLMUsage",
    "MCPListToolsSpanData",
    "MCPToolCallSpanData",
    "ProxyRequestSpanData",
    "RequestContext",
    "RequestIdentity",
    "ServerInfo",
    "ServiceSpanData",
    "SpanError",
    "ToolDefinition",
    "is_mcp_list_tools",
    "is_mcp_tool_call",
]

if TYPE_CHECKING:
    from litellm.types.services import ServiceLoggerPayload
    from litellm.types.utils import (
        StandardLoggingGuardrailInformation,
        StandardLoggingPayload,
    )


# --- typed sub-structures ---------------------------------------------------- #


def _cache_token_value(*values: object) -> int | None:
    explicit_zero = False
    invalid_before_zero = False
    for raw_value in values:
        if raw_value is None:
            continue
        if isinstance(raw_value, bool):
            parsed = None
        else:
            try:
                parsed = as_int(raw_value)
            except (OverflowError, ValueError):
                parsed = None
        if parsed is None:
            if not explicit_zero:
                invalid_before_zero = True
        elif parsed > 0:
            return parsed
        elif parsed == 0:
            explicit_zero = True
        elif not explicit_zero:
            invalid_before_zero = True
    return 0 if explicit_zero and not invalid_before_zero else None


@dataclass(frozen=True)
class LLMRequestParams:
    temperature: float | None = None
    top_p: float | None = None
    top_k: int | None = None
    max_tokens: int | None = None
    frequency_penalty: float | None = None
    presence_penalty: float | None = None
    stop_sequences: tuple[str, ...] | None = None
    seed: int | None = None

    @classmethod
    def from_model_parameters(cls, params: Mapping[str, object]) -> LLMRequestParams:
        max_tokens = as_int(params.get("max_tokens"))
        if max_tokens is None:
            max_tokens = as_int(params.get("max_completion_tokens"))
        return cls(
            temperature=as_float(params.get("temperature")),
            top_p=as_float(params.get("top_p")),
            top_k=as_int(params.get("top_k")),
            max_tokens=max_tokens,
            frequency_penalty=as_float(params.get("frequency_penalty")),
            presence_penalty=as_float(params.get("presence_penalty")),
            stop_sequences=as_str_tuple(params.get("stop")),
            seed=as_int(params.get("seed")),
        )


@dataclass(frozen=True)
class LLMUsage:
    input_tokens: int | None = None
    output_tokens: int | None = None
    total_tokens: int | None = None
    cache_creation_input_tokens: int | None = None
    cache_read_input_tokens: int | None = None

    @classmethod
    def from_standard_logging_payload(cls, payload: StandardLoggingPayload) -> LLMUsage:
        # Cache token counts only exist on the raw provider usage object under metadata
        metadata: Final[Mapping[str, object]] = payload.get("metadata") or {}
        raw_usage: Final = metadata.get("usage_object")
        usage_object: Final[Mapping[str, object]] = raw_usage if isinstance(raw_usage, Mapping) else {}
        raw_details: Final = usage_object.get("prompt_tokens_details")
        prompt_details: Final[Mapping[str, object]] = (
            raw_details if isinstance(raw_details, Mapping) else MappingProxyType({})
        )
        return cls(
            input_tokens=as_int(payload.get("prompt_tokens")),
            output_tokens=as_int(payload.get("completion_tokens")),
            total_tokens=as_int(payload.get("total_tokens")),
            cache_creation_input_tokens=_cache_token_value(
                usage_object.get("cache_creation_input_tokens"),
                prompt_details.get("cache_write_tokens"),
                prompt_details.get("cache_creation_tokens"),
                prompt_details.get("cache_creation_input_tokens"),
            ),
            cache_read_input_tokens=_cache_token_value(
                usage_object.get("cache_read_input_tokens"),
                prompt_details.get("cached_tokens"),
                usage_object.get("prompt_cache_hit_tokens"),
            ),
        )


@dataclass(frozen=True)
class LLMCost:
    """Per-component cost breakdown, from the StandardLoggingPayload
    ``cost_breakdown`` (``litellm.types.utils.CostBreakdown``).

    Each field is the USD cost of one component, or ``None`` when the source did
    not report it — so the mapper omits absent components instead of emitting 0.
    The final (post-discount/post-margin) total is carried separately on
    ``LLMCallSpanData.response_cost``. Free-form ``additional_costs`` are not
    surfaced here: span attributes are scalar and there is no agreed key shape
    for them yet.
    """

    input: float | None = None
    output: float | None = None
    cache_read: float | None = None
    cache_creation: float | None = None
    tool_usage: float | None = None
    original: float | None = None
    discount_amount: float | None = None
    discount_percent: float | None = None
    margin_fixed_amount: float | None = None
    margin_percent: float | None = None
    margin_total_amount: float | None = None

    @classmethod
    def from_breakdown(cls, breakdown: Mapping[str, object] | None) -> LLMCost:
        b: Final = breakdown or {}
        return cls(
            input=as_float(b.get("input_cost")),
            output=as_float(b.get("output_cost")),
            cache_read=as_float(b.get("cache_read_cost")),
            cache_creation=as_float(b.get("cache_creation_cost")),
            tool_usage=as_float(b.get("tool_usage_cost")),
            original=as_float(b.get("original_cost")),
            discount_amount=as_float(b.get("discount_amount")),
            discount_percent=as_float(b.get("discount_percent")),
            margin_fixed_amount=as_float(b.get("margin_fixed_amount")),
            margin_percent=as_float(b.get("margin_percent")),
            margin_total_amount=as_float(b.get("margin_total_amount")),
        )


@dataclass(frozen=True)
class SpanError:
    error_type: str | None = None
    message: str | None = None
    code: str | None = None
    stack_trace: str | None = None
    llm_provider: str | None = None


@dataclass(frozen=True)
class ServerInfo:
    address: str | None = None
    port: int | None = None

    @classmethod
    def from_api_base(cls, api_base: str | None) -> ServerInfo | None:
        if not api_base:
            return None
        parsed: Final = urlsplit(api_base if "://" in api_base else f"//{api_base}")
        if not parsed.hostname:
            return None
        return cls(address=parsed.hostname, port=parsed.port)


@dataclass(frozen=True)
class GuardrailSpanData:
    guardrail_name: str
    mode: str | None = None
    status: str | None = None
    masked_entity_count: int | None = None
    provider: str | None = None
    action: str | None = None
    # The guardrail verdict / provider response (e.g. the moderation result),
    # JSON-serialized. This is the detail that belongs on the guardrail span.
    response_json: str | None = None
    violation_categories: tuple[str, ...] = ()
    confidence_score: float | None = None
    risk_score: float | None = None
    duration: float | None = None
    # Actual execution window (epoch seconds) from the logging entry, so the span
    # is placed when the guardrail really ran — a pre_call guardrail before the
    # LLM call — rather than at post-call emission time.
    start_time: float | None = None
    end_time: float | None = None
    # Provider-agnostic configuration/detection metadata (see
    # ``StandardLoggingGuardrailInformation``). Present for any guardrail that
    # populates them, not just one provider's shape.
    guardrail_id: str | None = None
    policy_template: str | None = None
    detection_method: str | None = None
    # Provider-reported billable usage counters (JSON-serialized) and the USD cost
    # priced from them by the provider hook (``guardrail_usage`` /
    # ``guardrail_cost`` on ``StandardLoggingGuardrailInformation``).
    usage_json: str | None = None
    cost: float | None = None
    # Whether ``cost`` participates in the request's billed spend (absent means
    # billed, the default; False means report-only). Mirrors
    # ``guardrail_cost_in_spend`` so trace consumers can avoid double-counting.
    cost_in_spend: bool | None = None
    # Set when the guardrail intervened/blocked or failed, so the emitter marks
    # the span ERROR — a blocking guardrail is an error outcome for that span.
    error: SpanError | None = None

    # Guardrail statuses that mean the guardrail did not pass the request through.
    _ERROR_STATUSES: ClassVar[frozenset[str]] = frozenset({"guardrail_intervened", "guardrail_failed_to_respond"})

    @classmethod
    def from_logging_entry(cls, entry: StandardLoggingGuardrailInformation) -> GuardrailSpanData:
        """Build from one ``standard_logging_guardrail_information`` entry.

        Reads the canonical, provider-agnostic ``StandardLoggingGuardrailInformation``
        keys only — no guessing at a single provider's field names. Values that are
        typed as enums or lists (e.g. ``guardrail_mode``) are normalized to a
        stable string rather than assumed to already be plain strings.
        """
        get: Final = cast(Mapping[str, object], entry).get
        status: Final = as_str(get("guardrail_status"))
        response: Final = get("guardrail_response")
        usage: Final = get("guardrail_usage")
        in_spend: Final = get("guardrail_cost_in_spend")
        error: Final = (
            SpanError(error_type=status, message=as_str(get("guardrail_action")))
            if status in cls._ERROR_STATUSES
            else None
        )
        return cls(
            guardrail_name=as_str(get("guardrail_name")) or "guardrail",
            mode=_guardrail_mode_str(get("guardrail_mode")),
            status=status,
            masked_entity_count=_total_masked_entities(get("masked_entity_count")),
            provider=as_str(get("guardrail_provider")),
            action=as_str(get("guardrail_action")),
            response_json=_json_or_none(response) if response is not None else None,
            violation_categories=as_str_tuple(get("violation_categories")) or (),
            confidence_score=as_float(get("confidence_score")),
            risk_score=as_float(get("risk_score")),
            duration=as_float(get("duration")),
            start_time=as_float(get("start_time")),
            end_time=as_float(get("end_time")),
            guardrail_id=as_str(get("guardrail_id")),
            policy_template=as_str(get("policy_template")),
            detection_method=as_str(get("detection_method")),
            usage_json=_json_or_none(usage) if usage is not None else None,
            cost=as_float(get("guardrail_cost")),
            cost_in_spend=in_spend if isinstance(in_spend, bool) else None,
            error=error,
        )


@dataclass(frozen=True)
class ServiceSpanData:
    service_name: str
    call_type: str | None = None
    error: SpanError | None = None
    # Caller-supplied attributes to stamp on the service span, passed through
    # from ``async_service_*_hook(event_metadata=...)``. The mapper owns how
    # these are namespaced: the canonical vocabulary uses ``litellm.metadata.*``
    # keys, the semconv-ai / Traceloop vocabulary uses the bare key names.
    event_metadata: Mapping[str, str] = field(default_factory=dict)

    @classmethod
    def from_payload(
        cls,
        payload: ServiceLoggerPayload,
        event_metadata: Mapping[str, object] | None = None,
    ) -> ServiceSpanData:
        # ``payload.service`` is a ``ServiceTypes(str, Enum)`` and ``error`` is
        # ``Optional[str]`` on the Pydantic model — no defensive reads needed.
        # ``event_metadata`` is sanitized: the legacy service decorators pass raw
        # call-site data (live objects, full request metadata, response headers),
        # none of which belongs on a span.
        return cls(
            service_name=payload.service.value,
            call_type=payload.call_type,
            error=SpanError(message=payload.error) if payload.error else None,
            event_metadata=sanitize_event_metadata(event_metadata),
        )


@dataclass(frozen=True)
class ProxyRequestSpanData:
    http_method: str
    route: str
    url_path: str | None = None
    status_code: int | None = None
    identity: RequestIdentity | None = None


# --- the primary LLM-call model ---------------------------------------------- #


@dataclass(frozen=True)
class ToolDefinition:
    """A single function/tool declared on a chat-completion request."""

    name: str
    description: str | None = None
    parameters_json: str | None = None  # JSON-serialized schema (str so it's an AttrValue)


@dataclass(frozen=True)
class LLMCallSpanData:
    operation: GenAIOperation
    provider: str
    request_model: str
    response_model: str | None
    response_id: str | None
    request_params: LLMRequestParams
    usage: LLMUsage
    finish_reasons: tuple[str, ...]
    error: SpanError | None
    response_cost: float | None
    server: ServerInfo | None
    identity: RequestIdentity
    is_streaming: bool | None = None
    cost: LLMCost = field(default_factory=LLMCost)
    tools: tuple[ToolDefinition, ...] = ()
    # Raw messages and response, needed by vendor mappers (OpenInference,
    # Langfuse, Weave) that stamp message-level attributes. ``messages_in`` is
    # the request payload; ``choices_out`` mirrors ``response.choices`` from
    # the StandardLoggingPayload. Both are tuples of immutable mappings so the
    # dataclass stays hashable and frozen.
    messages_in: tuple[Mapping[str, object], ...] = ()
    choices_out: tuple[Mapping[str, object], ...] = ()
    system_fingerprint: str | None = None
    time_to_first_chunk_seconds: float | None = None
    # The requested output modality, set only on the routes that pin one (image
    # generation, speech, transcription, OCR), and the litellm route itself, which
    # keeps routes the convention folds into one operation distinguishable.
    output_type: GenAIOutputType | None = None
    call_type: str | None = None

    @classmethod
    def from_standard_logging_payload(
        cls,
        payload: StandardLoggingPayload,
        capture_content: bool = False,
        time_to_first_chunk_seconds: float | None = None,
    ) -> LLMCallSpanData:
        params: Final = cast(Mapping[str, object], payload.get("model_parameters") or {})
        # The single parse of the request's metadata — the request-vs-provider
        # model split, the response model, api base, and identity all come from
        # here rather than being re-derived from the raw payload dicts.
        context: Final = RequestContext.from_standard_logging_payload(payload)
        # Normalize ``response`` to a dict once so the content/id reads below are a
        # plain ``.get`` — no repeated ``isinstance`` guards.
        raw_response: Final = payload.get("response")
        response: Final = cast(Mapping[str, object], raw_response if isinstance(raw_response, dict) else {})
        choices_out: Final = _dicts(response.get("choices"))
        # ``finish_reasons`` is metadata, not content, so derive it from
        # ``choices_out`` before gating. The raw message/choice bodies are only
        # retained when content capture is enabled (see ``capture_span_content``);
        # otherwise the content-bearing mappers receive empty sequences and emit
        # no prompt/response text.
        finish_reasons: Final = _finish_reasons(choices_out)
        call_type: Final = as_str(payload.get("call_type"))
        return cls(
            operation=resolve_operation(call_type),
            provider=resolve_provider(as_str(payload.get("custom_llm_provider"))),
            request_model=context.request_model,
            response_model=context.response_model,
            response_id=as_str(response.get("id")),
            request_params=LLMRequestParams.from_model_parameters(params),
            usage=LLMUsage.from_standard_logging_payload(payload),
            finish_reasons=finish_reasons,
            error=_parse_error(payload),
            response_cost=as_float(payload.get("response_cost")),
            cost=LLMCost.from_breakdown(cast("Mapping[str, object] | None", payload.get("cost_breakdown"))),
            server=ServerInfo.from_api_base(context.api_base),
            identity=context.identity,
            is_streaming=as_bool(payload.get("stream")),
            tools=_extract_tools(params),
            messages_in=_dicts(payload.get("messages")) if capture_content else (),
            choices_out=choices_out if capture_content else (),
            system_fingerprint=as_str(response.get("system_fingerprint")),
            time_to_first_chunk_seconds=time_to_first_chunk_seconds,
            output_type=resolve_output_type(call_type),
            call_type=call_type or None,
        )


# --- the MCP tool-call model ------------------------------------------------- #


def _upstream_address_port(resource: str | None) -> tuple[str | None, int | None]:
    """Split a redacted MCP server origin into ``server.address`` / ``server.port``.

    ``mcp_server_resource`` is a scheme + host + port origin with userinfo, path,
    query and fragment already stripped. The port falls back to the scheme default
    when the origin omits it, because a consumer that keys a downstream dependency
    off the address renders a missing port as ``0``.

    The origin is rebuilt without its IPv6 brackets upstream, so reading the port can
    raise on an address the host check still admits: a zone-scoped ``fe80::1%25eth0``
    leaves a truthy hostname of ``fe80`` behind. Both halves are read inside the guard
    so an unparseable origin yields no address rather than propagating out of span
    construction, matching how the redactor guards the same split.
    """
    if not resource:
        return None, None
    try:
        parsed: Final = urlsplit(resource)
        hostname: Final = parsed.hostname
        port: Final = parsed.port
    except ValueError:
        return None, None
    if not hostname:
        return None, None
    default_port: Final = 443 if parsed.scheme == "https" else 80 if parsed.scheme == "http" else None
    return hostname, port or default_port


@dataclass(frozen=True)
class MCPToolCallSpanData:
    """One MCP ``tools/call`` execution, parsed from a closed request's payload.

    The proxy is an MCP *client* to the upstream server it forwards the call to,
    so this is a CLIENT span. ``arguments_json``/``result_json`` are the tool's
    input/output — sensitive content, so they're only retained when content
    capture is enabled, mirroring ``LLMCallSpanData``'s message bodies.
    """

    operation: GenAIOperation
    method: str
    tool_name: str
    server_name: str | None
    server_address: str | None
    server_port: int | None
    session_id: str | None
    arguments_json: str | None
    result_json: str | None
    error: SpanError | None
    response_cost: float | None
    identity: RequestIdentity

    @classmethod
    def from_standard_logging_payload(
        cls, payload: StandardLoggingPayload, capture_content: bool = False
    ) -> MCPToolCallSpanData:
        meta: Final = _mcp_tool_call_metadata(cast(Mapping[str, object], payload))
        address, port = _upstream_address_port(as_str(meta.get("mcp_server_resource")) or None)
        return cls(
            operation=resolve_operation(as_str(payload.get("call_type"))),
            method=MCPMethod.TOOLS_CALL.value,
            tool_name=as_str(meta.get("name")) or "",
            server_name=as_str(meta.get("mcp_server_name")),
            server_address=address,
            server_port=port,
            session_id=as_str(meta.get("mcp_session_id")),
            arguments_json=(
                _json_or_none(meta.get("arguments")) if capture_content and meta.get("arguments") is not None else None
            ),
            result_json=(
                _json_or_none(meta.get("result")) if capture_content and meta.get("result") is not None else None
            ),
            error=_parse_error(payload),
            response_cost=as_float(payload.get("response_cost")),
            identity=RequestContext.from_standard_logging_payload(payload).identity,
        )


def _mcp_tool_call_metadata(payload: Mapping[str, object]) -> Mapping[str, object]:
    """The MCP gateway's tool-call metadata, which lives under
    ``StandardLoggingPayload.metadata`` (a ``StandardLoggingMetadata`` key), not
    at the payload's top level."""
    metadata: Final = payload.get("metadata")
    if not isinstance(metadata, Mapping):
        return {}
    meta: Final = metadata.get("mcp_tool_call_metadata")
    return meta if isinstance(meta, Mapping) else {}


def is_mcp_tool_call(payload: Mapping[str, object]) -> bool:
    """Whether a closed request's payload is an MCP tool call rather than an LLM
    call — true when the MCP gateway stamped its tool-call metadata, or the call
    type says so on a path that hasn't populated the metadata yet."""
    return bool(_mcp_tool_call_metadata(payload)) or (payload.get("call_type") == "call_mcp_tool")


@dataclass(frozen=True)
class MCPListToolsSpanData:
    """One MCP ``tools/list`` discovery call, parsed from a closed request's payload.

    The proxy is an MCP *client* enumerating an upstream server's tools, so this is
    a CLIENT span. It carries neither ``gen_ai.operation.name`` nor ``gen_ai.tool.name``:
    the GenAI semconv sets ``execute_tool`` (and the tool name) only for tool *calls*,
    and listing executes no tool.
    """

    method: str
    session_id: str | None
    error: SpanError | None
    identity: RequestIdentity

    @classmethod
    def from_standard_logging_payload(
        cls, payload: StandardLoggingPayload, capture_content: bool = False
    ) -> MCPListToolsSpanData:
        # The list-tools logging path does not thread an MCP session id into the
        # payload (only the tool-call path stamps ``mcp_tool_call_metadata``), so
        # there is none to read here; ``mcp.session.id`` is simply omitted.
        return cls(
            method=MCPMethod.TOOLS_LIST.value,
            session_id=None,
            error=_parse_error(payload),
            identity=RequestContext.from_standard_logging_payload(payload).identity,
        )


def is_mcp_list_tools(payload: Mapping[str, object]) -> bool:
    """Whether a closed request's payload is an MCP ``tools/list`` discovery call
    rather than a tool call or an LLM call — true when the call type says so."""
    return payload.get("call_type") == "list_mcp_tools"


# --- service event_metadata sanitization ------------------------------------ #

# Substrings (case-insensitive) of keys that must never reach a span: secrets,
# tokens, and raw request/response dumps the legacy service decorators pass.
_SENSITIVE_METADATA_SUBSTRINGS: Final[tuple[str, ...]] = (
    "api_key",
    "token",
    "secret",
    "password",
    "cookie",
    "authorization",
    "header",
    "hidden_params",
)
# Keys that carry raw call-site internals — live objects, full kwargs/args. The
# operation name is already the span's ``call_type``, so ``function_name`` is
# redundant.
_DROP_METADATA_KEYS: Final[frozenset] = frozenset({"function_kwargs", "function_args", "function_name"})
_MAX_METADATA_VALUE_LEN: Final = 1024
_MAX_METADATA_ITEMS: Final = 32


def sanitize_event_metadata(
    event_metadata: Mapping[str, object] | None,
) -> dict[str, str]:
    """Reduce caller-supplied ``event_metadata`` to span-safe string attributes.

    Keeps only primitive values (str/int/float/bool) under non-sensitive keys —
    never ``repr()``-ing objects, dicts, or lists, never stamping secrets/headers,
    and bounding the count and per-value length. This is the single chokepoint:
    both the GenAI and legacy mappers read the cleaned result.
    """
    if not event_metadata:
        return {}
    clean: Final[dict[str, str]] = {}
    for key, value in event_metadata.items():
        if len(clean) >= _MAX_METADATA_ITEMS:
            break
        if not isinstance(key, str) or key in _DROP_METADATA_KEYS:
            continue
        lowered = key.lower()
        if any(token in lowered for token in _SENSITIVE_METADATA_SUBSTRINGS):
            continue
        # ``bool`` is a subclass of ``int``, so it's covered. Non-primitive values
        # (objects, dicts, lists) are dropped rather than stringified.
        if isinstance(value, (str, int, float)):
            clean[key] = str(value)[:_MAX_METADATA_VALUE_LEN]
    return clean


def _json_or_none(value: object) -> str | None:
    """JSON-serialize ``value`` (already-string values pass through). ``None`` on failure."""
    if isinstance(value, str):
        return value
    try:
        return json.dumps(value, default=str)
    except Exception:
        return None


def _guardrail_mode_str(value: object) -> str | None:
    """Normalize ``guardrail_mode`` to a stable string.

    ``guardrail_mode`` is typed as a ``GuardrailEventHooks`` enum, a list of them,
    or a ``GuardrailMode`` — not a plain string. Emit the enum *value* (e.g.
    ``"pre_call"``) rather than ``str(enum)`` (``"GuardrailEventHooks.pre_call"``),
    and join a list of modes so a guardrail that runs at multiple hooks is
    represented faithfully.
    """
    if value is None:
        return None
    if isinstance(value, (list, tuple)):
        parts: Final[list[str]] = []
        for item in value:
            if item is None:
                continue
            part = as_str(item.value) if isinstance(item, Enum) else as_str(item)
            if part:
                parts.append(part)
        return ",".join(parts) or None
    if isinstance(value, Enum):
        return as_str(value.value)
    return as_str(value)


def _total_masked_entities(value: object) -> int | None:
    """``masked_entity_count`` is a ``{entity_type: count}`` map — sum to a total."""
    if isinstance(value, Mapping):
        total: Final = sum(v for v in value.values() if isinstance(v, int))
        return total or None
    return as_int(value)


def _dicts(value: object) -> tuple[Mapping[str, object], ...]:
    """The dict items of ``value`` (when it's a list), as a tuple. Else empty."""
    if not isinstance(value, list):
        return ()
    return tuple(item for item in value if isinstance(item, dict))


def _finish_reasons(choices: tuple[Mapping[str, object], ...]) -> tuple[str, ...]:
    """Non-empty ``finish_reason`` of each response choice."""
    return tuple(r for c in choices if (r := as_str(c.get("finish_reason"))))


def _parse_error(payload: StandardLoggingPayload) -> SpanError | None:
    """A ``SpanError`` for a failed request, or ``None`` on success."""
    if payload.get("status") != "failure":
        return None
    info: Final = cast(Mapping[str, object], payload.get("error_information") or {})
    return SpanError(
        error_type=as_str(info.get("error_class")) or as_str(info.get("error_code")),
        message=as_str(info.get("error_message")) or as_str(payload.get("error_str")),
        code=as_str(info.get("error_code")),
        stack_trace=as_str(info.get("traceback")),
        llm_provider=as_str(info.get("llm_provider")),
    )


def _tool_from_entry(entry: object) -> ToolDefinition | None:
    """One ``tools``/``functions`` entry → ``ToolDefinition``, or ``None`` if unusable."""
    if not isinstance(entry, dict):
        return None
    fn: Final = entry.get("function") if "function" in entry else entry
    if not isinstance(fn, dict):
        return None
    name: Final = as_str(fn.get("name"))
    if not name:
        return None
    params: Final = fn.get("parameters")
    parameters_json: str | None = None
    if params is not None:
        try:
            parameters_json = json.dumps(params, default=str)
        except Exception:
            parameters_json = None
    return ToolDefinition(
        name=name,
        description=as_str(fn.get("description")),
        parameters_json=parameters_json,
    )


def _extract_tools(
    model_parameters: Mapping[str, object],
) -> tuple[ToolDefinition, ...]:
    """Pull declared tools from request params (OpenAI / Anthropic shape).

    Accepts the chat-completion ``tools=[{"type":"function", "function":
    {...}}, ...]`` shape, and falls back to the ``functions=[...]`` shape.
    Returns an empty tuple when neither is present.
    """
    raw_tools = model_parameters.get("tools")
    if not isinstance(raw_tools, list):
        raw_tools = model_parameters.get("functions")  # ``functions`` shape
    if not isinstance(raw_tools, list):
        return ()
    return tuple(t for entry in raw_tools if (t := _tool_from_entry(entry)) is not None)
