services/audit_engine.py

"""
audit_engine.py — Qualora RAG-Augmented LLM Scoring Engine
===========================================================

Orchestrates the LLM inference pipeline, enforcing strict JSON schemas
and executing the Multi-Tier Fallback Cascade for high availability.

Architecture:
  1. retrieve_policy_context()   → ChromaDB / MongoDB Vector Search → policy chunks
  2. _fetch_agent_history()      → MongoDB db.audits → last 5 scores for context
  3. _build_prompt()             → structured prompt with all context layers
  4. _run_llm_cascade()          → Gemini → OpenRouter → Groq → HF Inference
  5. _parse_llm_response()       → validate & coerce via centralized JSON repair

Output schema (stable contract for frontend & MongoDB):
  {
    summary:                 str,
    agent_f1_score:          float  [0.0–1.0],
    satisfaction_prediction: str   ["High"|"Medium"|"Low"],
    compliance_risk:         str    ["Green"|"Amber"|"Red"],
    quality_matrix: {
      language_proficiency:  int  [0–10],
      cognitive_empathy:     int  [0–10],
      efficiency:            int  [0–10],
      bias_reduction:        int  [0–10],
      active_listening:      int  [0–10],
    },
    compliance_flags:        list[str],
    behavioral_nudges:       list[str],
    emotions: {
      agent:    str,
      customer: str,
    },
    _audit_metadata: {
      rag_provider:    str,
      llm_provider:    str,
      llm_model:       str,
      policy_chunks:   int,
      history_audits:  int,
      tier:            str,
      latency_ms:      int,
    }
  }
"""

import os
import json
import logging
import time
import hashlib
from datetime import datetime, timezone
import copy
import re as _re
from bson import ObjectId
from bson.errors import InvalidId

import httpx

# Optional provider SDKs: import lazily to avoid import-time crashes when
# environment lacks heavy vendor libraries. Use factory functions below
# to initialize clients only when needed.

from core import (
    GROQ_API_KEY, OPENROUTER_API_KEY,
    HF_SPACE_TOKEN, HF_SPACE_URL, ELEVENLABS_API_KEY, DEEPGRAM_API_KEY,
    get_db, repair_json, RAG_NULL_SENTINEL
)
from services.rag import retrieve_policy_context

log = logging.getLogger(__name__)


class LLMRateLimitError(RuntimeError):
    """Raised when a provider signals it is rate-limited so the cascade
    should skip this provider and move to the next tier immediately."""


# Optional Semantic Emotion Inference (sentence-transformers)
_ST_AVAILABLE = False
_st_model = None
try:
    from sentence_transformers import SentenceTransformer, util as st_util
    _ST_AVAILABLE = True
except Exception:
    SentenceTransformer = None
    st_util = None
    _ST_AVAILABLE = False
    log.warning("sentence_transformers not installed — falling back to keyword heuristics for emotion inference.")

# Caching for prototype embeddings (lazy init)
_st_proto_emb = None
_st_proto_map = None
_st_proto_texts = None
_st_prototypes_initialized = False

# ─────────────────────────────────────────────────────────────────────────────
# Transcription Cascade
# ─────────────────────────────────────────────────────────────────────────────

try:
    from gradio_client import Client as GradioClient, handle_file as gradio_handle_file
    GRADIO_CLIENT_AVAILABLE = True
except ImportError:
    GRADIO_CLIENT_AVAILABLE = False
    log.warning("gradio_client not installed.")

try:
    from deepgram import DeepgramClient
    DEEPGRAM_AVAILABLE = True
except ImportError:
    DeepgramClient = None
    DEEPGRAM_AVAILABLE = False
    log.warning("deepgram-sdk not installed.")

# Groq / ElevenLabs lazy initialization
try:
    from groq import Groq
    GROQ_AVAILABLE = True
except Exception:
    Groq = None
    GROQ_AVAILABLE = False
    log.warning("groq SDK not installed or unavailable.")

try:
    from elevenlabs.client import ElevenLabs
    ELEVENLABS_AVAILABLE = True
except Exception:
    ElevenLabs = None
    ELEVENLABS_AVAILABLE = False
    log.warning("elevenlabs SDK not installed or unavailable.")


def get_elevenlabs_client():
    if not ELEVENLABS_AVAILABLE or not ELEVENLABS_API_KEY:
        return None
    try:
        return ElevenLabs(api_key=ELEVENLABS_API_KEY)
    except Exception as e:
        log.warning("Failed to initialize ElevenLabs client: %s", e)
        return None


def get_groq_client():
    if not GROQ_AVAILABLE or not GROQ_API_KEY:
        return None
    try:
        return Groq(api_key=GROQ_API_KEY)
    except Exception as e:
        log.warning("Failed to initialize Groq client: %s", e)
        return None


def get_deepgram_client():
    if not DEEPGRAM_AVAILABLE or not DEEPGRAM_API_KEY:
        return None
    try:
        return DeepgramClient(api_key=DEEPGRAM_API_KEY)
    except Exception as e:
        log.warning("Failed to initialize Deepgram client: %s", e)
        return None

# Deterministic Persistence Cache for identical inputs
_AUDIT_CACHE = {}
_MAX_AUDIT_CACHE = int(os.environ.get("MAX_AUDIT_CACHE", "1024"))
_PERSIST_AUDIT_CACHE = os.environ.get("PERSIST_AUDIT_CACHE", "1") != "0"
_AUDIT_CACHE_COLLECTION = os.environ.get("AUDIT_CACHE_COLLECTION", "audit_cache")
_AUDIT_CACHE_TTL = int(os.environ.get("AUDIT_CACHE_TTL_SECONDS", "0"))
_CACHE_INDEXED = False
_LLM_TEMPERATURE = float(os.environ.get("AUDIT_LLM_TEMPERATURE", "0.0"))


def _get_cache_collection():
    """Return the MongoDB collection used for persisting audit cache or None.
    Ensures indexes are created once per process.
    """
    global _CACHE_INDEXED
    if not _PERSIST_AUDIT_CACHE:
        return None
    try:
        db = get_db()
        if db is None:
            return None
        coll = db[_AUDIT_CACHE_COLLECTION]
        if not _CACHE_INDEXED:
            try:
                coll.create_index([("cache_key", 1)], unique=True)
                if _AUDIT_CACHE_TTL > 0:
                    coll.create_index([("created_at", 1)], expireAfterSeconds=_AUDIT_CACHE_TTL)
            except Exception as e:
                log.warning("Failed to ensure audit cache indexes: %s", e)
            _CACHE_INDEXED = True
        return coll
    except Exception as e:
        log.warning("Failed to get DB collection for audit cache: %s", e)
        return None


def _canonicalize_transcript_for_cache(transcript: str) -> str:
    """Normalize transcript text so trivial formatting differences do not
    cause cache misses (e.g., CRLF vs LF, extra spaces, trailing whitespace).
    """
    if transcript is None:
        return ""

    txt = str(transcript).replace("\r\n", "\n").replace("\r", "\n")
    lines = []
    for ln in txt.split("\n"):
        # Collapse internal runs of spaces/tabs while preserving line breaks.
        lines.append(" ".join(ln.split()))

    # Remove duplicate blank lines and trim edges.
    out = []
    prev_blank = False
    for ln in lines:
        is_blank = (ln == "")
        if is_blank and prev_blank:
            continue
        out.append(ln)
        prev_blank = is_blank

    return "\n".join(out).strip()


def _policy_cache_fingerprint(org_id: str) -> str:
    """Return a lightweight fingerprint of KB state for cache invalidation.

    This prevents stale audits (computed before KB updates) from being reused
    after documents/chunks change for the same org.
    """
    try:
        db = get_db()
        if db is None:
            return "kb:none"

        safe_org = None
        try:
            safe_org = ObjectId(str(org_id))
        except (InvalidId, TypeError, ValueError):
            safe_org = None

        org_filter = {"org_id": safe_org} if safe_org is not None else {"org_id": org_id}

        docs_coll = getattr(db, "kb_docs", None)
        chunks_coll = getattr(db, "kb_chunks", None)

        doc_count = docs_coll.count_documents(org_filter) if docs_coll is not None else 0
        chunk_count = chunks_coll.count_documents(org_filter) if chunks_coll is not None else 0

        newest_doc = None
        if docs_coll is not None:
            newest_doc = docs_coll.find_one(org_filter, {"updated_at": 1, "created_at": 1}, sort=[("updated_at", -1), ("created_at", -1)])

        newest_chunk = None
        if chunks_coll is not None:
            newest_chunk = chunks_coll.find_one(org_filter, {"created_at": 1, "doc_id": 1}, sort=[("created_at", -1)])

        doc_ts = ""
        if newest_doc:
            ts = newest_doc.get("updated_at") or newest_doc.get("created_at")
            if ts is not None:
                doc_ts = ts.isoformat() if hasattr(ts, "isoformat") else str(ts)

        chunk_marker = ""
        if newest_chunk:
            did = newest_chunk.get("doc_id")
            cts = newest_chunk.get("created_at")
            chunk_marker = f"{did}:{cts.isoformat() if hasattr(cts, 'isoformat') else cts}"

        return f"kb:d{doc_count}:c{chunk_count}:dt{doc_ts}:cm{chunk_marker}"
    except Exception as e:
        log.warning("Failed to build policy fingerprint for cache key: %s", e)
        return "kb:error"

def _build_acoustic_context(speaker_profiles: dict) -> str:
    """
    Converts acoustic profiles (pitch, intensity, emotion) into a plain-English preamble.
    Grounds LLM analysis in real physical signal evidence (pyannote/SpeechBrain).
    """
    # Accept either a plain speaker_profiles mapping OR the acoustic_profile
    # wrapper used by the caller (e.g. { 'speaker_profiles': {...}, 'turns': [...] }).
    if not speaker_profiles:
        return ""

    # Unwrap wrapper if present
    sp = speaker_profiles.get('speaker_profiles') if isinstance(speaker_profiles, dict) and 'speaker_profiles' in speaker_profiles else speaker_profiles

    # Normalize list-form profiles into a dict {Speaker N: profile}
    if isinstance(sp, list):
        norm = {}
        for i, item in enumerate(sp):
            if isinstance(item, dict):
                key = item.get('speaker') or item.get('speaker_id') or item.get('spk') or f"Speaker {i+1}"
                norm[str(key)] = item
            else:
                norm[f"Speaker {i+1}"] = {}
        sp = norm

    if not isinstance(sp, dict):
        return ""

    lines = ["\n\n[ACOUSTIC SIGNAL ANALYSIS — from pyannote 3.1 + parselmouth + SpeechBrain wav2vec2]"]
    for spk, profile in sp.items():
        # Defensive: ensure profile is a dict-like object
        if not isinstance(profile, dict):
            try:
                profile = dict(profile)
            except Exception:
                profile = {}

        pitch  = profile.get("avg_pitch_hz")
        intens = profile.get("avg_intensity_db")
        emo    = profile.get("dominant_emotion", "unknown")
        turns  = profile.get("turn_count", 0)
        
        # Male baseline ~120 Hz, Female baseline ~210 Hz; elevated pitch → stress/anxiety
        pitch_note = ""
        if pitch:
            if pitch > 260: pitch_note = " (elevated — stress signal)"
            elif pitch < 100: pitch_note = " (depressed — fatigue signal)"
            
        line = f"  {spk}: avg_pitch={pitch:.0f}Hz{pitch_note}" if pitch else f"  {spk}:"
        if intens: line += f", avg_intensity={intens:.1f}dB"
        line += f", dominant_acoustic_emotion={emo}, turn_count={turns}"
        lines.append(line)
        
    lines.append("Use these acoustic signals to enrich scoring — especially cognitive_empathy "
                 "and compliance_risk. Acoustic evidence is ground-truth, not NLP inferred.")
    return "\n".join(lines)


def transcribe_via_hf_space(audio_path: str) -> dict:
    if not GRADIO_CLIENT_AVAILABLE:
        raise RuntimeError("gradio_client package not installed.")
    if not HF_SPACE_URL:
        raise RuntimeError("HF_SPACE_URL env var not set.")
    if not os.path.exists(audio_path):
        raise FileNotFoundError(f"Audio file not found: {audio_path}")
    
    log.info("--- [HF Space T1] Connecting to Qualora ASR node ---")
    token = HF_SPACE_TOKEN if HF_SPACE_TOKEN else None
    # gradio-client API changed from `token=` to `hf_token=` in newer releases.
    # Keep both paths for backward/forward compatibility.
    try:
        if token:
            client = GradioClient(HF_SPACE_URL, hf_token=token)
        else:
            client = GradioClient(HF_SPACE_URL)
    except TypeError:
        client = GradioClient(HF_SPACE_URL, token=token) if token else GradioClient(HF_SPACE_URL)
    raw = client.predict(gradio_handle_file(audio_path), api_name="/predict")
    
    if isinstance(raw, dict):
        data = raw
    elif isinstance(raw, str):
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            return {"transcript": raw.strip(), "speaker_profiles": {}, "turns": []}
    else:
        data = {"transcript": str(raw), "speaker_profiles": {}, "turns": []}
    
    if "error" in data:
        raise RuntimeError(f"HF Space returned error: {data['error']}")
    
    transcript = str(data.get("transcript", "")).strip()
    if not transcript:
        raise RuntimeError("HF Space returned empty transcript.")
    
    log.info("✅ [HF Space T1] Done.")
    return {
        "transcript": transcript,
        "speaker_profiles": data.get("speaker_profiles", {}),
        "turns": data.get("turns", []),
    }


def _elevenlabs_transcribe(audio_path: str) -> str:
    client = get_elevenlabs_client()
    if not client:
        raise RuntimeError("ElevenLabs client not configured or SDK missing.")
    with open(audio_path, "rb") as f:
        response = client.speech_to_text.convert(
            file=f, model_id="scribe_v1", diarize=True,
            language_code="en", timestamps_granularity="word"
        )
    
    lines = []
    if hasattr(response, 'words') and response.words:
        current_speaker = None
        current_text = []
        for w in response.words:
            spk = getattr(w, 'speaker_id', None)
            txt = getattr(w, 'text', '') or getattr(w, 'punctuated_word', '')
            if not txt.strip():
                continue
            if spk != current_speaker:
                if current_speaker is not None and current_text:
                    label = str(current_speaker).replace('speaker_', 'Speaker ').strip()
                    lines.append(f"{label}: {' '.join(current_text).strip()}")
                current_speaker = spk
                current_text = [txt.strip()]
            else:
                current_text.append(txt.strip())
        if current_speaker is not None and current_text:
            label = str(current_speaker).replace('speaker_', 'Speaker ').strip()
            lines.append(f"{label}: {' '.join(current_text).strip()}")
    
    if lines:
        return "\n\n".join(lines)
    if hasattr(response, 'text') and response.text:
        return response.text
    return str(response)


def _deepgram_transcribe(audio_path: str) -> str:
    client = get_deepgram_client()
    if not client:
        raise RuntimeError("Deepgram client not configured or SDK missing.")

    with open(audio_path, "rb") as f:
        buf = f.read()

    payload = {"buffer": buf}
    options = {
        "model": "nova-2", "smart_format": True, "diarize": True,
        "punctuate": True, "utterances": True, "language": "en"
    }
    
    try:
        response = client.listen.rest.v("1").transcribe_file(payload, options)
        resp_dict = response if isinstance(response, dict) else (response.to_dict() if hasattr(response, 'to_dict') else {})
        
        if "results" in resp_dict and "utterances" in resp_dict["results"]:
            lines = [f"Speaker {u['speaker']}: {u['transcript']}" for u in resp_dict["results"]["utterances"]]
            if lines:
                return "\n\n".join(lines)
        
        if "results" in resp_dict and "channels" in resp_dict["results"]:
            alt = resp_dict["results"]["channels"][0]["alternatives"][0]
            if "transcript" in alt:
                return alt["transcript"]
    except Exception as e:
        raise RuntimeError(f"Deepgram transcription failed: {str(e)}")
    
    raise RuntimeError("Deepgram response contained no parseable transcript.")


def _groq_transcribe(audio_path: str) -> str:
    client = get_groq_client()
    if not client:
        raise RuntimeError("Groq client not configured or SDK missing.")
    models = [("whisper-large-v3", "T1"), ("whisper-large-v3-turbo", "T2")]
    last_err = None
    for model, tier in models:
        try:
            with open(audio_path, "rb") as f:
                transcription = client.audio.transcriptions.create(
                    file=(os.path.basename(audio_path), f.read()),
                    model=model, response_format="verbose_json",
                    language="en", temperature=0.0
                )
            if hasattr(transcription, 'text') and transcription.text and transcription.text.strip():
                log.info(f"✅ [Groq Whisper] [{tier}] {model} transcribed successfully.")
                return transcription.text
            else:
                log.warning(f"⚠️  [{tier}] {model} returned empty transcript.")
                last_err = RuntimeError(f"{model} returned empty transcript")
        except Exception as e:
            log.warning(f"⚠️  [{tier}] {model} failed: {e}")
            last_err = e
    raise RuntimeError(f"All Groq Whisper models failed. Last error: {last_err}")


def perform_voice_capture_apis(audio_path: str) -> tuple:
    # Try ElevenLabs -> Deepgram -> Groq in order, using lazy factories
    try:
        log.info("--- [API Chain T2a] ElevenLabs Scribe (if available) ---")
        return _elevenlabs_transcribe(audio_path), "ElevenLabs Scribe"
    except Exception as e:
        log.warning(f"⚠️  [ElevenLabs] {e}")

    try:
        log.info("--- [API Chain T2b] Deepgram Nova-2 (if available) ---")
        return _deepgram_transcribe(audio_path), "Deepgram Nova-2"
    except Exception as e:
        log.warning(f"⚠️  [Deepgram] {e}")

    try:
        log.info("--- [API Chain T2c] Groq Whisper-large-v3 (if available) ---")
        return _groq_transcribe(audio_path), "Groq Whisper-large-v3"
    except Exception as e:
        log.warning(f"⚠️  [Groq] {e}")

    raise RuntimeError("All configured API-chain transcription providers failed.")


def _get_fallbacks_available() -> list:
    out = []
    if ELEVENLABS_API_KEY: out.append("elevenlabs")
    if DEEPGRAM_API_KEY:   out.append("deepgram")
    if GROQ_API_KEY:       out.append("groq")
    return out


# ─────────────────────────────────────────────────────────────────────────────
# Structured Prompt Template
# ─────────────────────────────────────────────────────────────────────────────

_SYSTEM_PROMPT = """You are Qualora, a Strict Quality Auditor AI acting as a Legal Judge.
Your absolute and only source of truth is the provided [POLICY CONTEXT] (Input A).

SPEAKER ROLE INFERENCE:
Analyze the [INTERACTION TRANSCRIPT] to resolve generic labels (e.g., SPEAKER_00) to Agent or Customer.
- AGENT side: initiates greetings, offers solutions, references policies, stays professional.
- CUSTOMER side: describes a problem, asks for help, expressing a need for service.
Once identified, apply consistently for all turns.

SECURITY MANDATE:
1. Meticulously scrub the transcript for violations matching constraints of Input A.
2. YOU MUST strictly adhere to Input A (the provided policies). Do not invent or hallucinate corporate rules outside of what is provided in Input A.
3. For every compliance flag, cite the specific clause from Input A evaluating it. Treat Input B strictly as Evidence.
4. Return ONLY valid JSON matching the exact requested schema. No markdown formatting.
"""

_USER_PROMPT_TEMPLATE = """## AUDIT TASK

Evaluate the following customer support interaction transcript and return a quality audit in JSON.

---
### POLICY CONTEXT (from KB)
{policy_context}

---
### AGENT HISTORICAL PERFORMANCE (last {history_count} audits)
{agent_history}

---
{acoustic_context}

---
### INTERACTION TRANSCRIPT
<transcript>
{transcript}
</transcript>

---
### INSTRUCTIONS
Score the agent objectively. Use the policy context to flag compliance violations.
Use agent history for calibration only — do not let it inflate scores.

Return ONLY this JSON schema (no other text):
{{
  "summary": "<2-3 sentence interaction summary>",
  "agent_f1_score": <float 0.0-1.0>,
  "satisfaction_prediction": "<High|Medium|Low>",
  "compliance_risk": "<Green|Amber|Red>",
  "quality_matrix": {{
    "language_proficiency": <int 0-10>,
    "cognitive_empathy": <int 0-10>,
    "efficiency": <int 0-10>,
    "bias_reduction": <int 0-10>,
    "active_listening": <int 0-10>
  }},
  "compliance_flags": ["<flag 1>", "<flag 2>"],
  "behavioral_nudges": ["<nudge 1>", "<nudge 2>", "<nudge 3>"],
    "emotions": {{
        "agent": "<dominant emotion>",
        "customer": "<dominant emotion>",
        "timeline": [{{ "turn": <int>, "speaker": "<Agent|Customer>", "emotion": "<label>", "intensity": <int 1-10> }}]
    }}
}}"""

_DEFAULT_QUALITY_POLICY = """# STANDARD QUALITY AUDIT POLICY (Global Baseline)

1. GREETING & BRANDING:
   - Must use an approved opening script (Greeting + Name + Offer of assistance).
   - Tone must be professional and welcoming.

2. COGNITIVE EMPATHY & ACTIVE LISTENING:
   - Acknowledge customer sentiment explicitly (e.g., "I understand how frustrating that must be").
   - Confirm customer intent before proceeding with a solution.
   - Avoid interruptions; allow the customer to complete their thought.

3. PROBLEM RESOLUTION & ACCURACY:
   - Provide accurate, policy-aligned information.
   - Proactively offer solutions rather than wait for customer prompts.
   - If an issue cannot be resolved, explain why clearly and offer an escalation path or follow-up timeframe.

4. EFFICIENCY & PROTOCOL:
   - Maintain a logical flow without excessive silence or dead-air.
   - Use transition statements when placing the customer on hold.
   - Summarize the resolution or next steps before closing the interaction.

5. BIAS REDUCTION & PROFESSIONALISM:
   - Maintain neutrality and objectivity at all times.
   - Treat all customers with equal dignity and respect regardless of their communication style.
"""

_FALLBACK = {
    "summary": "Audit completed. Review the quality matrix for details.",
    "agent_f1_score": 0.70,
    "quality_matrix": {
        "language_proficiency": 7,
        "cognitive_empathy": 7,
        "efficiency": 7,
        "bias_reduction": 9,
        "active_listening": 7
    },
    "compliance_risk": "Amber",
    "emotional_timeline": [],
    "behavioral_nudges": [
        "Internalize policy constraints for more accurate scoring.",
        "Ensure all customer concerns are addressed before closing."
    ],
    "citations": ["General Quality Benchmark"]
}

_SUMMARIZATION_PROMPT = """You are a Quality Audit Refiner.
Your task is to distilled the provided middle section of a support call transcript.
Focus ONLY on:
1. Agent troubleshooting steps taken.
2. Customer sentiment and emotional shifts.
3. Any mention of policies or rules.
Return a dense, bulleted summary (Max 500 characters)."""


def _normalize_speaker_role(speaker_label: str) -> str:
    """Heuristic mapping of speaker labels to either 'Agent' or 'Customer'.
    Keeps logic conservative: prefer 'Agent' only when explicit cues present.
    """
    if not speaker_label:
        return 'Customer'
    s = speaker_label.lower()
    if any(k in s for k in ['agent', 'rep', 'support', 'csr', 'advisor']):
        return 'Agent'
    if any(k in s for k in ['customer', 'client', 'caller', 'user']):
        return 'Customer'
    # If label looks like 'speaker 0' prefer Customer for safety
    if 'speaker' in s or _re.match(r'spk|s\d', s):
        return 'Customer'
    # Default fallback
    return 'Customer'


def _infer_emotional_timeline_v2(transcript: str, max_turns: int = 120) -> list:
    """Semantic emotion inference using sentence-transformers when available.

    Produces a list of {turn, speaker, emotion, intensity} entries.
    Falls back to empty list on any failure so callers can use legacy logic.
    """
    if not transcript or not transcript.strip():
        return []

    # Parse simple speaker: utterance lines
    lines = [ln.strip() for ln in transcript.split('\n') if ':' in ln]
    if not lines:
        return []

    # Limit to a reasonable number of turns to avoid excessive compute
    lines = lines[:max_turns]

    # Define emotion prototypes for semantic matching (tuned probes)
    EMOTION_PROBES = {
        'Angry': [
            "This is unacceptable and I want a refund.",
            "I am furious about this service.",
            "I will escalate this and consider legal action."
        ],
        'Frustrated': [
            "This keeps failing and it's frustrating.",
            "I've repeated these steps multiple times.",
            "This is very annoying and slow." 
        ],
        'Anxious': [
            "I'm worried about this situation.",
            "I'm anxious and concerned about the outcome.",
            "This makes me nervous."
        ],
        'Confused': [
            "I don't understand what's happening.",
            "This is confusing to me.",
            "Can you explain that again?"
        ],
        'Neutral': ["Okay.", "I see.", "Understood."],
        'Professional': [
            "Thank you for your assistance.",
            "Understood, moving forward.",
            "I appreciate the help."
        ],
        'Calm': ["That's fine.", "No problem.", "All good."],
        'Empathetic': [
            "I understand how frustrating that must be.",
            "I appreciate you telling me that.",
            "That sounds difficult, I'm sorry to hear that."
        ],
        'Relieved': ["That's a relief.", "I'm glad that's resolved.", "Phew, thanks."],
        'Satisfied': ["That works, thank you.", "Great, thanks!", "Perfect."] ,
        'Happy': ["Awesome!", "That's fantastic!", "Love it!"]
    }

    # Semantic path: use cached prototype embeddings + batched encoding
    global _st_proto_emb, _st_proto_map, _st_proto_texts, _st_prototypes_initialized, _st_model
    try:
        # Build speaker/text arrays
        speakers = []
        texts = []
        turns = []
        for i, ln in enumerate(lines):
            try:
                spk, txt = ln.split(':', 1)
                txt = txt.strip()
                if not txt:
                    continue
                speakers.append(spk.strip())
                texts.append(txt)
                turns.append(i+1)
            except Exception:
                continue

        if not texts:
            return []

        # Lazy-init prototype texts and embeddings
        if not _st_prototypes_initialized:
            pt = []
            pm = []
            for emo, probes in EMOTION_PROBES.items():
                for p in probes:
                    pt.append(p)
                    pm.append(emo)
            _st_proto_texts = pt
            _st_proto_map = pm

            if _ST_AVAILABLE:
                if _st_model is None:
                    try:
                        _st_model = SentenceTransformer('all-MiniLM-L6-v2')
                    except Exception as e:
                        log.warning('Failed to load ST model: %s', e)
                        _st_model = None

                if _st_model is not None:
                    try:
                        _st_proto_emb = _st_model.encode(_st_proto_texts, convert_to_tensor=True)
                    except Exception as e:
                        log.warning('Failed to encode prototype texts: %s', e)
                        _st_proto_emb = None

            _st_prototypes_initialized = True

        # If embeddings available, do batched similarity search
        results = []
        if _ST_AVAILABLE and _st_model is not None and _st_proto_emb is not None:
            try:
                batch_size = 64
                n = len(texts)
                for start in range(0, n, batch_size):
                    batch_texts = texts[start:start+batch_size]
                    batch_emb = _st_model.encode(batch_texts, convert_to_tensor=True)
                    sims = st_util.cos_sim(batch_emb, _st_proto_emb)
                    for bi in range(sims.shape[0]):
                        row = sims[bi].cpu().tolist()
                        best_idx = int(max(range(len(row)), key=lambda k: row[k]))
                        best_score = float(row[best_idx])
                        best_emo = _st_proto_map[best_idx]

                        base_threshold = 0.25
                        max_range = 0.75
                        if best_score < base_threshold:
                            chosen = 'Neutral'
                            intensity = 3
                        else:
                            chosen = best_emo
                            scaled = max(0.0, min(1.0, (best_score - base_threshold) / max_range))
                            intensity = int(round(3 + scaled * 7))

                        text = batch_texts[bi]
                        lower = text.lower()
                        if any(k in lower for k in ['unacceptable', 'sue', 'refund', 'never', 'disgusting', 'scam']):
                            chosen = 'Angry'
                            intensity = max(intensity, 9)
                        if '!' in text and intensity < 9:
                            intensity = min(10, intensity + 1)

                        idx = start + bi
                        results.append({"turn": turns[idx], "speaker": speakers[idx], "emotion": chosen, "intensity": intensity})

                if results:
                    return results
            except Exception as e:
                log.warning('Embedding batch processing failed: %s', e)

    except Exception as e:
        log.warning('Semantic emotion inference path failed: %s', e)

    # Fallback heuristic mapping for each turn (deterministic)
    inferred = []
    for i, txt in enumerate(texts):
        tnum = turns[i]
        spk = speakers[i]
        text_lower = txt.lower()
        emo, intensity = 'Neutral', 3
        if any(w in text_lower for w in ['sorry', 'apologize', 'understand', 'apologies']):
            emo, intensity = 'Empathetic', 7
        elif any(w in text_lower for w in ['angry', 'mad', 'unacceptable', 'sue', 'complain', 'furious']):
            emo, intensity = 'Angry', 9
        elif any(w in text_lower for w in ['thank', 'great', 'excellent', 'awesome', 'fantastic']):
            emo, intensity = 'Satisfied', 8
        elif any(w in text_lower for w in ['confused', "don't understand", 'not sure', "i'm not sure", 'what do you mean']):
            emo, intensity = 'Confused', 6
        elif any(w in text_lower for w in ['worried', 'anxious', 'concerned', 'nervous']):
            emo, intensity = 'Anxious', 7
        inferred.append({"turn": tnum, "speaker": spk, "emotion": emo, "intensity": intensity})

    return inferred


def _apply_defensive_merge(parsed_audit: dict, transcript: str, acoustic_profile: dict = None) -> dict:
    """
    Applies v1-style safety nets: fills missing keys, infers F1 mathematically,
    and generates a keyword-based emotional timeline if the LLM output is empty.
    """
    # 1. Base key merge
    for key, val in _FALLBACK.items():
        if key not in parsed_audit or parsed_audit[key] is None:
            parsed_audit[key] = copy.deepcopy(val)
    
    # 2. Quality Matrix inference
    qm = parsed_audit.get("quality_matrix", {})
    if not isinstance(qm, dict): qm = {}
    for k, v in _FALLBACK["quality_matrix"].items():
        if k not in qm or not isinstance(qm[k], (int, float)):
            qm[k] = v
    parsed_audit["quality_matrix"] = qm
    
    # 3. Mathematical F1 Logical Inference (Harmonic Mean of Matrix)
    f1 = parsed_audit.get("agent_f1_score")
    if f1 is None or not isinstance(f1, (int, float)) or f1 == 0:
        # p = proficiency/efficiency/bias, r = empathy/listening
        p_val = (qm.get("language_proficiency", 5) + qm.get("efficiency", 5) + qm.get("bias_reduction", 5)) / 30.0
        r_val = (qm.get("cognitive_empathy", 5) + qm.get("active_listening", 5)) / 20.0
        if (p_val + r_val) > 0:
            parsed_audit["agent_f1_score"] = round((2 * p_val * r_val) / (p_val + r_val), 2)
        else:
            parsed_audit["agent_f1_score"] = 0.50

    # 4. Emotional Timeline Inference and LLM + Embedding Fusion
    # Prefer any LLM-provided timeline, but augment/fill with semantic inference.
    llm_timeline = []
    if isinstance(parsed_audit.get("emotional_timeline"), list) and parsed_audit.get("emotional_timeline"):
        llm_timeline = parsed_audit.get("emotional_timeline")
    elif isinstance(parsed_audit.get("emotions", {}).get("timeline"), list) and parsed_audit.get("emotions", {}).get("timeline"):
        llm_timeline = parsed_audit.get("emotions", {}).get("timeline")

    if llm_timeline:
        # If the LLM already supplied a timeline, run semantic inference to fill gaps
        try:
            # If diarization/turns were provided by the STT stage, prefer
            # that structured turns data when running semantic inference so
            # speaker tokens remain accurate (avoids misclassification).
            if acoustic_profile and isinstance(acoustic_profile, dict) and acoustic_profile.get('turns'):
                try:
                    turns = acoustic_profile.get('turns') or []
                    turn_lines = []
                    for ti, tr in enumerate(turns):
                        if isinstance(tr, dict):
                            spk = tr.get('speaker') or tr.get('spk') or tr.get('speaker_id') or f"Speaker {ti+1}"
                            txt = tr.get('text') or tr.get('transcript') or tr.get('utterance') or ''
                        else:
                            # Already a formatted string like 'Speaker 1: text'
                            line = str(tr).strip()
                            if ':' in line:
                                turn_lines.append(line)
                                continue
                            spk = f"Speaker {ti+1}"
                            txt = line
                        turn_lines.append(f"{spk}: {txt}")
                    turns_transcript = '\n'.join([l for l in turn_lines if l.strip()])
                    sem = _infer_emotional_timeline_v2(turns_transcript)
                except Exception:
                    sem = _infer_emotional_timeline_v2(transcript)
            else:
                sem = _infer_emotional_timeline_v2(transcript)
        except Exception as e:
            log.warning("Semantic augmentation failed: %s", e)
            sem = []

        fused_map = {}
        # Add LLM entries first
        for e in llm_timeline:
            try:
                tnum = int(e.get("turn", 0))
                fused_map[tnum] = {
                    "turn": tnum,
                    "speaker": str(e.get("speaker", "")).strip(),
                    "emotion": str(e.get("emotion", "Neutral")),
                    "intensity": float(e.get("intensity", 3)) if e.get("intensity") is not None else 3
                }
            except Exception:
                continue

        # Use semantic entries to fill missing turns or calibrate intensity
        for se in (sem or []):
            try:
                tnum = int(se.get("turn", 0))
                if tnum not in fused_map:
                    fused_map[tnum] = se
                else:
                    # Average intensity when both present and numeric
                    try:
                        li = float(fused_map[tnum].get("intensity", 3))
                        si = float(se.get("intensity", 0))
                        fused_map[tnum]["intensity"] = round((li + si) / 2, 1)
                    except Exception:
                        pass
            except Exception:
                continue

        final = [fused_map[k] for k in sorted(fused_map.keys())]
        parsed_audit["emotional_timeline"] = final
        if "emotions" not in parsed_audit or not isinstance(parsed_audit["emotions"], dict):
            parsed_audit["emotions"] = parsed_audit.get("emotions", {})
        parsed_audit["emotions"]["timeline"] = final

    else:
        # If no LLM timeline, try semantic inference then fallback heuristics
        try:
            # Prefer diarization turns when present for inference
            if acoustic_profile and isinstance(acoustic_profile, dict) and acoustic_profile.get('turns'):
                turns = acoustic_profile.get('turns') or []
                turn_lines = []
                for ti, tr in enumerate(turns):
                    if isinstance(tr, dict):
                        spk = tr.get('speaker') or tr.get('spk') or tr.get('speaker_id') or f"Speaker {ti+1}"
                        txt = tr.get('text') or tr.get('transcript') or tr.get('utterance') or ''
                    else:
                        line = str(tr).strip()
                        if ':' in line:
                            turn_lines.append(line)
                            continue
                        spk = f"Speaker {ti+1}"
                        txt = line
                    turn_lines.append(f"{spk}: {txt}")
                turns_transcript = '\n'.join([l for l in turn_lines if l.strip()])
                inferred = _infer_emotional_timeline_v2(turns_transcript)
            else:
                inferred = _infer_emotional_timeline_v2(transcript)
        except Exception as e:
            log.warning("Advanced emotional inference failed: %s", e)
            inferred = []

        if not inferred:
            # Legacy lightweight keyword fallback (small, deterministic)
            inferred = []
            lines = [line.strip() for line in transcript.split('\n') if ':' in line][:30]
            for i, line in enumerate(lines):
                try:
                    spk, text = line.split(':', 1)
                    text_lower = text.lower()
                    emo, intensity = 'Neutral', 3
                    if any(w in text_lower for w in ['sorry', 'apologize', 'understand']):
                        emo, intensity = 'Empathetic', 7
                    elif any(w in text_lower for w in ['angry', 'mad', 'unacceptable', 'sue', 'complain']):
                        emo, intensity = 'Angry', 9
                    elif any(w in text_lower for w in ['thank', 'great', 'excellent', 'awesome']):
                        emo, intensity = 'Satisfied', 8
                    inferred.append({"turn": i+1, "speaker": spk.strip(), "emotion": emo, "intensity": intensity})
                except Exception:
                    continue

        parsed_audit["emotional_timeline"] = inferred or [{"turn": 1, "speaker": "System", "emotion": "Neutral", "intensity": 5}]
        if "emotions" not in parsed_audit or not isinstance(parsed_audit["emotions"], dict):
            parsed_audit["emotions"] = parsed_audit.get("emotions", {})
        parsed_audit["emotions"]["timeline"] = parsed_audit["emotional_timeline"]

    # Aggregate dominant emotion per role (agent/customer) when possible
    try:
        # If diarization/acoustic metadata is present, skip automatic
        # role aggregation here to avoid erroneously mapping numeric
        # speaker tokens to 'Customer'. The frontend (which has direct
        # access to `turns`/`speaker_profiles`) will handle role mapping
        # deterministically when rendering charts.
        if acoustic_profile and isinstance(acoustic_profile, dict) and acoustic_profile.get('turns'):
            # Preserve per-turn timeline but do not aggregate into Agent/Customer
            pass
        else:
            role_map = {}
            for item in parsed_audit["emotional_timeline"]:
                spk = str(item.get("speaker", "")).strip()
                role = _normalize_speaker_role(spk)
                role_map.setdefault(role, []).append(item)

            for role_label in ("Agent", "Customer"):
                entries = role_map.get(role_label, [])
                if not entries:
                    parsed_audit["emotions"][role_label.lower()] = parsed_audit.get("emotions", {}).get(role_label.lower(), "neutral")
                    continue
                counts = {}
                intens = {}
                for e in entries:
                    emo = e.get("emotion", "Neutral")
                    counts[emo] = counts.get(emo, 0) + 1
                    try:
                        intens[emo] = intens.get(emo, 0) + float(e.get("intensity") or 0)
                    except Exception:
                        intens[emo] = intens.get(emo, 0) + 0
                best = max(counts.keys(), key=lambda k: (counts[k], (intens.get(k, 0) / counts[k]) if counts[k] else 0))
                avg_int = round((intens.get(best, 0) / counts.get(best, 1)), 1)
                parsed_audit["emotions"][role_label.lower()] = str(best)
                parsed_audit["emotions"][f"{role_label.lower()}_intensity"] = avg_int
    except Exception:
        pass

    return parsed_audit

# ─────────────────────────────────────────────────────────────────────────────
# Context Fetchers
# ─────────────────────────────────────────────────────────────────────────────

def _fetch_agent_history(org_id: str, limit: int = 5) -> tuple[str, int]:
    """Pull the last `limit` audit quality matrices from MongoDB for this org."""
    try:
        db = get_db()
        if db is None:
            return "No historical data available.", 0

        try:
            safe_org_id = ObjectId(str(org_id))
        except (InvalidId, ValueError, TypeError):
            return "Invalid Organization ID.", 0

        docs = list(
            db.audits.find(
                {"org_id": safe_org_id, "audit.agent_f1_score": {"$exists": True}},
                {"audit.agent_f1_score": 1, "audit.compliance_risk": 1,
                 "audit.quality_matrix": 1, "created_at": 1, "_id": 0}
            ).sort("created_at", -1).limit(limit)
        )

        if not docs:
            return "No prior audits found for this organization.", 0

        lines = []
        for i, doc in enumerate(docs, 1):
            a = doc.get("audit", {})
            qm = a.get("quality_matrix", {})
            lines.append(
                f"Audit -{i}: F1={a.get('agent_f1_score', '?')}, "
                f"Risk={a.get('compliance_risk', '?')}, "
                f"Empathy={qm.get('cognitive_empathy', '?')}/10, "
                f"Efficiency={qm.get('efficiency', '?')}/10"
            )
        return "\n".join(lines), len(docs)

    except Exception as e:
        log.warning("Failed to fetch agent history: %s", str(e))
        return "Agent history unavailable.", 0


def _summarize_middle_chunk(text: str) -> str:
    """Internal helper to distill long transcripts using the fast T1 8B model."""
    if not text.strip():
        return ""
    # Removed direct Google Gemini summarization integration. Use a
    # lightweight deterministic fallback to avoid external SDK deps.
    try:
        return text[:2000] + "\n[... Middle Content Truncated ...]"
    except Exception:
        return ""

def _distill_transcript(transcript: str, limit: int = 22000) -> str:
    """Enterprise-grade 'Map-Reduce' distillation for large transcripts."""
    if len(transcript) <= limit:
        return transcript
        
    # Standard quality audit practice: Keep the Opening and Closing verbatim.
    head = transcript[:7000]
    tail = transcript[-7000:]
    middle = transcript[7000:-7000]
    
    log.info(f"📦 Large transcript detected ({len(transcript)} chars). Distilling middle...")
    
    distilled_middle = _summarize_middle_chunk(middle)
    
    return f"{head}\n\n[... DISTILLED CONTEXT ...]\n{distilled_middle}\n[... END DISTILLED CONTEXT ...]\n\n{tail}"


# ─────────────────────────────────────────────────────────────────────────────
# Prompt Builder
# ─────────────────────────────────────────────────────────────────────────────

def _build_prompt(transcript: str, org_id: str, acoustic_profile: dict = None) -> tuple[str, dict]:
    """Assemble the full contextual prompt including RAG and Acoustic signals."""
    # RAG Retrieval - Voyage AI indexed chunks via MongoDB
    rag_result = retrieve_policy_context(transcript, org_id, top_k=5)
    policy_context = rag_result.get("context", "").strip() or RAG_NULL_SENTINEL
    rag_provider = rag_result.get("provider", "none")
    chunk_count = len(rag_result.get("chunks", []))
    
    # Logic Fallback: if no RAG context exists, inject the Global Default Policy
    if policy_context == RAG_NULL_SENTINEL:
        policy_context = _DEFAULT_QUALITY_POLICY
        rag_provider = "default_policy_injection"

    # Historical calibration
    agent_history, history_count = _fetch_agent_history(org_id)
    
    # Acoustic signal grounding
    acoustic_context = _build_acoustic_context(acoustic_profile)

    # Enterprise Distillation (Map-Reduce) to stay within TPM/Context limits
    # instead of raw truncation, we preserve the Head/Tail + Distilled Middle.
    processed_transcript = _distill_transcript(transcript)

    user_prompt = _USER_PROMPT_TEMPLATE.format(
        policy_context=policy_context[:4000],
        history_count=history_count,
        agent_history=agent_history,
        transcript=processed_transcript,
        acoustic_context=acoustic_context
    )

    chunks = rag_result.get("chunks", [])
    policy_ids = list(set([c.get("doc_id") for c in chunks if c.get("doc_id")]))

    metadata = {
        "rag_provider": rag_provider,
        "policy_chunks": chunk_count,
        "history_audits": history_count,
        "policy_id": policy_ids
    }
    return user_prompt, metadata


# Generic retry-with-backoff helper used by LLM call sites.
def _retry_with_backoff(call_fn, attempts: int = 2, sleep_seconds: int = 2, retriable_check=None):
    """
    Execute `call_fn()` up to `attempts` times, sleeping `sleep_seconds`
    between retries when `retriable_check` (or default heuristics) mark
    the error as transient.

    - call_fn: zero-arg callable that performs a single attempt and returns
      a value or raises an Exception on failure.
    - retriable_check: optional callable (exc, err_str) -> bool indicating
      whether the error is retriable. By default retries on "429", "quota",
      or "limit" appearing in the exception string.

    Raises the last exception if all attempts are exhausted or the error is
    considered non-retriable.
    """
    last_err = None
    for attempt in range(attempts):
        try:
            return call_fn()
        except Exception as e:
            last_err = e
            err_str = str(e).lower()
            if retriable_check:
                try:
                    retriable = bool(retriable_check(e, err_str))
                except Exception:
                    retriable = False
            else:
                retriable = ("429" in err_str or "quota" in err_str or "limit" in err_str)

            if retriable and attempt < attempts - 1:
                log.warning("⏳ Transient error, retrying (attempt %d/%d): %s", attempt + 1, attempts, err_str)
                time.sleep(sleep_seconds)
                continue
            # Not retriable or no attempts left: re-raise to allow caller to
            # move to the next model/provider in the cascade.
            raise


# ─────────────────────────────────────────────────────────────────────────────
# LLM Cascade: Gemini → OpenRouter → Groq → HF Inference
# ─────────────────────────────────────────────────────────────────────────────




def _call_openrouter(user_prompt: str) -> tuple[str, str]:
    """OpenRouter (free-tier models). Tier 2.

    Tries multiple models (configurable via `OPENROUTER_MODELS` env var).
    If any model returns HTTP 429 we treat the provider as rate-limited and
    raise `LLMRateLimitError` so the cascade immediately moves to the next
    tier.
    """
    if not OPENROUTER_API_KEY:
        raise RuntimeError("OPENROUTER_API_KEY not set.")

    headers = {
        "Authorization": f"Bearer {OPENROUTER_API_KEY}",
        "Content-Type": "application/json",
        "X-Title": "Qualora Quality Auditor",
    }

    models_env = os.environ.get("OPENROUTER_MODELS", "")
    if models_env:
        models = [m.strip() for m in models_env.split(",") if m.strip()]
    else:
        # Default set: prioritized free models on OpenRouter. Users can
        # override via `OPENROUTER_MODELS` env var (comma-separated).
        models = [
            "openrouter/free",
            "qwen/qwen3.6-plus:free",
            "x-ai/grok-4.20",
        ]

    last_err = None
    for model in models:
        try:
            payload = {
                "model": model,
                "messages": [
                    {"role": "system", "content": _SYSTEM_PROMPT},
                    {"role": "user",   "content": user_prompt},
                ],
                "temperature": _LLM_TEMPERATURE,
                "response_format": {"type": "json_object"}
            }
            log.info("🔗 [OpenRouter] HTTPX POST %s", model)

            # Reduced timeout to 15s for better cascade flow
            with httpx.Client(base_url="https://openrouter.ai/api/v1", headers=headers, timeout=15.0) as client:
                r = client.post("/chat/completions", json=payload)
                if r.status_code == 429:
                    log.warning("⏳ [OpenRouter 429] Rate limited for model %s. Skipping provider.", model)
                    raise LLMRateLimitError(f"OpenRouter rate limited on model {model}")
                if r.status_code in (401, 403):
                    raise RuntimeError(f"OpenRouter Authentication failed: {r.text}")
                try:
                    r.raise_for_status()
                except Exception as he:
                    last_err = he
                    log.warning("⚠️ [OpenRouter %s] HTTP error: %s", model, he)
                    continue

            try:
                data = r.json()
            except Exception as je:
                last_err = je
                log.warning("⚠️ [OpenRouter %s] invalid JSON response: %s", model, je)
                continue

            if "choices" in data and len(data["choices"]) > 0:
                text = data["choices"][0]["message"]["content"].strip()
                log.info("✅ [LLM Tier 2] OpenRouter %s responded.", model)
                return text, f"openrouter/{model}"

            last_err = RuntimeError("OpenRouter returned no choices")
            log.warning("⚠️ [OpenRouter %s] No choices in response.", model)

        except LLMRateLimitError:
            # Bubble up so cascade will skip the entire provider tier
            raise
        except Exception as e:
            log.warning("⚠️  [OpenRouter %s] %s", model, str(e))
            last_err = e
            continue

    raise RuntimeError(f"All OpenRouter models failed. Last: {last_err}")

def _call_groq(user_prompt: str) -> tuple[str, str]:
    """Groq Llama-3.3-70B. Tier 3."""
    groq_client = get_groq_client()
    if not groq_client:
        raise RuntimeError("GROQ_API_KEY not set or Groq client unavailable.")
    models_env = os.environ.get("GROQ_MODELS", "")
    if models_env:
        models = [m.strip() for m in models_env.split(",") if m.strip()]
    else:
        models = [
            "llama-3.3-70b-versatile",
            "llama-3.1-8b-instant",
            "groq/compound",
            "groq/compound-mini",
        ]

    last_err = None
    for model in models:
        def fn():
            log.info("🚀 [Groq %s] Sending prompt", model)
            resp = groq_client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system", "content": _SYSTEM_PROMPT},
                    {"role": "user",   "content": user_prompt},
                ],
                temperature=_LLM_TEMPERATURE,
                max_tokens=2048,
                response_format={"type": "json_object"}
            )
            # Best-effort extraction of the textual content
            try:
                text = resp.choices[0].message.content.strip()
            except Exception:
                text = str(resp)
            if text:
                log.info("✅ [LLM Tier 3] Groq %s responded.", model)
                return text, f"groq/{model}"
            raise RuntimeError(f"Groq {model} empty response")

        try:
            return _retry_with_backoff(fn, attempts=2, sleep_seconds=2)
        except Exception as e:
            err_str = str(e).lower()
            if "429" in err_str or "rate limit" in err_str or "quota" in err_str:
                log.warning("⏳ [Groq %s] Rate limited: %s", model, e)
                raise LLMRateLimitError(f"Groq rate limited on model {model}: {e}")
            log.warning("⚠️  [Groq %s] %s", model, str(e))
            last_err = e

    raise RuntimeError(f"All Groq models failed. Last: {last_err}")


def _call_hf_inference(user_prompt: str) -> tuple[str, str]:
    """HuggingFace Inference API (Tier 4 — last resort).

    Uses the modern huggingface_hub InferenceClient which handles routing
    to the correct free serverless inference endpoints automatically.
    """
    if not HF_SPACE_TOKEN:
        raise RuntimeError("HF_SPACE_TOKEN missing — cannot use HF Tier 4 fallback")

    from huggingface_hub import InferenceClient

    # Cascade of small instruction-tuned models warm on HF free Serverless tier.
    _HF_MODELS = [
        "Qwen/Qwen2.5-7B-Instruct",
        "mistralai/Mistral-7B-Instruct-v0.3",
        "meta-llama/Llama-3.1-8B-Instruct",
    ]
    models_env = os.environ.get("HF_MODELS", "")
    if models_env:
        hf_models = [m.strip() for m in models_env.split(",") if m.strip()]
    else:
        hf_models = _HF_MODELS

    client = InferenceClient(api_key=HF_SPACE_TOKEN)
    messages = [
        {"role": "system", "content": _SYSTEM_PROMPT},
        {"role": "user",   "content": user_prompt},
    ]

    last_err = None
    for model_id in hf_models:
        def fn():
            log.info("\U0001f517 [HF T4] Trying %s", model_id)
            response = client.chat.completions.create(
                model=model_id,
                messages=messages,
                temperature=_LLM_TEMPERATURE,
                max_tokens=1024,
            )

            text = response.choices[0].message.content.strip()
            if text:
                log.info("\u2705 [LLM Tier 4] HF Inference %s responded.", model_id)
                return text, f"hf/{model_id}"

            raise RuntimeError(f"HF {model_id}: empty content in response")

        try:
            return _retry_with_backoff(fn, attempts=2, sleep_seconds=2)
        except Exception as e:
            err_str = str(e).lower()
            if "429" in err_str or "rate limit" in err_str or "quota" in err_str:
                log.warning("⏳ [HF %s] Rate limited: %s", model_id, e)
                raise LLMRateLimitError(f"HF rate limited on model {model_id}: {e}")
            log.warning("\u26a0\ufe0f  [HF T4] %s failed: %s", model_id, str(e))
            last_err = e

    raise RuntimeError(f"All HF Inference models failed. Last: {last_err}")

def _run_llm_cascade(user_prompt: str) -> tuple[str, str, str]:
    """
    Run LLM providers in priority order. 
    1. Gemini (Stable/HQ) -> 2. OpenRouter -> 3. Groq -> 4. HF
    """
    tiers = [
        ("T1", _call_openrouter, "openrouter"),
        ("T2", _call_groq, "groq"),
        ("T3", _call_hf_inference, "hf"),
    ]

    for tier_label, fn, prov_key in tiers:
        try:
            # Send the ENTIRE prompt.
            raw_text, provider_str = fn(user_prompt)
            # Validates early exit logic: returns native strings instead of double-serializing JSON
            return raw_text, provider_str, tier_label
        except LLMRateLimitError as le:
            # Provider reports rate limiting — skip to next tier immediately.
            log.warning("LLM cascade %s rate-limited: %s. Skipping to next tier.", tier_label, str(le))
            continue
        except Exception as e:
            log.warning("LLM cascade %s failed: %s. Trying next tier.", tier_label, str(e))

    raise RuntimeError("All LLM tiers exhausted — no provider responded successfully.")


# ─────────────────────────────────────────────────────────────────────────────
# Response Parser / Schema Enforcer
# ─────────────────────────────────────────────────────────────────────────────

_VALID_RISKS = {"Green", "Amber", "Red"}
_VALID_SATISFACTION = {"High", "Medium", "Low"}
_MATRIX_KEYS = ["language_proficiency", "cognitive_empathy", "efficiency", "bias_reduction", "active_listening"]


def _parse_llm_response(raw: str) -> dict:
    """
    Parse and validate LLM JSON output. Applies enforcement rules:
    - Coerce quality_matrix values to int [0–10]
    - Clamp agent_f1_score to [0.0–1.0]
    - Default unknown enum values
    - Ensure list fields are actually lists
    Uses centralized `repair_json` utility to recover gracefully from token truncation.
    """
    try:
        data = repair_json(raw)
    except ValueError as e:
        log.error(f"JSON Parse Error: {e} | Raw LLM Output: {raw[:200]}")
        raise ValueError("LLM returned malformed and unrecoverable JSON.")

    # Normalization: some LLMs may emit a top-level JSON array. Prefer the
    # first object element when available so downstream code can rely on a
    # mapping/dict interface (calls to `.get()` etc.). If the array contains
    # no objects, coerce into a safe dict using a joined summary.
    if isinstance(data, list):
        first_obj = None
        for item in data:
            if isinstance(item, dict):
                first_obj = item
                break
        if first_obj is not None:
            data = first_obj
        else:
            log.warning("LLM returned JSON array without object items — coercing to dict summary.")
            try:
                data = {"summary": " ".join([str(x) for x in data])}
            except Exception:
                data = {"summary": str(data)}

    # ── Enforce schema ─────────────────────────────────────────────────────
    if not isinstance(data.get("summary"), str) or not data.get("summary", "").strip():
        data["summary"] = _FALLBACK["summary"]

    f1 = data.get("agent_f1_score", 0.5)
    try:    
        f1 = float(f1)
    except (TypeError, ValueError):
        f1 = 0.5
    data["agent_f1_score"] = round(max(0.0, min(1.0, f1)), 3)

    if data.get("satisfaction_prediction") not in _VALID_SATISFACTION:
        data["satisfaction_prediction"] = "Medium"

    if data.get("compliance_risk") not in _VALID_RISKS:
        data["compliance_risk"] = "Amber"

    qm = data.get("quality_matrix", {})
    if not isinstance(qm, dict):
        qm = {}
    for key in _MATRIX_KEYS:
        try:
            val = int(round(float(qm.get(key, 5))))
        except (TypeError, ValueError):
            val = 5
        qm[key] = max(0, min(10, val))
    data["quality_matrix"] = qm

    for field in ("compliance_flags", "behavioral_nudges"):
        v = data.get(field, [])
        if not isinstance(v, list):
            v = [str(v)] if v else []
        data[field] = [str(item) for item in v][:10]

    emotions = data.get("emotions", {})
    if not isinstance(emotions, dict):
        emotions = {}
    agent_val = str(emotions.get("agent", "neutral"))
    customer_val = str(emotions.get("customer", "neutral"))
    data["emotions"] = {"agent": agent_val, "customer": customer_val}
    # Preserve any structured per-turn timeline provided by the LLM
    timeline = None
    if isinstance(emotions.get("timeline"), list):
        timeline = emotions.get("timeline")
    elif isinstance(data.get("emotional_timeline"), list):
        timeline = data.get("emotional_timeline")
    if timeline:
        data["emotions"]["timeline"] = timeline

    return data


def _coerce_audit_to_dict(audit_obj: any) -> dict:
    """Ensure an audit representation is a dict. If a list is supplied,
    prefer the first dict item; otherwise coerce into a safe summary dict.
    This guards against legacy cached values or DB-persisted arrays.
    """
    if isinstance(audit_obj, dict):
        return audit_obj
    if isinstance(audit_obj, list):
        for item in audit_obj:
            if isinstance(item, dict):
                return item
        try:
            return {"summary": " ".join([str(x) for x in audit_obj])}
        except Exception:
            return {"summary": str(audit_obj)}
    # Fallback: wrap non-dict into summary
    try:
        return {"summary": str(audit_obj)}
    except Exception:
        return {"summary": ""}


# ─────────────────────────────────────────────────────────────────────────────
# Public Entry Point
# ─────────────────────────────────────────────────────────────────────────────

def run_audit(transcript: str, org_id: str, acoustic_profile: dict = None) -> dict:
    """
    Main entry point for auditing.
    Cascades: 
    1. Deterministic Cache (SHA256)
    2. RAG Retrieval (Voyage AI + MongoDB)
    3. LLM Cascade (Gemini -> OpenRouter -> Groq -> HF)
    4. Defensive Merge (Schema recovery + Logic Fallback)
    """
    t0 = time.time()
    if not transcript or not transcript.strip():
        raise ValueError("Transcript cannot be empty")

    # 1. Deterministic Cache Check
    # Include KB fingerprint so cache invalidates automatically when policy
    # documents/chunks change for the org.
    policy_fp = _policy_cache_fingerprint(org_id)
    transcript_for_cache = _canonicalize_transcript_for_cache(transcript)
    cache_key = hashlib.sha256(f"{transcript_for_cache}_{org_id}_{policy_fp}".encode()).hexdigest()
    if cache_key in _AUDIT_CACHE:
        log.info(f"Audit Cache HIT for {cache_key}")
        cached = copy.deepcopy(_AUDIT_CACHE[cache_key])
        try:
            cached = _coerce_audit_to_dict(cached)
        except Exception:
            pass
        return copy.deepcopy(cached)
    # If not found in-memory, check persistent MongoDB cache (if configured)
    try:
        coll = _get_cache_collection()
        if coll is not None:
            try:
                doc = coll.find_one({"cache_key": cache_key})
                if doc and doc.get('audit'):
                    loaded = copy.deepcopy(doc.get('audit'))
                    try:
                        loaded = _coerce_audit_to_dict(loaded)
                    except Exception:
                        pass
                    _AUDIT_CACHE[cache_key] = copy.deepcopy(loaded)
                    log.info(f"Audit DB Cache HIT for {cache_key}")
                    return copy.deepcopy(loaded)
            except Exception as e:
                log.warning("DB cache lookup failed: %s", e)
    except Exception:
        # Avoid failing the entire audit due to cache subsystem
        log.debug("Persistent cache lookup skipped or failed.")

    # 2. Build Multi-modal Prompt
    user_prompt, ctx_meta = _build_prompt(transcript, org_id, acoustic_profile)

    # 3. LLM Inference Cascade
    raw_text, llm_provider, tier = _run_llm_cascade(user_prompt)

    # 4. Parsing & Defensive Logic
    audit = _parse_llm_response(raw_text)
    audit = _apply_defensive_merge(audit, transcript, acoustic_profile)

    # Guardrail: when policies are available and risk is Amber/Red, ensure at
    # least one compliance flag exists for downstream compliance workflows.
    try:
        risk = str(audit.get("compliance_risk") or "").title()
        flags = audit.get("compliance_flags")
        if not isinstance(flags, list):
            flags = [str(flags)] if flags else []

        if ctx_meta.get("policy_chunks", 0) > 0 and risk in {"Amber", "Red"} and not flags:
            flags = [
                "Policy adherence concern detected from retrieved knowledge-base context. Review transcript against cited policy clauses."
            ]

        audit["compliance_flags"] = [str(f).strip() for f in flags if str(f).strip()]
    except Exception as e:
        log.warning("Compliance-flag guardrail failed: %s", e)

    # Step 5: Inject observability metadata
    latency_ms = int((time.time() - t0) * 1000)
    audit["_audit_metadata"] = {
        "rag_provider":   ctx_meta.get("rag_provider", "none"),
        "llm_provider":   llm_provider,
        "tier":           tier,
        "policy_chunks":  ctx_meta.get("policy_chunks", 0),
        "policy_fingerprint": policy_fp,
        "history_audits": ctx_meta.get("history_audits", 0),
        "latency_ms":     latency_ms,
        "audited_at":     datetime.now(timezone.utc).isoformat(),
    }

    # 6. Store deterministic result in in-memory cache so identical
    #    (transcript, org_id) inputs can return the same audit without
    #    re-invoking the LLM cascade. Use a simple size-limited FIFO
    #    eviction strategy controlled by _MAX_AUDIT_CACHE.
    try:
        if cache_key:
            if len(_AUDIT_CACHE) >= _MAX_AUDIT_CACHE:
                try:
                    _AUDIT_CACHE.pop(next(iter(_AUDIT_CACHE)))
                except Exception:
                    _AUDIT_CACHE.clear()
            _AUDIT_CACHE[cache_key] = copy.deepcopy(audit)
            log.debug("Stored audit in cache for key %s", cache_key)
    except Exception as e:
        log.warning("Failed to store audit in cache: %s", e)

    # Persist cache entry to MongoDB (if configured). Use upsert so repeated
    # runs overwrite the cached audit for the same input key.
    try:
        coll = _get_cache_collection()
        if coll is not None:
            try:
                coll.update_one(
                    {"cache_key": cache_key},
                    {"$set": {
                        "cache_key": cache_key,
                        "transcript": transcript,
                        "org_id": org_id,
                        "audit": audit,
                        "created_at": datetime.now(timezone.utc)
                    }},
                    upsert=True
                )
            except Exception as e:
                log.warning("Failed to persist audit to DB cache: %s", e)
    except Exception:
        log.debug("Skipping persistent cache write (not configured or DB unavailable)")

    log.info(
        "Audit complete",
        extra={
            "tier": tier,
            "provider": llm_provider,
            "rag": ctx_meta.get("rag_provider"),
            "latency_ms": latency_ms,
            "f1": audit.get("agent_f1_score"),
            "risk": audit.get("compliance_risk"),
        }
    )
    return audit