server/llm_adapter.py

"""Plug-and-play LLM adapter for TalkingHeadBench ingestion bundles."""

from __future__ import annotations

import ipaddress
import json
import os
from typing import Any, Literal
from urllib.parse import urlparse

from openai import APIConnectionError, APIStatusError, OpenAI

Provider = Literal["openai", "anthropic", "huggingface", "local"]
TaskTier = Literal[
    "image_audit",
    "clip_audit",
    "weight_audit",
    "easy",
    "medium",
    "hard",
]

_DRIFT_WARNING_THRESHOLD = 0.22
_DRIFT_HIGH_RISK_THRESHOLD = 0.35
_BLUR_WARNING_THRESHOLD = 0.30
_BLUR_HIGH_RISK_THRESHOLD = 0.20
_LIPSYNC_WARNING_THRESHOLD = 0.40
_LIPSYNC_HIGH_RISK_THRESHOLD = 0.30

_SYSTEM_PROMPT = (
    "You are a TalkingHeadBench diagnostic assistant specializing in talking-head LoRA pipelines.\n"
    "You receive only pre-extracted, deterministic signals and never raw generated outputs.\n"
    "Rules:\n"
    "1. Every claim must cite a specific digest field name and measured value.\n"
    "2. Parameter recommendations must be directional only (increase/decrease/enable/disable/reconsider) and never include target numbers.\n"
    "3. If fallback extractor data is present, treat fallback clip lip-sync and phoneme signals as unreliable.\n"
    "4. Do not invent any signal that is not present in the digest.\n"
    "5. Follow the required section structure from the user prompt exactly."
)


class LLMAdapterError(Exception):
    """Structured error surfaced by the one-shot analysis adapter."""

    def __init__(
        self,
        *,
        code: str,
        message: str,
        status_code: int,
        retryable: bool = False,
    ) -> None:
        super().__init__(message)
        self.code = code
        self.message = message
        self.status_code = status_code
        self.retryable = retryable


def analyze_ingested_bundle(
    bundle: dict[str, Any],
    *,
    model_id: str | None,
    api_key: str | None,
    provider: str | None,
    base_url: str | None,
    max_tokens: int,
    temperature: float,
    timeout_s: float,
    task_tier: TaskTier = "weight_audit",
) -> dict[str, Any]:
    """Generate a one-shot natural-language report from an ingested signal bundle."""
    if not isinstance(bundle, dict):
        raise LLMAdapterError(
            code="invalid_ingestion_bundle",
            message="The ingested bundle payload is invalid.",
            status_code=400,
            retryable=False,
        )

    resolved_provider = _resolve_provider(
        provider=provider,
        model_id=model_id,
        api_key=api_key,
        base_url=base_url,
    )
    resolved_model = _resolve_model_id(resolved_provider, model_id)
    resolved_base_url = _resolve_custom_base_url(
        provider=resolved_provider,
        base_url=base_url,
    )
    signal_digest = _build_signal_digest(bundle)
    prompt = _build_prompt(signal_digest, task_tier=task_tier)

    if resolved_provider == "openai":
        report = _call_openai(
            model_id=resolved_model,
            api_key=api_key,
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=temperature,
            timeout_s=timeout_s,
            base_url=resolved_base_url,
        )
    elif resolved_provider == "anthropic":
        report = _call_anthropic(
            model_id=resolved_model,
            api_key=api_key,
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=temperature,
            timeout_s=timeout_s,
            base_url=resolved_base_url,
        )
    elif resolved_provider == "huggingface":
        report = _call_huggingface(
            model_id=resolved_model,
            api_key=api_key,
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=temperature,
            timeout_s=timeout_s,
            base_url=resolved_base_url,
        )
    else:
        report = _call_local(
            model_id=resolved_model,
            api_key=api_key,
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=temperature,
            timeout_s=timeout_s,
            base_url=resolved_base_url,
        )

    text = (report or "").strip()
    if not text:
        raise LLMAdapterError(
            code="empty_model_response",
            message="The model provider returned an empty analysis response.",
            status_code=502,
            retryable=True,
        )

    return {
        "provider": resolved_provider,
        "model_id": resolved_model,
        "report": text,
        "signal_digest": signal_digest,
    }


def _resolve_provider(
    *,
    provider: str | None,
    model_id: str | None,
    api_key: str | None,
    base_url: str | None,
) -> Provider:
    requested = (provider or "auto").strip().lower()
    if requested == "hf":
        requested = "huggingface"

    if requested not in {"auto", "openai", "anthropic", "huggingface", "local"}:
        raise LLMAdapterError(
            code="unsupported_provider",
            message=(
                "provider must be one of: auto, openai, anthropic, huggingface, local"
            ),
            status_code=400,
            retryable=False,
        )

    if requested != "auto":
        return requested  # type: ignore[return-value]

    key = (api_key or "").strip()
    model = (model_id or os.getenv("MODEL_NAME") or "").strip().lower()
    resolved_base = (base_url or os.getenv("API_BASE_URL") or "").strip().lower()

    if key.startswith("sk-ant-") or model.startswith("claude"):
        return "anthropic"
    if key.startswith("hf_"):
        return "huggingface"
    if key.startswith("sk-") or model.startswith(("gpt-", "o1", "o3", "o4")):
        return "openai"
    if "/" in model:
        return "huggingface"

    if "router.huggingface.co" in resolved_base or "api-inference.huggingface.co" in resolved_base:
        return "huggingface"
    if "api.openai.com" in resolved_base:
        return "openai"
    if "api.anthropic.com" in resolved_base:
        return "anthropic"

    if resolved_base and not any(
        marker in resolved_base
        for marker in ["localhost", "127.0.0.1", "0.0.0.0", "11434"]
    ):
        # Any configured remote OpenAI-compatible base should avoid local fallback.
        return "openai"

    return "local"


def _resolve_model_id(provider: Provider, model_id: str | None) -> str:
    cleaned = (model_id or "").strip()
    if cleaned:
        return cleaned

    global_model = os.getenv("MODEL_NAME", "").strip()
    if global_model:
        return global_model

    if provider == "openai":
        return os.getenv("THB_OPENAI_MODEL", "gpt-4o-mini")
    if provider == "anthropic":
        return os.getenv("THB_ANTHROPIC_MODEL", "claude-3-5-haiku-latest")
    if provider == "huggingface":
        env_model = os.getenv("THB_HF_MODEL", "").strip()
        if env_model:
            return env_model
        # Default to a well-supported chat model via the HF router.
        return "Qwen/Qwen2.5-72B-Instruct"
    return os.getenv("THB_LOCAL_MODEL", "llama3.1:8b-instruct-q4_K_M")


def _env_truthy(name: str, *, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def _allowed_base_url_prefixes() -> list[str]:
    raw = os.getenv("THB_ALLOWED_BASE_URL_PREFIXES", "")
    prefixes = [item.strip().rstrip("/") for item in raw.split(",") if item.strip()]
    return prefixes


def _resolve_custom_base_url(*, provider: Provider, base_url: str | None) -> str | None:
    custom = (base_url or "").strip()
    if not custom:
        return None

    configured_api_base = (os.getenv("API_BASE_URL") or "").strip()
    if configured_api_base and _normalize_base_url(custom) == _normalize_base_url(configured_api_base):
        return custom

    if not _env_truthy("THB_ALLOW_CUSTOM_BASE_URLS", default=False):
        raise LLMAdapterError(
            code="custom_base_url_disabled",
            message="Custom base_url is disabled for this deployment.",
            status_code=400,
            retryable=False,
        )

    _validate_custom_base_url(custom, provider=provider)

    allowed_prefixes = _allowed_base_url_prefixes()
    if allowed_prefixes:
        normalized = custom.rstrip("/")
        if not any(normalized.startswith(prefix) for prefix in allowed_prefixes):
            raise LLMAdapterError(
                code="base_url_not_allowed",
                message="base_url is not in THB_ALLOWED_BASE_URL_PREFIXES.",
                status_code=400,
                retryable=False,
            )

    return custom


def _normalize_base_url(url: str) -> str:
    return url.strip().rstrip("/")


def _validate_custom_base_url(base_url: str, *, provider: Provider) -> None:
    parsed = urlparse(base_url)
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        raise LLMAdapterError(
            code="invalid_base_url",
            message="base_url must be a valid http(s) URL.",
            status_code=400,
            retryable=False,
        )

    if parsed.username or parsed.password:
        raise LLMAdapterError(
            code="invalid_base_url",
            message="base_url must not contain embedded credentials.",
            status_code=400,
            retryable=False,
        )

    host = (parsed.hostname or "").strip().lower()
    if not host:
        raise LLMAdapterError(
            code="invalid_base_url",
            message="base_url must include a hostname.",
            status_code=400,
            retryable=False,
        )

    if host in {"localhost"} or host.endswith(".local"):
        raise LLMAdapterError(
            code="unsafe_base_url",
            message=(
                f"base_url host is not allowed for public deployment (provider={provider})."
            ),
            status_code=400,
            retryable=False,
        )

    try:
        ip = ipaddress.ip_address(host)
    except ValueError:
        return

    if (
        ip.is_private
        or ip.is_loopback
        or ip.is_link_local
        or ip.is_multicast
        or ip.is_reserved
    ):
        raise LLMAdapterError(
            code="unsafe_base_url",
            message=(
                f"base_url host is not allowed for public deployment (provider={provider})."
            ),
            status_code=400,
            retryable=False,
        )


def _require_api_key(provider: str, api_key: str | None) -> str:
    key = (api_key or "").strip()
    if key:
        return key
    raise LLMAdapterError(
        code="missing_api_key",
        message=f"Provider '{provider}' requires api_key.",
        status_code=400,
        retryable=False,
    )


def _build_signal_digest(bundle: dict[str, Any]) -> dict[str, Any]:
    image_obs = _as_dict(bundle.get("image_observation"))
    clip_obs = _as_list_of_dicts(bundle.get("clip_signal_observations"))
    weight_obs = _as_dict(bundle.get("weight_observation"))
    metadata = _as_dict(bundle.get("ingestion_metadata"))
    extractor_metadata = _as_dict(metadata.get("extractor_metadata"))

    fallback_clip_ids = {
        str(item).strip()
        for item in _as_list(extractor_metadata.get("fallback_clip_ids"))
        if str(item).strip()
    }

    drifts: list[float] = []
    blurs: list[float] = []
    lips: list[float] = []
    reliable_lips: list[float] = []
    fallback_clip_count = 0
    high_risk_clip_ids: list[str] = []
    high_risk_clips: list[dict[str, Any]] = []
    quality_distribution: dict[str, int] = {
        "good": 0,
        "acceptable": 0,
        "marginal": 0,
        "poor": 0,
    }
    per_clip_breakdown: list[dict[str, Any]] = []
    per_clip_signals: list[dict[str, Any]] = []
    max_per_clip_signals = _max_clip_signals_limit()

    for index, item in enumerate(clip_obs):
        clip_id = str(item.get("clip_id", "unknown"))
        drift = _as_float(item.get("identity_cosine_drift"), default=0.0)
        blur = _as_float(item.get("blur_score"), default=1.0)
        lip_sync = _as_float(item.get("lip_sync_confidence"), default=1.0)
        fallback_used = _as_bool(item.get("fallback_extractor_used")) or clip_id in fallback_clip_ids
        lip_sync_reliable = not fallback_used
        if fallback_used:
            fallback_clip_count += 1

        drifts.append(drift)
        blurs.append(blur)
        lips.append(lip_sync)
        if lip_sync_reliable:
            reliable_lips.append(lip_sync)

        high_risk_reasons, warning_reasons = _clip_risk_reasons(
            drift=drift,
            blur=blur,
            lip_sync=lip_sync,
            lip_sync_reliable=lip_sync_reliable,
            fallback_used=fallback_used,
        )
        high_risk_reason_count = len(high_risk_reasons)
        is_high_risk = high_risk_reason_count >= 2
        has_warning = bool(warning_reasons) or high_risk_reason_count == 1
        risk_level = "high_risk" if is_high_risk else ("warning" if has_warning else "clean")
        quality_tier = _clip_quality_tier(
            drift=drift,
            blur=blur,
            lip_sync=lip_sync,
            lip_sync_reliable=lip_sync_reliable,
            fallback_used=fallback_used,
            high_risk_reason_count=high_risk_reason_count,
        )
        quality_distribution[quality_tier] += 1

        if is_high_risk:
            high_risk_clip_ids.append(clip_id)
            high_risk_clips.append(
                {
                    "clip_id": clip_id,
                    "reasons": high_risk_reasons,
                    "risk_level": risk_level,
                    "quality_tier": quality_tier,
                    "lip_sync_reliable": lip_sync_reliable,
                }
            )

        if index < max_per_clip_signals:
            per_clip_signals.append(
                {
                    "clip_id": clip_id,
                    "identity_cosine_drift": drift,
                    "drift_severity": _drift_severity(drift),
                    "blur_score": blur,
                    "lip_sync_confidence": lip_sync,
                    "phoneme_coverage_new": _as_int(item.get("phoneme_coverage_new")),
                    "blink_count": _as_int(item.get("blink_count")),
                    "occlusion_frames": _as_int(item.get("occlusion_frames")),
                    "fallback_extractor_used": fallback_used,
                    "provisional_due_to_fallback": fallback_used,
                    "lip_sync_reliable": lip_sync_reliable,
                    "is_high_risk": is_high_risk,
                    "risk_level": risk_level,
                    "high_risk_reasons": high_risk_reasons,
                    "warning_reasons": warning_reasons,
                    "quality_tier": quality_tier,
                }
            )

        per_clip_breakdown.append(
            {
                "clip_id": clip_id,
                "identity_cosine_drift": round(drift, 4),
                "blur_score": round(blur, 4),
                "lip_sync_confidence": round(lip_sync, 4),
                "landmark_stability_score": _as_float(item.get("landmark_stability_score")),
                "phoneme_coverage_new": _as_float(item.get("phoneme_coverage_new")),
                "exposure_score": _as_float(item.get("exposure_score")),
                "occlusion_frames": _as_int(item.get("occlusion_frames")),
                "fallback_extractor_used": fallback_used,
                "risk_flags": (
                    (["high_drift"] if drift >= _DRIFT_WARNING_THRESHOLD else [])
                    + (["low_blur"] if blur <= _BLUR_WARNING_THRESHOLD else [])
                    + (["poor_lip_sync"] if lip_sync <= _LIPSYNC_WARNING_THRESHOLD else [])
                ),
            }
        )

    layer_entropy = _as_dict(weight_obs.get("canonical_entropy_per_layer"))
    rank_util = _as_dict(weight_obs.get("layer_rank_utilization"))
    layer_sparsity = _as_dict(weight_obs.get("layer_sparsity"))
    gradient_noise = _as_float(weight_obs.get("gradient_noise_estimate"))
    high_entropy_positions = _as_list(weight_obs.get("high_entropy_token_positions"))
    token_map = weight_obs.get("token_position_to_phoneme")

    configured_fallback_count = _as_int(extractor_metadata.get("clip_extractor_fallback_count"))
    if configured_fallback_count > fallback_clip_count:
        fallback_clip_count = configured_fallback_count

    rank_values = [_as_float(value) for value in rank_util.values()]
    mean_rank_utilization = (
        round(sum(rank_values) / len(rank_values), 4)
        if rank_values
        else None
    )
    min_rank_layer = (
        min(rank_util, key=lambda key: _as_float(rank_util[key]))
        if rank_util
        else None
    )
    min_rank_value = _dict_min(rank_util) if rank_util else None

    worst_sparsity_layer = (
        max(layer_sparsity, key=lambda key: _as_float(layer_sparsity[key]))
        if layer_sparsity
        else None
    )
    worst_sparsity_value = (
        _as_float(layer_sparsity.get(worst_sparsity_layer))
        if worst_sparsity_layer is not None
        else None
    )

    overfit_sig = _as_float(weight_obs.get("overfitting_signature"))
    mean_rank_for_assessment = (
        sum(_as_float(v) for v in rank_util.values()) / max(len(rank_util), 1)
        if rank_util
        else 0.5
    )
    if overfit_sig >= 0.6:
        training_quality_assessment = "overfit"
    elif gradient_noise >= 0.5:
        training_quality_assessment = "unstable"
    elif mean_rank_for_assessment <= 0.3:
        training_quality_assessment = "underfit"
    else:
        training_quality_assessment = "healthy"

    digest: dict[str, Any] = {
        "case_id": bundle.get("case_id"),
        "prompt": str(bundle.get("prompt", "")),
        "param_config": _as_dict(bundle.get("param_config")),
        "ingestion_metadata": {
            "created_at_unix": metadata.get("created_at_unix"),
            "clip_extractor_fallback_count": fallback_clip_count,
            "fallback_clip_ids": sorted(fallback_clip_ids),
        },
    }

    if image_obs:
        digest["image_summary"] = {
            "face_occupancy_ratio": _as_float(image_obs.get("face_occupancy_ratio")),
            "estimated_sharpness": _as_float(image_obs.get("estimated_sharpness")),
            "estimated_yaw_degrees": _as_float(image_obs.get("estimated_yaw_degrees")),
            "estimated_pitch_degrees": _as_float(image_obs.get("estimated_pitch_degrees")),
            "lighting_uniformity_score": _as_float(image_obs.get("lighting_uniformity_score")),
            "background_complexity_score": _as_float(
                image_obs.get("background_complexity_score")
            ),
            "occlusion_detected": _as_bool(image_obs.get("occlusion_detected")),
            "prompt_token_count": _as_int(image_obs.get("prompt_token_count")),
            "prompt_semantic_density": _as_float(image_obs.get("prompt_semantic_density")),
            "conflicting_descriptors": _as_list(image_obs.get("conflicting_descriptors")),
            "identity_anchoring_strength": _as_float(
                image_obs.get("identity_anchoring_strength")
            ),
            "face_detection_measured": _as_bool(
                extractor_metadata.get("face_detection_measured"),
                default=True,
            ),
        }

    if clip_obs:
        fallback_clips_note = (
            "Clips extracted via fallback extractor use synthetic lip_sync_confidence=0.35 "
            "and empty phoneme_sequence; treat lip-sync and phoneme evidence as unreliable."
            if fallback_clip_count > 0
            else None
        )
        digest["clip_summary"] = {
            "clip_count": len(clip_obs),
            "fallback_clip_count": fallback_clip_count,
            "non_fallback_clip_count": max(0, len(clip_obs) - fallback_clip_count),
            "all_clips_fallback": fallback_clip_count == len(clip_obs),
            "parameter_tuning_reliability": (
                "low" if fallback_clip_count == len(clip_obs) else "normal"
            ),
            "high_risk_clip_ids": high_risk_clip_ids,
            "high_risk_clips": high_risk_clips,
            "mean_identity_drift": _safe_mean(drifts),
            "mean_blur_score": _safe_mean(blurs),
            "mean_lip_sync_confidence": _safe_mean(lips),
            "mean_lip_sync_confidence_reliable_only": _safe_mean(reliable_lips),
            "quality_distribution": quality_distribution,
            "per_clip_signals": per_clip_signals,
            "per_clip_breakdown": per_clip_breakdown,
            "per_clip_signals_truncated": len(clip_obs) > max_per_clip_signals,
            "fallback_clips_note": fallback_clips_note,
        }

    if weight_obs:
        digest["weight_summary"] = {
            "available": True,
            "weight_file_id": str(weight_obs.get("weight_file_id", "")),
            "lora_rank": _as_int(weight_obs.get("lora_rank")),
            "target_module_count": len(_as_list(weight_obs.get("target_modules"))),
            "max_canonical_entropy": _dict_max(layer_entropy),
            "min_rank_utilization": _dict_min(rank_util),
            "high_entropy_token_positions": high_entropy_positions[:24],
            "suspected_anomalous_phonemes": _phonemes_from_positions(
                high_entropy_positions,
                token_map,
            ),
            "overfitting_signature": overfit_sig,
            "layer_rank_utilization_summary": {
                "mean": mean_rank_utilization,
                "min_layer": min_rank_layer,
                "min_value": min_rank_value,
            },
            "worst_sparsity_layer": worst_sparsity_layer,
            "worst_sparsity_value": worst_sparsity_value,
            "gradient_noise_estimate": gradient_noise,
            "training_quality_assessment": training_quality_assessment,
        }

    return digest


def _normalize_task_tier(task_tier: TaskTier) -> Literal["image_audit", "clip_audit", "weight_audit"]:
    mapped = {
        "easy": "image_audit",
        "medium": "clip_audit",
        "hard": "weight_audit",
    }
    return mapped.get(task_tier, task_tier)  # type: ignore[return-value]


def _build_prompt(
    signal_digest: dict[str, Any],
    *,
    task_tier: TaskTier = "weight_audit",
) -> str:
    _ = _normalize_task_tier(task_tier)
    fallback_note = (
        _as_dict(signal_digest.get("clip_summary")).get("fallback_clips_note")
        if isinstance(signal_digest.get("clip_summary"), dict)
        else None
    )
    fallback_warning = (
        f"\n\nWARNING - FALLBACK EXTRACTOR ACTIVE: {fallback_note}"
        if isinstance(fallback_note, str) and fallback_note.strip()
        else ""
    )

    return (
        "Analyze the TalkingHeadBench signal digest below and produce a structured report.\n\n"
        "## Signal Thresholds (use these to classify findings)\n"
        "Identity drift severity: <0.05=none, <0.12=minor, <0.22=moderate, >=0.22=severe\n"
        "Lip sync quality: >=0.75=good, >=0.50=acceptable, >=0.20=poor, <0.20=absent\n"
        "Blur score: >=0.60=sharp, >=0.30=acceptable, <0.30=blurry\n"
        "Overfitting signature: <0.30=low, 0.30-0.59=medium, 0.60-0.79=high, >=0.80=critical\n"
        "Rank utilization mean: >=0.65=efficient, 0.30-0.64=wasteful, <0.30=collapsed\n"
        "Face occupancy ratio: >=0.20=good, 0.10-0.19=marginal, <0.10=face too small\n\n"
        "## Required Report Sections\n"
        "Write exactly these sections in order. Each section must cite specific digest values.\n\n"
        "### Overall Readiness\n"
        "1-2 sentences. State readiness level (ready / marginal / not ready) and the single most critical blocker with its measured value.\n\n"
        "### Critical Risks\n"
        "Bullet list. For each risk: name the affected signal, cite exact value, and classify severity with thresholds above. Minimum 1 bullet, maximum 5.\n\n"
        "### Parameter Fixes\n"
        "Directional only - no specific numbers. For each anomalous param in param_config, state direction to adjust and why, citing the signal that drives it. If param_config is reasonable, state that explicitly.\n\n"
        "### Data and Weight Concerns\n"
        "Per-clip breakdown: for each entry in per_clip_breakdown, state clip_id and risk_flags (or clean if none). Then summarize weight health using overfitting_signature, layer_rank_utilization_summary.mean, and worst_sparsity_layer.\n\n"
        "### Weight Status\n"
        "If weight_summary.available is false or absent, state 'Not uploaded - no data'. Otherwise state training quality assessment (overfit/healthy/underfit/unstable) based on training_quality_assessment and supporting fields.\n\n"
        "### Top 3 Next Actions\n"
        "Numbered list. Each action must be actionable, specific, and evidence-cited.\n\n"
        "### Confidence\n"
        "State confidence level (high/medium/low) and what limits it (fallback extractor, missing weights, or insufficient clips)."
        f"{fallback_warning}"
        "\n\n## Signal Digest\n"
        f"{json.dumps(signal_digest, indent=2, sort_keys=True)}"
    )


def _call_openai(
    *,
    model_id: str,
    api_key: str | None,
    prompt: str,
    max_tokens: int,
    temperature: float,
    timeout_s: float,
    base_url: str | None,
) -> str:
    key = _require_api_key("openai", api_key)
    base = _resolve_openai_base_url(base_url)
    return _call_chat_completions(
        provider="openai",
        base_url=base,
        model_id=model_id,
        api_key=key,
        prompt=prompt,
        max_tokens=max_tokens,
        temperature=temperature,
        timeout_s=timeout_s,
    )


def _call_anthropic(
    *,
    model_id: str,
    api_key: str | None,
    prompt: str,
    max_tokens: int,
    temperature: float,
    timeout_s: float,
    base_url: str | None,
) -> str:
    key = _require_api_key("anthropic", api_key)
    base = _resolve_anthropic_base_url(base_url)
    return _call_chat_completions(
        provider="anthropic",
        base_url=base,
        model_id=model_id,
        api_key=key,
        prompt=prompt,
        max_tokens=max_tokens,
        temperature=temperature,
        timeout_s=timeout_s,
    )


def _call_huggingface(
    *,
    model_id: str,
    api_key: str | None,
    prompt: str,
    max_tokens: int,
    temperature: float,
    timeout_s: float,
    base_url: str | None,
) -> str:
    key = _require_api_key("huggingface", api_key)
    routed_model = model_id
    if not base_url and ":" not in model_id.split("/")[-1]:
        routed_model = f"{model_id}:fastest"

    base = _resolve_huggingface_base_url(base_url)
    return _call_chat_completions(
        provider="huggingface",
        base_url=base,
        model_id=routed_model,
        api_key=key,
        prompt=prompt,
        max_tokens=max_tokens,
        temperature=temperature,
        timeout_s=timeout_s,
    )


def _call_local(
    *,
    model_id: str,
    api_key: str | None,
    prompt: str,
    max_tokens: int,
    temperature: float,
    timeout_s: float,
    base_url: str | None,
) -> str:
    key = (api_key or os.getenv("HF_TOKEN") or "local").strip()
    base = _resolve_local_base_url(base_url)
    return _call_chat_completions(
        provider="local",
        base_url=base,
        model_id=model_id,
        api_key=key,
        prompt=prompt,
        max_tokens=max_tokens,
        temperature=temperature,
        timeout_s=timeout_s,
    )


def _call_chat_completions(
    *,
    provider: str,
    base_url: str,
    model_id: str,
    api_key: str,
    prompt: str,
    max_tokens: int,
    temperature: float,
    timeout_s: float,
) -> str:
    client = OpenAI(api_key=api_key, base_url=base_url, timeout=timeout_s)

    try:
        response = client.chat.completions.create(
            model=model_id,
            messages=[
                {"role": "system", "content": _SYSTEM_PROMPT},
                {"role": "user", "content": prompt},
            ],
            temperature=temperature,
            max_tokens=max_tokens,
        )
    except APIStatusError as exc:
        status_code = int(exc.status_code or 502)
        message = _format_http_error(provider, status_code)
        raise LLMAdapterError(
            code="provider_http_error",
            message=message,
            status_code=502,
            retryable=status_code >= 500 or status_code == 429,
        ) from exc
    except APIConnectionError as exc:
        raise LLMAdapterError(
            code="provider_connection_error",
            message=f"Unable to reach {provider} provider endpoint.",
            status_code=502,
            retryable=True,
        ) from exc
    except Exception as exc:  # noqa: BLE001
        raise LLMAdapterError(
            code="provider_invalid_response",
            message=f"{provider} provider returned an unexpected response.",
            status_code=502,
            retryable=True,
        ) from exc

    choices = getattr(response, "choices", None) or []
    if not choices:
        raise LLMAdapterError(
            code="invalid_provider_response",
            message="Provider response did not include choices.",
            status_code=502,
            retryable=True,
        )

    content = choices[0].message.content
    text = _extract_text_content(content)
    if not text:
        raise LLMAdapterError(
            code="invalid_provider_response",
            message="Provider response did not include message content.",
            status_code=502,
            retryable=True,
        )

    return text


def _resolve_openai_base_url(base_url: str | None) -> str:
    url = (
        base_url
        or os.getenv("THB_OPENAI_BASE_URL")
        or os.getenv("API_BASE_URL")
        or "https://api.openai.com/v1"
    ).rstrip("/")
    return _strip_chat_completions_suffix(url)


def _resolve_anthropic_base_url(base_url: str | None) -> str:
    url = (
        base_url
        or os.getenv("THB_ANTHROPIC_BASE_URL")
        or "https://api.anthropic.com/v1"
    ).rstrip("/")
    return _strip_chat_completions_suffix(url)


def _resolve_huggingface_base_url(base_url: str | None) -> str:
    url = (
        base_url
        or os.getenv("THB_HF_BASE_URL")
        or os.getenv("API_BASE_URL")
        or "https://router.huggingface.co/v1"
    ).rstrip("/")
    return _strip_chat_completions_suffix(url)


def _resolve_local_base_url(base_url: str | None) -> str:
    raw = (
        base_url
        or os.getenv("THB_LOCAL_LLM_URL")
        or os.getenv("API_BASE_URL")
        or "http://localhost:11434/v1"
    ).rstrip("/")

    url = _strip_chat_completions_suffix(raw)
    if url.endswith("/api/generate"):
        url = f"{url[:-len('/api/generate')]}/v1"
    if not url.endswith("/v1"):
        url = f"{url}/v1"
    return url


def _strip_chat_completions_suffix(url: str) -> str:
    if url.endswith("/chat/completions"):
        return url[: -len("/chat/completions")]
    return url


def _format_http_error(provider: str, status_code: int) -> str:
    if status_code in {401, 403}:
        return f"Authentication failed for {provider} provider."
    if status_code == 404:
        return f"Requested model or endpoint was not found for {provider}."
    if status_code == 429:
        return f"Rate limit exceeded for {provider} provider."
    if status_code >= 500:
        return f"{provider} provider is temporarily unavailable."
    return f"{provider} provider request failed."


def _extract_text_content(content: Any) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for item in content:
            if isinstance(item, str):
                parts.append(item)
                continue
            if isinstance(item, dict):
                text = item.get("text") or item.get("content")
                if isinstance(text, str):
                    parts.append(text)
        return "\n".join(part for part in parts if part).strip()
    if isinstance(content, dict):
        text = content.get("text") or content.get("content")
        if isinstance(text, str):
            return text
    return ""


def _max_clip_signals_limit() -> int:
    raw = os.getenv("THB_MAX_CLIP_SIGNALS_IN_DIGEST", "24").strip()
    try:
        value = int(raw)
    except ValueError:
        return 24
    return max(1, value)


def _drift_severity(drift: float) -> str:
    if drift < 0.05:
        return "none"
    if drift < 0.12:
        return "minor"
    if drift < 0.22:
        return "moderate"
    return "severe"


def _clip_risk_reasons(
    *,
    drift: float,
    blur: float,
    lip_sync: float,
    lip_sync_reliable: bool,
    fallback_used: bool,
) -> tuple[list[str], list[str]]:
    high_risk_reasons: list[str] = []
    warning_reasons: list[str] = []

    if fallback_used:
        warning_reasons.append("fallback_extractor_used = true (proxy-only clip signals)")
        if drift >= _DRIFT_WARNING_THRESHOLD:
            warning_reasons.append("identity_cosine_drift proxy >= 0.22")
        if blur <= _BLUR_WARNING_THRESHOLD:
            warning_reasons.append("blur_score proxy <= 0.30")
        warning_reasons.append("lip_sync_confidence unavailable (fallback_extractor_used)")
        return high_risk_reasons, warning_reasons

    if drift >= _DRIFT_HIGH_RISK_THRESHOLD:
        high_risk_reasons.append("identity_cosine_drift >= 0.35")
    elif drift >= _DRIFT_WARNING_THRESHOLD:
        warning_reasons.append("identity_cosine_drift >= 0.22")

    if blur <= _BLUR_HIGH_RISK_THRESHOLD:
        high_risk_reasons.append("blur_score <= 0.20")
    elif blur <= _BLUR_WARNING_THRESHOLD:
        warning_reasons.append("blur_score <= 0.30")

    if lip_sync_reliable:
        if lip_sync <= _LIPSYNC_HIGH_RISK_THRESHOLD:
            high_risk_reasons.append("lip_sync_confidence <= 0.30")
        elif lip_sync <= _LIPSYNC_WARNING_THRESHOLD:
            warning_reasons.append("lip_sync_confidence <= 0.40")
    else:
        warning_reasons.append("lip_sync_confidence unavailable (fallback_extractor_used)")

    return high_risk_reasons, warning_reasons


def _clip_quality_tier(
    *,
    drift: float,
    blur: float,
    lip_sync: float,
    lip_sync_reliable: bool,
    fallback_used: bool,
    high_risk_reason_count: int,
) -> str:
    if fallback_used:
        if drift < _DRIFT_WARNING_THRESHOLD and blur >= 0.12:
            return "acceptable"
        if drift < _DRIFT_HIGH_RISK_THRESHOLD and blur >= 0.08:
            return "marginal"
        return "poor"

    if high_risk_reason_count >= 2:
        return "poor"
    if high_risk_reason_count == 1:
        return "marginal"

    is_good = drift < 0.12 and blur > 0.40 and (lip_sync > 0.50 or not lip_sync_reliable)
    if is_good:
        return "good" if lip_sync_reliable else "acceptable"

    is_acceptable = drift < _DRIFT_WARNING_THRESHOLD and blur > 0.25
    if is_acceptable and (lip_sync > _LIPSYNC_WARNING_THRESHOLD or not lip_sync_reliable):
        return "acceptable"

    is_marginal = drift < _DRIFT_HIGH_RISK_THRESHOLD and blur > _BLUR_HIGH_RISK_THRESHOLD
    if is_marginal and (lip_sync > _LIPSYNC_HIGH_RISK_THRESHOLD or not lip_sync_reliable):
        return "marginal"

    return "poor"


def _as_dict(value: Any) -> dict[str, Any]:
    return value if isinstance(value, dict) else {}


def _as_list(value: Any) -> list[Any]:
    return value if isinstance(value, list) else []


def _as_list_of_dicts(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        return []
    return [item for item in value if isinstance(item, dict)]


def _as_float(value: Any, *, default: float = 0.0) -> float:
    if isinstance(value, (int, float)):
        return float(value)
    return default


def _as_int(value: Any, *, default: int = 0) -> int:
    if isinstance(value, bool):
        return default
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        return int(value)
    return default


def _as_bool(value: Any, *, default: bool = False) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        cleaned = value.strip().lower()
        if cleaned in {"1", "true", "yes", "on"}:
            return True
        if cleaned in {"0", "false", "no", "off"}:
            return False
    return default


def _safe_mean(values: list[float]) -> float:
    clean = [value for value in values if isinstance(value, (int, float))]
    if not clean:
        return 0.0
    return round(float(sum(clean) / len(clean)), 4)


def _dict_max(values: dict[str, Any]) -> float:
    if not values:
        return 0.0
    return max(_as_float(v) for v in values.values())


def _dict_min(values: dict[str, Any]) -> float:
    if not values:
        return 0.0
    return min(_as_float(v) for v in values.values())


def _phonemes_from_positions(
    positions: list[Any],
    token_position_to_phoneme: Any,
) -> list[str]:
    if not isinstance(token_position_to_phoneme, dict):
        return []

    mapped: list[str] = []
    for position in positions:
        idx = _as_int(position, default=-1)
        if idx < 0:
            continue
        key = str(idx)
        value = token_position_to_phoneme.get(key)
        if value is None and idx in token_position_to_phoneme:
            value = token_position_to_phoneme.get(idx)
        if isinstance(value, str) and value not in mapped:
            mapped.append(value)
        if len(mapped) >= 16:
            break
    return mapped