src/api/interviews.py

"""Interview session lifecycle endpoints."""

from __future__ import annotations

import base64
import os
import uuid
import urllib.parse
import asyncio
from datetime import datetime, timezone

from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, Form, BackgroundTasks
from fastapi.responses import StreamingResponse

from langchain_core.messages import HumanMessage

from src.api.deps import get_cache, get_deepgram_key, get_graph, get_llm, get_sarvam_key, get_storage, require_auth
from src.api.auth import authenticate_websocket
from src.api.concurrency import STT_TIMEOUT_SEC, TTS_TIMEOUT_SEC, run_blocking
from src.api.limits import (
    MAX_AUDIO_BYTES,
    MAX_EDITOR_CHARS,
    MIN_AUDIO_BYTES,
    enforce_max_bytes,
)
from src.api.models import (
    InterviewStartRequest,
    InterviewStartResponse,
    InterviewPrepareResponse,
    MessageRequest,
    MessageResponse,
    SessionEndResponse,
    SessionStateResponse,
)
from src.cache import BodhiCache
from src.services.llm import _extract_text
from src.storage import BodhiStorage

router = APIRouter(prefix="/api/interviews", tags=["interviews"])


def _assert_session_owner(storage: BodhiStorage, session_id: str, user_id: str) -> None:
    """Authorize that `user_id` owns `session_id`, else raise.

    The sessions table is the source of truth — clerk_user_id is recorded at
    create_session(). Non-owners get a 404 (not 403) so the endpoint does not
    reveal that a session exists.
    """
    info = storage.get_session_info(session_id)
    if not info or info.get("clerk_user_id") != user_id:
        raise HTTPException(404, f"Session '{session_id}' not found")


def _resolve_owned_profile_id(
    storage: BodhiStorage, body_user_id: str | None, user_id: str
) -> str | None:
    """Resolve the candidate profile id for this request, enforcing ownership.

    If the client supplies a profile id (body.user_id), verify it belongs to the
    authenticated user before using it — otherwise a user could run an interview
    against someone else's resume. When no id is supplied, derive it from the
    caller's Clerk identity. Profiles with no recorded owner (local/dev) are
    allowed through so the anonymous dev flow keeps working.
    """
    if body_user_id:
        if not storage:
            return body_user_id
        profile = storage.get_user_profile(body_user_id)
        if not profile:
            raise HTTPException(404, "Profile not found")
        owner = profile.get("clerk_user_id")
        if owner not in (None, "", user_id):
            raise HTTPException(403, "You do not have access to this profile")
        return body_user_id
    return storage.get_user_profile_id_by_clerk_user_id(user_id) if storage else None


def _load_entity_context(company: str, role: str, cache, storage) -> str:
    """Load company context + role profile, merging RAG and role data."""
    ctx_parts: list[str] = []

    # --- Role profile (independent of company) ---
    if storage and role:
        try:
            role_profile = storage.get_role(role)
            if role_profile:
                if role_profile.get("focus_areas"):
                    ctx_parts.append(f"Role focus areas: {role_profile['focus_areas']}")
                if role_profile.get("typical_topics"):
                    ctx_parts.append(f"Typical interview topics: {role_profile['typical_topics']}")
                if role_profile.get("description"):
                    ctx_parts.append(f"Role description: {role_profile['description']}")
        except Exception:
            pass

    # --- Company / RAG context ---
    if company:
        rag_ctx = ""
        if cache:
            cached = cache.get_rag_context(company, role)
            if cached:
                rag_ctx = cached

        if not rag_ctx and storage:
            try:
                from src.rag import retrieve_context
                rag_ctx = retrieve_context(company, role, storage) or ""
                if rag_ctx and cache:
                    cache.set_rag_context(company, role, rag_ctx)
            except Exception:
                pass

        if not rag_ctx and storage:
            entity = storage.get_entity(company)
            if entity:
                rag_ctx = (
                    f"{entity.get('description', '')} "
                    f"Hiring: {entity.get('hiring_patterns', '')} "
                    f"Tech: {entity.get('tech_stack', '')}"
                ).strip()
                if rag_ctx and cache:
                    cache.set_rag_context(company, role, rag_ctx)

        if rag_ctx:
            ctx_parts.append(rag_ctx)

    return "\n".join(ctx_parts)


def _load_candidate_context(
    mode: str,
    user_id: str | None,
    jd_text: str | None,
    storage,
    llm,
) -> tuple[dict, str, dict]:
    """Return (candidate_profile, jd_context, gap_map) for resume-based modes.

    For standard mode returns empty defaults. Raises HTTPException on missing inputs.
    """
    # normalise frontend aliases
    if mode == "mode_a":
        mode = "option_a"
    elif mode == "mode_b":
        mode = "option_b"

    if mode == "standard":
        return {}, "", {}

    if not user_id:
        raise HTTPException(400, f"user_id is required for mode '{mode}'")

    row = storage.get_user_profile(user_id)
    if not row:
        raise HTTPException(404, f"No profile found for user_id '{user_id}'")

    profile = row["professional_summary"]

    if mode == "option_a":
        return profile, "", {}

    # option_b — needs JD
    if not jd_text or not jd_text.strip():
        raise HTTPException(400, "jd_text is required for mode 'option_b'")

    gap_map: dict = {}
    if llm:
        try:
            from src.resume_parser import build_gap_map
            gap_map = build_gap_map(profile, jd_text, llm)
        except Exception:
            gap_map = {}

    return profile, jd_text, gap_map


def _load_suggested_topics(company: str, role: str, cache) -> str:
    if not cache:
        return ""
    topics = cache.get_topics(company, role)
    if not topics:
        return ""
    return "\n".join(f"  - {t}" for t in topics)


def _seniority_to_difficulty(candidate_profile: dict, explicit_level: str = "") -> int:
    """Derive an initial interview difficulty from explicit level or parsed resume profile."""
    level = (explicit_level or "").lower()
    if "fresher" in level or "intern" in level or "junior" in level:
        return 2
    if "senior" in level or "2+" in level:
        return 4
    if "mid" in level or "1-2" in level:
        return 3

    seniority = (candidate_profile.get("seniority_level") or "").lower()
    years = candidate_profile.get("years_of_experience") or 0
    try:
        years = float(years)
    except (TypeError, ValueError):
        years = 0

    level_map = {
        "intern": 2, "junior": 2,
        "mid": 3,
        "senior": 4,
        "staff": 5, "principal": 5, "executive": 5,
    }
    if seniority in level_map:
        return level_map[seniority]
    if years < 2: return 2
    if years < 7: return 3
    if years < 10: return 4
    return 5


_CURRICULUM_PROMPT = """\
You are an expert technical interviewer preparing a custom interview curriculum.
Generate exactly 2 targeted questions for each of the following 2 phases for a {role} at {company}.
The candidate's experience level is: {experience_level}. This is a HARD constraint, not a hint:
- intern / fresher / junior: fundamentals and understanding only. Do NOT ask deep optimization,
  production-scale, failure-mode, or architecture-design questions — even if the JD lists advanced
  tooling. Ask what things are and why they're used, not how to tune them at scale.
- mid: hands-on usage, real trade-offs, everyday debugging.
- senior / staff / lead: system design, scale, optimization, and architectural judgement.
Use the provided company profile and job description to make the questions specific and realistic,
but never exceed the difficulty appropriate for the stated experience level.

COMPANY PROFILE:
{profile_text}

{jd_block}

OUTPUT FORMAT:
Return a valid JSON object with EXACTLY two keys: "technical" and "dsa".
"technical": 2 domain-relevant technical questions (e.g. language internals, framework concepts, system design).
"dsa": 2 data structures & algorithms questions (e.g. array/tree/graph problems with clear input/output).
Each key must contain a list of exactly 2 question strings.
DO NOT include any markdown blocks (like ```json), just raw JSON.
"""

def generate_interview_curriculum(
    company: str,
    role: str,
    experience_level: str,
    storage: BodhiStorage,
    jd_text: str = "",
    candidate_profile: dict | None = None,
    gap_map: dict | None = None,
) -> dict:
    """Generate 2 technical + 2 DSA pre-decided questions based on company profile and JD.

    When jd_text names concrete technologies (Docker, CI/CD, AWS ELB, ...), those are
    extracted and matched against a reusable topic_questions cache (research once, reuse
    across every candidate who hits that topic) instead of asking the LLM to invent
    generic technical questions from scratch every time. gap_map (resume vs JD) — if
    available — picks the question depth per topic: verify claimed strengths deeply,
    go easy on topics the candidate never claimed.
    """
    from src.services.llm import create_llm, _extract_text
    from langchain_core.messages import HumanMessage
    import json
    import logging
    log = logging.getLogger("bodhi.curriculum")
    
    profile_parts = []
    try:
        if storage:
            entity = storage.get_entity(company)
            if entity:
                profile_parts.append(f"Company Description: {entity.get('description', '')}")
                profile_parts.append(f"Company Tech Stack: {entity.get('tech_stack', '')}")
                profile_parts.append(f"Company Hiring Patterns: {entity.get('hiring_patterns', '')}")
                
            profiles = storage.get_company_profiles(company)
            for p in profiles:
                if p.get("role", "").lower() == role.lower():
                    profile_parts.append(f"Role Specific Description: {p.get('description', '')}")
                    profile_parts.append(f"Role Tech Stack: {p.get('tech_stack', '')}")
                    profile_parts.append(f"Role Hiring Patterns: {p.get('hiring_patterns', '')}")
                    break
    except Exception as e:
        log.error(f"Failed to fetch profile from DB: {e}")
        
    profile_text = "\n".join(filter(None, profile_parts))
    if not profile_text.strip():
        profile_text = "No specific company data available. Generate standard questions."
    
    jd_block = ""
    if jd_text and jd_text.strip():
        jd_block = f"JOB DESCRIPTION (provided by candidate):\n{jd_text[:8000]}"
        log.info(f"[CURRICULUM] JD text provided ({len(jd_text)} chars)")
    
    llm = create_llm(api_key=os.getenv("GOOGLE_API_KEY", ""))
    prompt = _CURRICULUM_PROMPT.format(
        role=role, company=company, experience_level=experience_level, profile_text=profile_text, jd_block=jd_block
    )
    
    try:
        response = llm.invoke([HumanMessage(content=prompt)])
        raw = _extract_text(response.content).strip()
        import re
        match = re.search(r'\{.*\}', raw, re.DOTALL)
        if match:
            raw = match.group(0)
        
        try:
            data = json.loads(raw)
        except json.JSONDecodeError as e:
            log.error(f"JSON decode failed: {e}. Raw: {raw}")
            data = {"technical": [], "dsa": []}
        
        result = {
            "technical": data.get("technical", [])[:2],
            "dsa": data.get("dsa", [])[:2],
        }
        
        # DEBUG output
        log.info("==================================================")
        log.info(f"PRE-GENERATED CURRICULUM FOR {company} | {role}")
        log.info(f"  Technical ({len(result['technical'])} Qs): {result['technical']}")
        log.info(f"  DSA ({len(result['dsa'])} Qs): {result['dsa']}")
        if jd_text:
            log.info(f"  JD context: YES ({len(jd_text)} chars)")
        log.info("==================================================")

    except Exception as e:
        log.error(f"Curriculum generation failed: {e}")
        result = {"technical": [], "dsa": []}

    if jd_text and jd_text.strip():
        try:
            from src.rag import extract_jd_topics, get_topic_questions, tier_for_topic
            topics = extract_jd_topics(jd_text)
            jd_questions: list[str] = []
            for topic in topics:
                tier = tier_for_topic(
                    topic, gap_map,
                    experience_level=experience_level,
                    candidate_profile=candidate_profile,
                )
                jd_questions.extend(get_topic_questions(topic, storage, tier=tier, limit=2))
            if jd_questions:
                # JD-specific topics take priority over the generic technical Qs above.
                result["technical"] = (jd_questions + result.get("technical", []))[:6]
                log.info(f"  JD topics: {topics} -> {len(jd_questions)} cached/generated questions")
        except Exception as e:
            log.error(f"JD topic question lookup failed: {e}")

    return result


@router.post("/prepare", response_model=InterviewPrepareResponse, status_code=201)
async def prepare_interview(
    body: InterviewStartRequest,
    user_id: str = Depends(require_auth),
    storage: BodhiStorage = Depends(get_storage),
    cache: BodhiCache | None = Depends(get_cache),
    llm=Depends(get_llm),
):
    """Sync prepare: loads context, generates curriculum, and creates session_id."""
    session_id = uuid.uuid4().hex[:12]

    resolved_user_profile_id = _resolve_owned_profile_id(storage, body.user_id, user_id)

    candidate_profile, jd_context, gap_map = _load_candidate_context(
        body.mode, resolved_user_profile_id, body.jd_text, storage, llm
    )
    entity_context = _load_entity_context(body.company, body.role, cache, storage)
    suggested_topics = _load_suggested_topics(body.company, body.role, cache)
    
    experience_level_used = body.experience_level
    if body.mode != "standard" and resolved_user_profile_id and storage:
        db_exp = storage.get_user_experience_level(user_id)
        if db_exp:
            experience_level_used = db_exp

    # Pre-generate curriculum (2 technical + 2 DSA questions) unless in resume-based mode
    curriculum = {}
    if body.mode != "option_a":
        curriculum = generate_interview_curriculum(
            body.company, body.role, experience_level_used, storage, jd_text=body.jd_text,
            candidate_profile=candidate_profile, gap_map=gap_map,
        )
        if cache:
            for phase, questions in curriculum.items():
                cache.set_question_queue(session_id, phase, questions)

    try:
        storage.create_session(
            session_id,
            body.candidate_name,
            body.company,
            body.role,
            clerk_user_id=user_id,
            user_profile_id=resolved_user_profile_id,
        )
    except Exception:
        pass

    # Determine initial difficulty from candidate seniority
    difficulty_level = _seniority_to_difficulty(candidate_profile, explicit_level=experience_level_used) if candidate_profile or experience_level_used else 3

    initial_state_data = {
        "session_id": session_id,
        "candidate_name": body.candidate_name,
        "target_company": body.company,
        "target_role": body.role,
        "current_phase": "intro",
        "difficulty_level": difficulty_level,
        "phase_scores": {},
        "entity_context": entity_context,
        "suggested_topics": suggested_topics,
        "should_end": False,
        "interviewer_persona": body.interviewer_persona,
        "queued_questions": curriculum,
        "target_question": "",
        "interview_mode": body.mode,
        "candidate_profile": candidate_profile,
        "jd_context": jd_context,
        "gap_map": gap_map,
        "clerk_user_id": user_id,
        "user_profile_id": resolved_user_profile_id,
        "quick_demo": body.quick_demo,
    }
    
    if cache:
        cache.save_initial_state(session_id, initial_state_data)
        # Verify the save actually persisted
        verify = cache.get_initial_state(session_id)
        if not verify:
            raise HTTPException(503, "Failed to persist session state to cache. Check Redis connection.")
    else:
        raise HTTPException(503, "Cache unavailable — cannot prepare interview session.")

    return InterviewPrepareResponse(session_id=session_id)


@router.post("", response_model=InterviewStartResponse, status_code=201)
async def start_interview(
    body: InterviewStartRequest,
    user_id: str = Depends(require_auth),
    graph=Depends(get_graph),
    storage: BodhiStorage = Depends(get_storage),
    cache: BodhiCache | None = Depends(get_cache),
    sarvam_key: str = Depends(get_sarvam_key),
    llm=Depends(get_llm),
):
    session_id = uuid.uuid4().hex[:12]

    resolved_user_profile_id = _resolve_owned_profile_id(storage, body.user_id, user_id)

    candidate_profile, jd_context, gap_map = _load_candidate_context(
        body.mode, resolved_user_profile_id, body.jd_text, storage, llm
    )
    entity_context = _load_entity_context(body.company, body.role, cache, storage)
    suggested_topics = _load_suggested_topics(body.company, body.role, cache)
    
    experience_level_used = body.experience_level
    if body.mode != "standard" and resolved_user_profile_id and storage:
        db_exp = storage.get_user_experience_level(user_id)
        if db_exp:
            experience_level_used = db_exp

    # Pre-generate curriculum (2 technical + 2 DSA questions) unless in resume-based mode
    curriculum = {}
    if body.mode != "option_a":
        curriculum = generate_interview_curriculum(
            body.company, body.role, experience_level_used, storage, jd_text=body.jd_text,
            candidate_profile=candidate_profile, gap_map=gap_map,
        )
        if cache:
            for phase, questions in curriculum.items():
                cache.set_question_queue(session_id, phase, questions)

    try:
        storage.create_session(
            session_id,
            body.candidate_name,
            body.company,
            body.role,
            clerk_user_id=user_id,
            user_profile_id=resolved_user_profile_id,
        )
    except Exception:
        pass

    # Determine initial difficulty from candidate seniority
    difficulty_level = _seniority_to_difficulty(candidate_profile, explicit_level=experience_level_used) if candidate_profile or experience_level_used else 3

    graph_config = {"configurable": {"thread_id": session_id}}
    initial_state = {
        "messages": [HumanMessage(content="Hello, I'm ready for my interview.")],
        "session_id": session_id,
        "candidate_name": body.candidate_name,
        "target_company": body.company,
        "target_role": body.role,
        "current_phase": "intro",
        "difficulty_level": difficulty_level,
        "phase_scores": {},
        "entity_context": entity_context,
        "suggested_topics": suggested_topics,
        "should_end": False,
        "interviewer_persona": body.interviewer_persona,
        "queued_questions": curriculum,
        "target_question": "",  # intro is ad-hoc, no target question
        "interview_mode": body.mode,
        "candidate_profile": candidate_profile,
        "jd_context": jd_context,
        "gap_map": gap_map,
    }

    result = await asyncio.to_thread(graph.invoke, initial_state, graph_config)
    greeting = _extract_text(
        result["messages"][-1].content
        if result["messages"] and hasattr(result["messages"][-1], "content")
        else ""
    )

    audio_b64 = ""
    if sarvam_key and greeting:
        try:
            from src.services.tts import text_to_speech_bytes
            audio_bytes = await run_blocking(
                text_to_speech_bytes,
                greeting, api_key=sarvam_key, target_language_code="hi-IN", speaker="shubh",
                timeout=TTS_TIMEOUT_SEC, label="Speech synthesis",
            )
            audio_b64 = base64.b64encode(audio_bytes).decode()
        except Exception:
            pass

    return InterviewStartResponse(
        session_id=session_id,
        greeting_text=greeting,
        greeting_audio_b64=audio_b64,
    )


@router.post("/{session_id}/message", response_model=MessageResponse)
async def send_message(
    session_id: str,
    body: MessageRequest,
    user_id: str = Depends(require_auth),
    graph=Depends(get_graph),
    storage: BodhiStorage = Depends(get_storage),
    cache: BodhiCache | None = Depends(get_cache),
    sarvam_key: str = Depends(get_sarvam_key),
):
    _assert_session_owner(storage, session_id, user_id)
    graph_config = {"configurable": {"thread_id": session_id}}

    try:
        state = graph.get_state(graph_config)
        if not state or not state.values:
            raise HTTPException(404, f"Session '{session_id}' not found")
    except HTTPException:
        raise
    except Exception:
        raise HTTPException(404, f"Session '{session_id}' not found")

    result = await asyncio.to_thread(
        graph.invoke,
        {"messages": [HumanMessage(content=body.text)]},
        graph_config,
    )

    reply = ""
    if result["messages"] and hasattr(result["messages"][-1], "content"):
        reply = _extract_text(result["messages"][-1].content).strip()

    phase = result.get("current_phase", "unknown")
    should_end = result.get("should_end", False)

    audio_b64 = ""
    if sarvam_key and reply:
        try:
            from src.services.tts import text_to_speech_bytes
            audio_bytes = await run_blocking(
                text_to_speech_bytes,
                reply, api_key=sarvam_key, target_language_code="hi-IN", speaker="shubh",
                timeout=TTS_TIMEOUT_SEC, label="Speech synthesis",
            )
            audio_b64 = base64.b64encode(audio_bytes).decode()
        except Exception:
            pass

    if cache:
        try:
            cache.save_session_state(session_id, {
                "phase": phase,
                "difficulty": result.get("difficulty_level", 3),
                "scores": result.get("phase_scores", {}),
            })
        except Exception:
            pass

    if should_end:
        _flush_session_async(session_id, result, graph_config)

    return MessageResponse(
        transcript=body.text,
        reply_text=reply,
        reply_audio_b64=audio_b64,
        phase=phase,
        should_end=should_end,
    )


@router.post("/{session_id}/audio", response_model=MessageResponse)
async def send_audio(
    session_id: str,
    file: UploadFile = File(...),
    user_id: str = Depends(require_auth),
    graph=Depends(get_graph),
    storage: BodhiStorage = Depends(get_storage),
    cache: BodhiCache | None = Depends(get_cache),
    sarvam_key: str = Depends(get_sarvam_key),
):
    """Upload WAV audio, transcribe via STT, then process through interview graph."""
    _assert_session_owner(storage, session_id, user_id)
    graph_config = {"configurable": {"thread_id": session_id}}

    try:
        state = graph.get_state(graph_config)
        if not state or not state.values:
            raise HTTPException(404, f"Session '{session_id}' not found")
    except HTTPException:
        raise
    except Exception:
        raise HTTPException(404, f"Session '{session_id}' not found")

    audio_bytes = await file.read()
    enforce_max_bytes(audio_bytes, MAX_AUDIO_BYTES, "Audio")
    if not audio_bytes or len(audio_bytes) < MIN_AUDIO_BYTES:
        raise HTTPException(400, "Audio file too small or empty")

    if not sarvam_key:
        raise HTTPException(500, "SARVAM_API_KEY not configured")

    from src.services.stt import transcribe_audio
    transcript = await run_blocking(
        transcribe_audio,
        audio_bytes, api_key=sarvam_key, model="saaras:v3", language_code="en-IN",
        timeout=STT_TIMEOUT_SEC, label="Transcription",
    )

    transcript = (transcript or "").strip()
    if not transcript:
        raise HTTPException(422, "Could not transcribe audio")

    result = await asyncio.to_thread(
        graph.invoke,
        {"messages": [HumanMessage(content=transcript)]},
        graph_config,
    )

    reply = ""
    if result["messages"] and hasattr(result["messages"][-1], "content"):
        reply = _extract_text(result["messages"][-1].content).strip()

    phase = result.get("current_phase", "unknown")
    should_end = result.get("should_end", False)

    audio_b64 = ""
    if sarvam_key and reply:
        try:
            from src.services.tts import text_to_speech_bytes
            audio_bytes_out = await run_blocking(
                text_to_speech_bytes,
                reply, api_key=sarvam_key, target_language_code="hi-IN", speaker="shubh",
                timeout=TTS_TIMEOUT_SEC, label="Speech synthesis",
            )
            audio_b64 = base64.b64encode(audio_bytes_out).decode()
        except Exception:
            pass

    if cache:
        try:
            cache.save_session_state(session_id, {
                "phase": phase,
                "difficulty": result.get("difficulty_level", 3),
                "scores": result.get("phase_scores", {}),
            })
        except Exception:
            pass

    if should_end:
        _flush_session_async(session_id, result, graph_config)

    return MessageResponse(
        transcript=transcript,
        reply_text=reply,
        reply_audio_b64=audio_b64,
        phase=phase,
        should_end=should_end,
    )


@router.get("/{session_id}", response_model=SessionStateResponse)
async def get_session(
    session_id: str,
    user_id: str = Depends(require_auth),
    graph=Depends(get_graph),
    storage: BodhiStorage = Depends(get_storage),
):
    _assert_session_owner(storage, session_id, user_id)
    graph_config = {"configurable": {"thread_id": session_id}}
    try:
        state = graph.get_state(graph_config)
        if not state or not state.values:
            raise HTTPException(404, f"Session '{session_id}' not found")
    except HTTPException:
        raise
    except Exception:
        raise HTTPException(404, f"Session '{session_id}' not found")

    vals = state.values
    return SessionStateResponse(
        session_id=vals.get("session_id", session_id),
        phase=vals.get("current_phase", "unknown"),
        difficulty_level=vals.get("difficulty_level", 3),
        phase_scores=vals.get("phase_scores", {}),
        company=vals.get("target_company", ""),
        role=vals.get("target_role", ""),
    )


@router.post("/{session_id}/end", response_model=SessionEndResponse)
async def end_interview(
    session_id: str,
    background_tasks: BackgroundTasks,
    user_id: str = Depends(require_auth),
    graph=Depends(get_graph),
    storage: BodhiStorage = Depends(get_storage),
    cache: BodhiCache | None = Depends(get_cache),
):
    _assert_session_owner(storage, session_id, user_id)
    graph_config = {"configurable": {"thread_id": session_id}}
    try:
        state = graph.get_state(graph_config)
        if not state or not state.values:
            raise HTTPException(404, f"Session '{session_id}' not found")
    except HTTPException:
        raise
    except Exception:
        raise HTTPException(404, f"Session '{session_id}' not found")

    vals = state.values
    # Schedule the synchronous flushing (and report generation) to run in the background
    background_tasks.add_task(_flush_session_sync, session_id, vals, storage, cache)

    return SessionEndResponse(
        session_id=session_id,
        summary="Report generation in progress...",
        overall_score=None,
    )


def _flush_session_async(session_id: str, result: dict, graph_config: dict):
    """Best-effort session flush (non-blocking in the response path)."""
    pass


def _flush_session_sync(
    session_id: str,
    state: dict,
    storage: BodhiStorage,
    cache: BodhiCache | None,
) -> tuple[str, float | None]:
    """Flush session data to NeonDB, trigger RAG contribution, clean up Redis.
    Returns (summary, overall_score)."""
    from src.report import generate_report
    
    transcript_text = ""
    summary = ""
    overall_score: float | None = None
    report_data = None

    try:
        messages = []
        for msg in state.get("messages", []):
            role = "user" if isinstance(msg, HumanMessage) else "assistant"
            content = msg.content if hasattr(msg, "content") else str(msg)
            messages.append({"role": role, "content": _extract_text(content)})

        transcript_text = "\n".join(
            f"{m['role']}: {m['content']}" for m in messages
        )

        storage.save_transcript_batch(
            session_id, messages, state.get("current_phase", "unknown"),
        )

        scores = state.get("phase_scores", {})
        total_score = 0.0
        total_q = 0
        for phase, data in scores.items():
            q = data.get("questions", 0)
            s = data.get("total_score", 0)
            total_score += s
            total_q += q

        overall_score = total_score / total_q if total_q else None
        
        # Generate comprehensive report
        try:
            phase_memories = state.get("phase_memories", {})
            answer_scores = state.get("answer_scores", [])
            proctoring_violations = storage.get_proctoring_violations(session_id)
            sentiment_data = storage.get_sentiment_data(session_id)
            session_info = {
                "candidate_name": state.get("candidate_name", ""),
                "target_company": state.get("target_company", ""),
                "target_role": state.get("target_role", ""),
                "session_id": session_id,
            }
            
            # Look up custom metrics for this company+role
            company_custom_metrics: list[str] = []
            try:
                company_profiles = storage.get_company_profiles(state.get("target_company", ""))
                target_role_lower = state.get("target_role", "").lower()
                for cp in company_profiles:
                    if cp.get("role", "").lower() in (target_role_lower, "general"):
                        raw_cm = cp.get("custom_metrics") or []
                        if isinstance(raw_cm, str):
                            import json as _cjson
                            raw_cm = _cjson.loads(raw_cm)
                        if raw_cm:
                            company_custom_metrics = raw_cm
                            break
            except Exception:
                pass

            report_data = generate_report(
                phase_memories=phase_memories,
                answer_scores=answer_scores,
                phase_scores=scores,
                proctoring_violations=proctoring_violations,
                sentiment_data=sentiment_data,
                session_info=session_info,
                transcript_text=transcript_text,
                custom_metrics=company_custom_metrics,
            )
            
            summary = report_data.get("hiring_recommendation", f"Interview complete. {total_q} questions across {len(scores)} phases.")
        except Exception as e:
            _stream_log.warning(f"Failed to generate report: {e}")
            summary = f"Interview complete. {total_q} questions across {len(scores)} phases."
        
        storage.end_session(session_id, overall_score=overall_score, summary=summary, report_data=report_data)
    except Exception:
        pass

    if transcript_text:
        try:
            from src.rag import extract_and_contribute
            company = state.get("target_company", "")
            role = state.get("target_role", "")
            extract_and_contribute(company, role, transcript_text, storage)
        except Exception:
            pass

    if cache:
        try:
            cache.delete_session(session_id)
        except Exception:
            pass

    return summary, overall_score


# ── Streaming endpoints ───────────────────────────────────────────

import json as _json
import re
import asyncio
import logging
from typing import Optional
from fastapi import WebSocket, WebSocketDisconnect

from src.services.sentiment import analyze_tone as _analyze_tone
from src.services.stt import transcribe_audio

_stream_log = logging.getLogger("bodhi.api.stream")

@router.websocket("/{session_id}/ws")
async def interview_websocket(
    websocket: WebSocket,
    session_id: str,
    graph=Depends(get_graph),
    storage: BodhiStorage = Depends(get_storage),
    cache: BodhiCache | None = Depends(get_cache),
    sarvam_key: str = Depends(get_sarvam_key),
    deepgram_key: str = Depends(get_deepgram_key),
):
    """Live interview voice pipeline.

    Mechanics (ported from the standalone streaming engine):
      - Client streams raw PCM-16 (16 kHz mono) continuously.
      - Deepgram Nova-3 transcribes with server-side endpointing; its
        utterance_end event triggers the LLM → TTS turn (no client-side
        silence detection, no batch-WAV upload, no blocking STT round-trip).
      - The LangGraph "brain" is unchanged: tokens come from
        graph.astream_events() and are spoken via a persistent linear16
        Sarvam TTS WebSocket, streamed to the client as raw PCM.
      - Barge-in: an `interrupt` control cancels the in-flight turn and
        drops pending TTS audio.

    NOTE: per-turn audio sentiment/behavioral analysis is deferred here and
    wired back later; `reply_complete.sentiment` is `{}` for now.
    """
    from src.services.tts import SarvamTTSStream, split_sentences
    from src.services.stt_deepgram import DeepgramStreamingSTT

    # Authenticate the handshake before accepting the socket.
    ws_user_id = await authenticate_websocket(websocket)
    if ws_user_id is None:
        await websocket.close(code=1008, reason="Authentication required")
        return

    await websocket.accept()

    tts_sample_rate = getattr(websocket.app.state, "tts_sample_rate", 22050)

    session_tts: "SarvamTTSStream | None" = None
    session_stt: "DeepgramStreamingSTT | None" = None
    pipeline_task: asyncio.Task | None = None

    # Set on the first partial of a turn, used to measure speaking duration.
    turn_speech_start: float | None = None

    # Latest code-editor content the client has pushed (debounced). Injected into
    # the turn during technical/DSA phases so the interviewer can see the code.
    latest_editor_content: str = ""

    # "Relax" line state, triggered when proctoring reports a high violation rate.
    last_reassure_at = 0.0
    reassure_idx = 0
    REASSURE_COOLDOWN_SEC = 60.0
    REASSURE_LINES = [
        "Hey, no rush at all — take a breath and relax. You're doing great.",
        "Just take your time and stay relaxed. There's no pressure here.",
        "Take a moment if you need it. Stay calm — you've got this.",
    ]

    try:
        initial_state = None
        if cache:
            initial_state = cache.get_initial_state(session_id)
            # Retry once after a short delay (race condition safety)
            if not initial_state:
                await asyncio.sleep(0.5)
                initial_state = cache.get_initial_state(session_id)

        if not initial_state:
            _stream_log.error(f"WS error: No setup found for session {session_id} (cache={'present' if cache else 'None'})")
            await websocket.close(code=1008, reason="Session not prepared")
            return

        # Authorize: the authenticated user must own this session.
        # Sessions without a recorded owner (local/dev) are left accessible.
        session_owner = initial_state.get("clerk_user_id")
        if session_owner not in (None, "", ws_user_id):
            _stream_log.warning(
                f"WS auth: user {ws_user_id} attempted to access session {session_id} "
                f"owned by {session_owner}"
            )
            await websocket.close(code=1008, reason="Not authorized for this session")
            return

        graph_config = {"configurable": {"thread_id": session_id}}

        # Load messages back (we didn't store HumanMessage in Redis)
        initial_state["messages"] = [HumanMessage(content="Hello, I'm ready for my interview.")]

        # ── Per-session services ─────────────────────────────────────────
        # Created BEFORE the greeting LLM call so the TTS WebSocket handshake can
        # be pre-warmed concurrently with generation, instead of stacking serially
        # after it — this removes the Sarvam handshake RTT from first-audio latency.
        persona = initial_state.get("interviewer_persona", "bodhi")
        voice = "shreya" if persona == "riya" else "shubh"
        session_tts = SarvamTTSStream(api_key=sarvam_key, voice=voice, sample_rate=tts_sample_rate)
        session_stt = DeepgramStreamingSTT(api_key=deepgram_key)

        # Pre-warm the Sarvam TTS connection in parallel with greeting generation.
        tts_prewarm = asyncio.create_task(session_tts.connect()) if sarvam_key else None

        # First graph invocation (Greeting) — blocking full-text generation.
        result = await asyncio.to_thread(graph.invoke, initial_state, graph_config)
        greeting = ""
        if result["messages"] and hasattr(result["messages"][-1], "content"):
            greeting = _extract_text(result["messages"][-1].content).strip()

        # The handshake should be done by now; await it to surface any error (so
        # the inline connect() below retries cleanly) and avoid a dangling task.
        if tts_prewarm is not None:
            try:
                await tts_prewarm
            except Exception as e:
                _stream_log.warning(f"TTS pre-warm connect failed: {e!r} — retrying inline")

        if cache:
            cache.save_session_state(session_id, {
                "phase": result.get("current_phase", initial_state.get("current_phase")),
                "difficulty": result.get("difficulty_level", 3),
                "scores": result.get("phase_scores", {}),
            })

        # Tell the client the audio sample rate so it can build PCM buffers.
        await websocket.send_json({
            "type": "control",
            "event": "session_config",
            "sample_rate": tts_sample_rate,
        })

        # ── Greeting (streamed via persistent linear16 TTS) ──────────────
        await websocket.send_json({
            "type": "control",
            "event": "greeting_start",
            "text": greeting,
            "phase": result.get("current_phase", "intro"),
        })
        if sarvam_key and greeting:
            async def _greeting_sentences():
                for s in split_sentences(greeting):
                    yield s
            try:
                await session_tts.connect()
                async for chunk in session_tts.stream_tts(_greeting_sentences()):
                    await websocket.send_bytes(chunk)
            except Exception as e:
                _stream_log.exception(f"WS TTS greeting error: {type(e).__name__}: {e!r}")
        await websocket.send_json({
            "type": "control",
            "event": "greeting_complete",
            "phase": result.get("current_phase", "intro"),
        })

        # ── One user turn: transcript → graph tokens → TTS audio ─────────
        async def run_turn(transcript: str, speech_duration: float = 0.0) -> None:
            behavioral: dict = {}
            try:
                from src.services.behavioral import compute_speech_behavioral
                behavioral = compute_speech_behavioral(transcript, speech_duration)
                if storage:
                    await asyncio.to_thread(
                        storage.save_sentiment_data,
                        session_id,
                        sentiment=behavioral.get("sentiment"),
                        confidence_score=behavioral.get("confidence_score"),
                        speaking_rate_wpm=behavioral.get("speaking_rate_wpm"),
                        filler_rate=behavioral.get("filler_rate"),
                        flags=behavioral.get("flags") or None,
                    )
            except Exception as exc:
                _stream_log.warning("behavioral metrics failed: %s", exc)

            async def on_token(token: str):
                try:
                    await websocket.send_json({"type": "control", "event": "text_chunk", "text": token})
                except Exception:
                    pass

            # Append the code-editor content for technical/DSA phases so the LLM
            # actually reviews what the candidate wrote (mirrors the REST /audio path).
            user_input = transcript
            if latest_editor_content.strip():
                try:
                    state = graph.get_state(graph_config)
                    current_phase = state.values.get("current_phase", "") if state and state.values else ""
                    if current_phase in ("technical", "dsa"):
                        code = latest_editor_content.strip()[:MAX_EDITOR_CHARS]
                        user_input = f"{transcript}\n\n[Code Editor Content]:\n```\n{code}\n```"
                        _stream_log.info(
                            "[WS] Including editor content (%d chars) for %s phase",
                            len(code), current_phase,
                        )
                except Exception as e:
                    _stream_log.warning("Failed to attach editor content: %s", e)

            result_holder: dict = {}
            try:
                async for chunk in _session_pipeline_audio(
                    graph, graph_config, user_input, session_tts, result_holder, token_callback=on_token
                ):
                    if chunk:
                        await websocket.send_bytes(chunk)
            except asyncio.CancelledError:
                # Barge-in: turn was interrupted — drop it silently.
                raise
            except Exception as exc:
                # graph.invoke() (Gemini call) or the TTS stream blew past its own
                # retries. Previously this exception died silently in the
                # background task — the socket stayed open but nothing further
                # was ever sent, so the session just went dead mid-interview.
                # Always surface *something* so the client can recover instead
                # of waiting forever for a reply that will never come.
                _stream_log.error("[WS-PIPE] Turn failed: %s", exc, exc_info=True)
                try:
                    await websocket.send_json({
                        "type": "control",
                        "event": "turn_error",
                        "text": "Sorry, I didn't quite catch that — could you say it again?",
                    })
                except Exception:
                    pass
                return

            phase = result_holder.get("phase", "unknown")
            should_end = result_holder.get("should_end", False)

            if cache:
                try:
                    state_dict = graph.get_state(graph_config).values
                    cache.save_session_state(session_id, {
                        "phase": phase,
                        "difficulty": state_dict.get("difficulty_level", 3),
                        "scores": state_dict.get("phase_scores", {}),
                    })
                except Exception:
                    pass

            await websocket.send_json({
                "type": "control",
                "event": "reply_complete",
                "text": result_holder.get("reply_text"),
                "phase": phase,
                "should_end": should_end,
                "sentiment": behavioral,  # WPM / filler / confidence / tone
            })

            if should_end:
                try:
                    state_dict = graph.get_state(graph_config).values
                    # Generate + persist the report (blocking) before closing.
                    try:
                        await asyncio.to_thread(
                            _flush_session_sync, session_id, state_dict, storage, cache
                        )
                    except Exception as flush_exc:
                        _stream_log.error("Session flush failed: %s", flush_exc, exc_info=True)
                finally:
                    await websocket.close()

        async def _cancel_pipeline() -> None:
            nonlocal pipeline_task
            if pipeline_task and not pipeline_task.done():
                pipeline_task.cancel()
                try:
                    await pipeline_task
                except (asyncio.CancelledError, Exception):
                    pass
            pipeline_task = None

        # ── Deepgram callbacks ───────────────────────────────────────────
        async def on_interim(text: str) -> None:
            nonlocal turn_speech_start
            if turn_speech_start is None and text.strip():
                turn_speech_start = asyncio.get_running_loop().time()
            await websocket.send_json({"type": "control", "event": "interim_transcript", "text": text})

        async def on_utterance_end(transcript: str) -> None:
            nonlocal pipeline_task, turn_speech_start
            transcript = (transcript or "").strip()
            if not transcript:
                return
            # duration = first partial to now, minus Deepgram's trailing silence.
            speech_duration = 0.0
            if turn_speech_start is not None:
                elapsed = asyncio.get_running_loop().time() - turn_speech_start
                speech_duration = max(0.0, elapsed - session_stt.utterance_end_ms / 1000.0)
            turn_speech_start = None
            # Safety: cancel any still-running prior turn.
            await _cancel_pipeline()
            await websocket.send_json({"type": "control", "event": "transcript", "text": transcript})
            pipeline_task = asyncio.create_task(run_turn(transcript, speech_duration))

        await session_stt.connect(
            on_interim=on_interim,
            on_utterance_end=on_utterance_end,
        )

        async def speak_reassurance() -> None:
            """Speak a short 'relax' line when proctoring reports a high violation
            rate. Rate-limited by REASSURE_COOLDOWN_SEC and skipped mid-turn so it
            never talks over the candidate or an active question."""
            nonlocal last_reassure_at, reassure_idx
            now = asyncio.get_running_loop().time()
            if now - last_reassure_at < REASSURE_COOLDOWN_SEC:
                return
            if pipeline_task and not pipeline_task.done():
                return  # don't interrupt an in-flight turn
            last_reassure_at = now
            line = REASSURE_LINES[reassure_idx % len(REASSURE_LINES)]
            reassure_idx += 1
            await websocket.send_json(
                {"type": "control", "event": "reassurance", "text": line}
            )

            async def _one_line():
                yield line

            try:
                await session_tts.connect()
                async for chunk in session_tts.stream_tts(_one_line()):
                    await websocket.send_bytes(chunk)
            except Exception as e:
                _stream_log.error(f"reassurance TTS error: {type(e).__name__}: {e!r}")

        # ── Main receive loop ────────────────────────────────────────────
        while True:
            message = await websocket.receive()

            if message["type"] == "websocket.disconnect":
                break

            if "bytes" in message:
                # Continuous PCM-16 frames → forward straight to Deepgram.
                await session_stt.send_audio(message["bytes"])

            elif "text" in message:
                data = _json.loads(message["text"])
                msg_type = data.get("type", "")

                if msg_type == "speech.start":
                    session_stt.reset_transcript()

                elif msg_type == "interrupt":
                    # Barge-in: stop the current turn and drop pending audio.
                    await _cancel_pipeline()
                    await session_tts.close()
                    await session_tts.connect()
                    session_stt.reset_transcript()
                    await websocket.send_json({"type": "control", "event": "interrupted"})

                elif msg_type == "editor_update":
                    # Client pushed the latest code-editor content (debounced).
                    # Cached here; injected into the next turn (technical/DSA only).
                    latest_editor_content = str(data.get("content", ""))[:MAX_EDITOR_CHARS]

                elif msg_type == "proctor_alert":
                    # Browser CV saw a high violation rate; Bodhi reassures (cooldown).
                    await speak_reassurance()

                elif msg_type == "ping":
                    await websocket.send_json({"type": "control", "event": "pong"})

    except WebSocketDisconnect:
        _stream_log.info(f"WebSocket disconnected for {session_id}")
    except Exception as e:
        _stream_log.error(f"WebSocket unexpected error: {e}", exc_info=True)
    finally:
        if pipeline_task and not pipeline_task.done():
            pipeline_task.cancel()
            try:
                await pipeline_task
            except (asyncio.CancelledError, Exception):
                pass
        if session_stt is not None:
            await session_stt.close()
        if session_tts is not None:
            await session_tts.close()
        try:
            await websocket.close()
        except Exception:
            pass

_stream_log = logging.getLogger("bodhi.api.stream")


async def _tts_stream_generator(text: str, sarvam_key: str, speaker: str = "shubh"):
    """Yield MP3 audio chunks from TTS streaming (legacy full-text mode)."""
    _stream_log.info("Starting TTS stream for %d chars of text", len(text))
    from src.services.tts import text_to_speech_stream
    chunk_count = 0
    try:
        async for chunk in text_to_speech_stream(
            text, api_key=sarvam_key, target_language_code="hi-IN", speaker=speaker,
        ):
            chunk_count += 1
            _stream_log.debug("Yielding chunk #%d (%d bytes)", chunk_count, len(chunk))
            yield chunk
    except Exception as exc:
        _stream_log.error("TTS stream generator error: %s: %s", type(exc).__name__, exc, exc_info=True)
        raise
    _stream_log.info("TTS stream generator done: %d chunks yielded", chunk_count)


_SENTENCE_END = re.compile(r'(?<=[.!?])\s')


async def _sentence_accumulator(token_aiter):
    """Consume an async iterator of LLM tokens and yield complete sentences.

    Splits on sentence boundaries (.!?) so TTS gets coherent phrases.
    Flushes any remaining buffer at the end.
    """
    buf = ""
    async for token in token_aiter:
        buf += token
        # Check for sentence boundaries
        parts = _SENTENCE_END.split(buf)
        if len(parts) > 1:
            # All but last part are complete sentences
            for sentence in parts[:-1]:
                sentence = sentence.strip()
                if "[END_INTERVIEW]" in sentence:
                    sentence = sentence.replace("[END_INTERVIEW]", "").strip()
                if sentence:
                    _stream_log.info("[ACCUMULATOR] Yielding sentence: %s", sentence[:80])
                    yield sentence
            buf = parts[-1]
    # Flush remainder
    buf = buf.strip()
    if "[END_INTERVIEW]" in buf:
        buf = buf.replace("[END_INTERVIEW]", "").strip()
    if buf:
        _stream_log.info("[ACCUMULATOR] Yielding final fragment: %s", buf[:80])
        yield buf


async def _llm_tts_pipeline(graph, graph_config, user_input, sarvam_key: str, speaker: str = "shubh", token_callback=None):
    """Pipeline: LLM tokens → sentence accumulator → TTS audio chunks.

    Uses graph.astream_events() to get individual LLM tokens, accumulates
    them into sentences, feeds sentences to TTS concurrently, and
    fires token_callback for real-time text streaming.

    Yields:
        (audio_chunk: bytes | None, meta: dict | None)
    """
    from src.services.tts import tts_stream_sentences

    collected_text = []
    phase = "unknown"
    should_end = False

    # Async generator that extracts LLM tokens from astream_events
    async def _llm_tokens():
        nonlocal phase, should_end
        # See _session_pipeline_audio for why this buffers per-invocation:
        # the interviewer node re-runs after every tool call within one turn,
        # and only the final (no-tool-call) invocation is the real answer.
        pending_text: list[str] = []
        try:
            async for event in graph.astream_events(
                {"messages": [HumanMessage(content=user_input)]},
                config=graph_config,
                version="v2",
            ):
                kind = event.get("event", "")
                node_name = event.get("metadata", {}).get("langgraph_node", "")

                if kind == "on_chat_model_stream":
                    if node_name in ("interviewer", ""):
                        chunk = event.get("data", {}).get("chunk")
                        if chunk and hasattr(chunk, "content"):
                            token_text = _extract_text(chunk.content)
                            if token_text:
                                pending_text.append(token_text)

                elif kind == "on_chat_model_end":
                    if node_name in ("interviewer", ""):
                        output = event.get("data", {}).get("output")
                        tool_calls = getattr(output, "tool_calls", None) if output is not None else None
                        if not tool_calls:
                            # Use streamed tokens, or the full output if the model
                            # ran non-streaming (no on_chat_model_stream events).
                            texts = pending_text
                            if not texts and output is not None and hasattr(output, "content"):
                                full = _extract_text(output.content)
                                texts = [full] if full else []
                            for token_text in texts:
                                collected_text.append(token_text)
                                if token_callback:
                                    await token_callback(token_text)
                                yield token_text
                            if "[END_INTERVIEW]" in "".join(collected_text):
                                should_end = True
                        pending_text = []

                elif kind == "on_tool_end":
                    output = event.get("data", {}).get("output", "")
                    if hasattr(output, "content"):
                        output = output.content
                    output = str(output)
                    if output.startswith("TRANSITION:"):
                        phase = output.split(":", 1)[1]
                        _stream_log.info("[PIPELINE] Phase transition → %s", phase)
                    elif output.startswith("END:"):
                        should_end = True
                        _stream_log.info("[PIPELINE] Interview end triggered")

        except Exception as e:
            _stream_log.error("[PIPELINE] astream_events error: %s", e, exc_info=True)

    # Pipeline: LLM tokens → sentences → TTS audio
    sentence_stream = _sentence_accumulator(_llm_tokens())

    chunk_count = 0
    try:
        async for audio_chunk in tts_stream_sentences(
            sentence_stream,
            api_key=sarvam_key,
            target_language_code="hi-IN",
            speaker=speaker,
        ):
            chunk_count += 1
            yield audio_chunk, None
    except Exception as exc:
        _stream_log.error("[PIPELINE] TTS pipeline error: %s", exc, exc_info=True)

    # After all audio, get the current state for headers/cache
    try:
        state = graph.get_state(graph_config)
        if state and state.values:
            phase = state.values.get("current_phase", phase)
            should_end = state.values.get("should_end", should_end)
    except Exception:
        pass

    reply_text = "".join(collected_text).strip()
    if "[END_INTERVIEW]" in reply_text:
        should_end = True
        reply_text = reply_text.replace("[END_INTERVIEW]", "").strip()

    _stream_log.info("[PIPELINE] Done: %d audio chunks, %d chars reply, phase=%s, end=%s",
                     chunk_count, len(reply_text), phase, should_end)

    yield None, {"reply_text": reply_text, "phase": phase, "should_end": should_end}


async def _pipeline_audio_generator(graph, graph_config, user_input, sarvam_key, result_holder: dict, speaker: str = "shubh", token_callback=None):
    """Async generator that yields only audio bytes from the pipeline.
    Stores the final metadata in result_holder for the caller to inspect."""
    async for audio_chunk, meta in _llm_tts_pipeline(graph, graph_config, user_input, sarvam_key, speaker=speaker, token_callback=token_callback):
        if audio_chunk is not None:
            yield audio_chunk
        elif meta is not None:
            result_holder.update(meta)


async def _session_pipeline_audio(graph, graph_config, user_input, tts, result_holder: dict, token_callback=None):
    """Run one turn via graph.invoke and stream the reply to the persistent TTS WS.

    We use invoke (not astream_events) because the interviewer node is sync and
    runs in a threadpool, where the chat-model callbacks astream_events needs
    never fire, so the reply would never reach TTS. Gemini is non-streaming
    anyway, so there's no token stream to lose.
    """
    from src.services.tts import split_sentences

    result = await asyncio.to_thread(
        graph.invoke,
        {"messages": [HumanMessage(content=user_input)]},
        graph_config,
    )

    reply_text = ""
    msgs = result.get("messages") if isinstance(result, dict) else None
    if msgs and hasattr(msgs[-1], "content"):
        reply_text = _extract_text(msgs[-1].content).strip()

    phase = result.get("current_phase", "unknown") if isinstance(result, dict) else "unknown"
    should_end = bool(result.get("should_end", False)) if isinstance(result, dict) else False
    if "[END_INTERVIEW]" in reply_text:
        should_end = True
        reply_text = reply_text.replace("[END_INTERVIEW]", "").strip()

    _stream_log.info("[WS-PIPE] turn reply=%d chars phase=%s end=%s", len(reply_text), phase, should_end)

    # Surface the full reply text to the client (live transcript) up front.
    if token_callback and reply_text:
        try:
            await token_callback(reply_text)
        except Exception:
            pass

    # Stream the reply through the persistent linear16 TTS WS, sentence by sentence.
    if reply_text:
        async def _sentences():
            for s in split_sentences(reply_text):
                if s.strip():
                    yield s
        try:
            async for audio_chunk in tts.stream_tts(_sentences()):
                yield audio_chunk
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            _stream_log.error("[WS-PIPE] TTS pipeline error: %s", exc, exc_info=True)

    result_holder.update({"reply_text": reply_text, "phase": phase, "should_end": should_end})


def _stream_headers(**kwargs: str) -> dict[str, str]:
    """Build custom response headers for streaming endpoints.
    Values are URL-encoded to safely transport arbitrary text in HTTP headers."""
    headers = {}
    for key, val in kwargs.items():
        if val is not None:
            headers[f"X-Bodhi-{key}"] = urllib.parse.quote(str(val), safe="")
    return headers


@router.post("/start-stream")
async def start_interview_stream(
    body: InterviewStartRequest,
    user_id: str = Depends(require_auth),
    graph=Depends(get_graph),
    storage: BodhiStorage = Depends(get_storage),
    cache: BodhiCache | None = Depends(get_cache),
    sarvam_key: str = Depends(get_sarvam_key),
    llm=Depends(get_llm),
):
    """Start interview and stream greeting audio as MP3.
    Metadata is returned in response headers."""
    import json
    loop = asyncio.get_event_loop()
    session_id = uuid.uuid4().hex[:12]

    resolved_user_profile_id = _resolve_owned_profile_id(storage, body.user_id, user_id)

    _stream_log.info("[START-STREAM] Session %s: loading context...", session_id)

    # Run all blocking I/O in thread pool to avoid blocking event loop
    candidate_profile, jd_context, gap_map = await loop.run_in_executor(
        None, lambda: _load_candidate_context(body.mode, resolved_user_profile_id, body.jd_text, storage, llm)
    )
    entity_context = await loop.run_in_executor(
        None, lambda: _load_entity_context(body.company, body.role, cache, storage)
    )
    suggested_topics = await loop.run_in_executor(
        None, lambda: _load_suggested_topics(body.company, body.role, cache)
    )

    # Pre-generate curriculum (2 technical + 2 DSA questions)
    _stream_log.info("[START-STREAM] Session %s: generating curriculum...", session_id)
    curriculum = await loop.run_in_executor(
        None, lambda: generate_interview_curriculum(
            body.company, body.role, body.experience_level, storage, jd_text=body.jd_text,
            candidate_profile=candidate_profile, gap_map=gap_map,
        )
    )
    if cache:
        for phase, questions in curriculum.items():
            cache.set_question_queue(session_id, phase, questions)

    try:
        await loop.run_in_executor(
            None,
            lambda: storage.create_session(
                session_id,
                body.candidate_name,
                body.company,
                body.role,
                clerk_user_id=user_id,
                user_profile_id=resolved_user_profile_id,
            ),
        )
    except Exception:
        pass

    graph_config = {"configurable": {"thread_id": session_id}}
    initial_state = {
        "messages": [HumanMessage(content="Hello, I'm ready for my interview.")],
        "session_id": session_id,
        "candidate_name": body.candidate_name,
        "target_company": body.company,
        "target_role": body.role,
        "current_phase": "intro",
        "difficulty_level": 3,
        "interviewer_persona": body.interviewer_persona,
        "phase_scores": {},
        "entity_context": entity_context,
        "suggested_topics": suggested_topics,
        "should_end": False,
        "queued_questions": curriculum,
        "target_question": "",  # intro is ad-hoc, no target question
        "interview_mode": body.mode,
        "candidate_profile": candidate_profile,
        "jd_context": jd_context,
        "gap_map": gap_map,
        "quick_demo": body.quick_demo,
    }

    _stream_log.info("[START-STREAM] Session %s: invoking graph for greeting...", session_id)
    result = await loop.run_in_executor(
        None, lambda: graph.invoke(initial_state, config=graph_config)
    )
    greeting = _extract_text(
        result["messages"][-1].content
        if result["messages"] and hasattr(result["messages"][-1], "content")
        else ""
    )
    _stream_log.info("[START-STREAM] Session %s: greeting ready (%d chars)", session_id, len(greeting))

    if not sarvam_key or not greeting:
        raise HTTPException(500, "TTS not available")

    # Serialize curriculum for frontend debugging
    curriculum_json = json.dumps(curriculum) if curriculum else "{}"

    headers = _stream_headers(
        Session=session_id,
        Text=greeting,
        Phase="intro",
        End="false",
        Curriculum=curriculum_json,
    )

    speaker = "shreya" if body.interviewer_persona == "riya" else "shubh"
    return StreamingResponse(
        _tts_stream_generator(greeting, sarvam_key, speaker=speaker),
        media_type="audio/mpeg",
        headers=headers,
    )


@router.post("/{session_id}/message-stream")
async def send_message_stream(
    session_id: str,
    body: MessageRequest,
    user_id: str = Depends(require_auth),
    graph=Depends(get_graph),
    storage: BodhiStorage = Depends(get_storage),
    cache: BodhiCache | None = Depends(get_cache),
    sarvam_key: str = Depends(get_sarvam_key),
):
    """Send text message and stream reply audio as MP3 (low-latency pipeline)."""
    _assert_session_owner(storage, session_id, user_id)
    graph_config = {"configurable": {"thread_id": session_id}}

    try:
        state = graph.get_state(graph_config)
        if not state or not state.values:
            raise HTTPException(404, f"Session '{session_id}' not found")
    except HTTPException:
        raise
    except Exception:
        raise HTTPException(404, f"Session '{session_id}' not found")

    if not sarvam_key:
        raise HTTPException(500, "SARVAM_API_KEY not configured")

    result_holder: dict = {}
    persona = state.values.get("interviewer_persona", "bodhi")
    speaker = "shreya" if persona == "riya" else "shubh"

    async def _gen():
        async for chunk in _pipeline_audio_generator(
            graph, graph_config, body.text, sarvam_key, result_holder, speaker=speaker
        ):
            yield chunk
        # Post-stream: cache update
        if cache:
            try:
                st = graph.get_state(graph_config)
                if st and st.values:
                    cache.save_session_state(session_id, {
                        "phase": st.values.get("current_phase", "unknown"),
                        "difficulty": st.values.get("difficulty_level", 3),
                        "scores": st.values.get("phase_scores", {}),
                    })
            except Exception:
                pass
        if result_holder.get("should_end"):
            _flush_session_async(session_id, {}, graph_config)

    headers = _stream_headers(
        Transcript=body.text,
        Phase="streaming",
        End="false",
    )

    return StreamingResponse(
        _gen(),
        media_type="audio/mpeg",
        headers=headers,
    )


@router.post("/{session_id}/audio-stream")
async def send_audio_stream(
    session_id: str,
    file: UploadFile = File(...),
    image_file: Optional[UploadFile] = File(None),
    editor_content: Optional[str] = Form(None),
    user_id: str = Depends(require_auth),
    graph=Depends(get_graph),
    storage: BodhiStorage = Depends(get_storage),
    cache: BodhiCache | None = Depends(get_cache),
    sarvam_key: str = Depends(get_sarvam_key),
):
    """Upload WAV audio (+ optional webcam frame + optional editor content) and stream reply audio as MP3."""
    _assert_session_owner(storage, session_id, user_id)
    graph_config = {"configurable": {"thread_id": session_id}}

    try:
        state = graph.get_state(graph_config)
        if not state or not state.values:
            raise HTTPException(404, f"Session '{session_id}' not found")
    except HTTPException:
        raise
    except Exception:
        raise HTTPException(404, f"Session '{session_id}' not found")

    audio_bytes = await file.read()
    enforce_max_bytes(audio_bytes, MAX_AUDIO_BYTES, "Audio")
    if not audio_bytes or len(audio_bytes) < MIN_AUDIO_BYTES:
        raise HTTPException(400, "Audio file too small or empty")

    image_bytes = (await image_file.read()) if image_file else None

    if not sarvam_key:
        raise HTTPException(500, "SARVAM_API_KEY not configured")

    from src.services.stt import transcribe_audio
    transcript = await run_blocking(
        transcribe_audio,
        audio_bytes, api_key=sarvam_key, model="saaras:v3", language_code="en-IN",
        timeout=STT_TIMEOUT_SEC, label="Transcription",
    )

    transcript = (transcript or "").strip()
    if not transcript:
        raise HTTPException(422, "Could not transcribe audio")

    # ── Append editor content if provided (for technical/DSA phases) ──────────
    if editor_content and len(editor_content) > MAX_EDITOR_CHARS:
        raise HTTPException(413, f"Editor content too large (max {MAX_EDITOR_CHARS} characters)")
    user_input = transcript
    if editor_content and editor_content.strip():
        # Get current phase to determine if we should include editor context
        try:
            state = graph.get_state(graph_config)
            current_phase = state.values.get("current_phase", "") if state and state.values else ""
            
            # Only include editor content for technical and DSA phases
            if current_phase in ["technical", "dsa"]:
                user_input = (
                    f"{transcript}\n\n"
                    f"[Code Editor Content]:\n"
                    f"```\n{editor_content.strip()}\n```"
                )
                _stream_log.info(f"[AUDIO-STREAM] Including editor content ({len(editor_content)} chars) for {current_phase} phase")
        except Exception as e:
            _stream_log.warning(f"Failed to check phase for editor content: {e}")

    # ── Sentiment analysis ────────────────────────────────────────────────────
    loop = asyncio.get_running_loop()
    sentiment_payload: dict = {}

    try:
        # Rule-based (~1ms, always runs)
        rb = _analyze_tone(transcript, audio_bytes)
        sentiment_payload = rb.to_dict()

        # HuggingFace speech emotion in thread pool (~300ms)
        async def _hf_analysis():
            try:
                from src.behavioral_analysis.services.speech_service import analyze_speech
                return await loop.run_in_executor(
                    None, lambda: analyze_speech(audio_bytes, file.filename or "audio.wav")
                )
            except Exception as e:
                _stream_log.warning("HF speech analysis failed: %s", e)
                return {}

        # MediaPipe posture in thread pool (~100ms, only when frame sent)
        async def _posture_analysis():
            if not image_bytes:
                return {}
            try:
                from src.behavioral_analysis.services.posture_service import analyze_posture
                return await loop.run_in_executor(None, lambda: analyze_posture(image_bytes))
            except Exception as e:
                _stream_log.warning("Posture analysis failed: %s", e)
                return {}

        hf_result, posture_result = await asyncio.gather(_hf_analysis(), _posture_analysis())

        if hf_result:
            sentiment_payload.update({
                "hf_emotion":       hf_result.get("emotion"),
                "hf_confidence":    hf_result.get("emotion_confidence"),
                "sentiment":        hf_result.get("sentiment"),
                "pitch_variance":   hf_result.get("pitch_variance"),
                "confidence_score": hf_result.get("confidence_score"),
                "flags":            hf_result.get("flags", []),
            })
        if posture_result:
            sentiment_payload.update({
                "posture":         posture_result.get("posture"),
                "head_tilt_angle": posture_result.get("head_tilt_angle"),
                "gaze_direction":  posture_result.get("gaze_direction"),
                "spine_score":     posture_result.get("spine_score"),
                "face_visible":    posture_result.get("face_visible"),
                "posture_flags":   posture_result.get("flags", []),
            })
        
        # Save sentiment data to database
        try:
            storage.save_sentiment_data(
                session_id=session_id,
                emotion=sentiment_payload.get("hf_emotion") or sentiment_payload.get("emotion"),
                sentiment=sentiment_payload.get("sentiment"),
                confidence_score=sentiment_payload.get("confidence_score"),
                speaking_rate_wpm=sentiment_payload.get("speaking_rate_wpm"),
                filler_rate=sentiment_payload.get("filler_rate"),
                posture=sentiment_payload.get("posture"),
                gaze_direction=sentiment_payload.get("gaze_direction"),
                spine_score=sentiment_payload.get("spine_score"),
                flags=sentiment_payload.get("flags", []),
            )
        except Exception as e:
            _stream_log.warning(f"Failed to save sentiment data: {e}")
    except Exception as e:
        _stream_log.error("Sentiment block failed: %s", e)


    # ── Get LLM reply first so we can put it in headers ──────────────────────
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(
        None,
        lambda: graph.invoke(
            {"messages": [HumanMessage(content=user_input)]},
            config=graph_config,
        ),
    )

    reply_text = ""
    if result.get("messages") and hasattr(result["messages"][-1], "content"):
        reply_text = _extract_text(result["messages"][-1].content).strip()

    current_phase = result.get("current_phase", "streaming")
    should_end = result.get("should_end", False)

    # ── Phase transition retry: if reply is empty (tool-only turn), re-invoke ─
    # This happens when the LLM emits a TRANSITION/SCORE tool call but no spoken
    # text.  We send "[continue]" so it generates the next question/statement.
    if not reply_text and not should_end:
        _stream_log.info("[AUDIO-STREAM] Empty reply after graph.invoke — re-invoking with [continue]")
        result = await loop.run_in_executor(
            None,
            lambda: graph.invoke(
                {"messages": [HumanMessage(content="[continue]")]},
                config=graph_config,
            ),
        )
        if result.get("messages") and hasattr(result["messages"][-1], "content"):
            reply_text = _extract_text(result["messages"][-1].content).strip()
        current_phase = result.get("current_phase", current_phase)
        should_end = result.get("should_end", should_end)

    if cache:
        try:
            cache.save_session_state(session_id, {
                "phase": current_phase,
                "difficulty": result.get("difficulty_level", 3),
                "scores": result.get("phase_scores", {}),
            })
        except Exception:
            pass

    if should_end:
        _flush_session_async(session_id, result, graph_config)

    # ── Stream TTS for the reply ──────────────────────────────────────────────
    persona = state.values.get("interviewer_persona", "bodhi")
    speaker = "shreya" if persona == "riya" else "shubh"

    async def _gen():
        if reply_text:
            async for chunk in _tts_stream_generator(reply_text, sarvam_key, speaker=speaker):
                yield chunk

    headers = _stream_headers(
        Transcript=transcript,
        Text=reply_text,
        Phase=current_phase,
        End="true" if should_end else "false",
        Sentiment=_json.dumps(sentiment_payload),
    )

    return StreamingResponse(
        _gen(),
        media_type="audio/mpeg",
        headers=headers,
    )




@router.get("/{session_id}/report")
async def get_interview_report(
    session_id: str,
    user_id: str = Depends(require_auth),
    storage: BodhiStorage = Depends(get_storage),
):
    """Get the comprehensive interview report for a session."""
    _assert_session_owner(storage, session_id, user_id)
    try:
        report_data = storage.get_session_report_data(session_id)
        if not report_data:
            raise HTTPException(404, f"Report not found for session '{session_id}'")

        return report_data
    except HTTPException:
        raise
    except Exception:
        _stream_log.exception(f"Failed to retrieve report for session {session_id}")
        raise HTTPException(500, "Failed to retrieve report")


@router.get("/{session_id}/report/pdf")
async def download_interview_report_pdf(
    session_id: str,
    user_id: str = Depends(require_auth),
    storage: BodhiStorage = Depends(get_storage),
):
    """Generate and download the interview report as a PDF."""
    from fastapi.responses import StreamingResponse
    import io

    _assert_session_owner(storage, session_id, user_id)
    try:
        report_data = storage.get_session_report_data(session_id)
        if not report_data:
            raise HTTPException(404, f"Report not found for session '{session_id}'")

        # Generate PDF
        pdf_bytes = _generate_pdf_report(report_data)

        # Return as downloadable file
        return StreamingResponse(
            io.BytesIO(pdf_bytes),
            media_type="application/pdf",
            headers={
                "Content-Disposition": f"attachment; filename=interview_report_{session_id}.pdf"
            }
        )
    except HTTPException:
        raise
    except Exception:
        _stream_log.exception(f"Failed to generate PDF for session {session_id}")
        raise HTTPException(500, "Failed to generate PDF")


def _generate_pdf_report(report_data: dict) -> bytes:
    """Generate a PDF report from the report data using ReportLab."""
    from reportlab.lib.pagesizes import letter, A4
    from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
    from reportlab.lib.units import inch
    from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, PageBreak
    from reportlab.lib import colors
    from reportlab.lib.enums import TA_CENTER, TA_LEFT, TA_RIGHT
    import io
    
    buffer = io.BytesIO()
    doc = SimpleDocTemplate(buffer, pagesize=letter, topMargin=0.5*inch, bottomMargin=0.5*inch)
    story = []
    styles = getSampleStyleSheet()
    
    # Brand colors
    BRAND_PRIMARY = colors.HexColor('#37322F')
    BRAND_BG = colors.HexColor('#F7F5F3')
    BRAND_WARM_MID = colors.HexColor('#6B5E58')
    BRAND_TABLE_HEADER = colors.HexColor('#EDE9E4')
    BRAND_ALT_ROW = colors.HexColor('#F7F5F3')
    BRAND_GRID = colors.HexColor('#D9D3CC')

    # Custom styles
    title_style = ParagraphStyle(
        'CustomTitle',
        parent=styles['Heading1'],
        fontSize=24,
        textColor=BRAND_PRIMARY,
        spaceAfter=30,
        alignment=TA_CENTER,
    )

    heading_style = ParagraphStyle(
        'CustomHeading',
        parent=styles['Heading2'],
        fontSize=16,
        textColor=BRAND_PRIMARY,
        spaceAfter=12,
        spaceBefore=20,
    )

    subheading_style = ParagraphStyle(
        'CustomSubHeading',
        parent=styles['Heading3'],
        fontSize=13,
        textColor=BRAND_WARM_MID,
        spaceAfter=8,
        spaceBefore=12,
    )

    body_style = ParagraphStyle(
        'CustomBody',
        parent=styles['BodyText'],
        fontSize=10,
        textColor=BRAND_PRIMARY,
        spaceAfter=6,
    )

    # Brand header strip
    brand_data = [["BODHI", "AI Mock Interview Platform"]]
    brand_table = Table(brand_data, colWidths=[2*inch, 4.5*inch])
    brand_table.setStyle(TableStyle([
        ('BACKGROUND', (0, 0), (-1, -1), BRAND_PRIMARY),
        ('FONTNAME', (0, 0), (0, 0), 'Helvetica-Bold'),
        ('FONTNAME', (1, 0), (1, 0), 'Helvetica'),
        ('FONTSIZE', (0, 0), (0, 0), 14),
        ('FONTSIZE', (1, 0), (1, 0), 10),
        ('TEXTCOLOR', (0, 0), (-1, -1), colors.white),
        ('ALIGN', (0, 0), (0, 0), 'LEFT'),
        ('ALIGN', (1, 0), (1, 0), 'RIGHT'),
        ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'),
        ('TOPPADDING', (0, 0), (-1, -1), 10),
        ('BOTTOMPADDING', (0, 0), (-1, -1), 10),
        ('LEFTPADDING', (0, 0), (0, 0), 14),
        ('RIGHTPADDING', (1, 0), (1, 0), 14),
    ]))
    story.append(brand_table)
    story.append(Spacer(1, 0.2*inch))

    # Title
    story.append(Paragraph("Interview Performance Report", title_style))
    story.append(Spacer(1, 0.2*inch))
    
    # Session Info
    session_info = report_data.get("session_info", {})
    if session_info:
        info_data = [
            ["Candidate:", session_info.get("candidate_name", "N/A")],
            ["Company:", session_info.get("target_company", "N/A")],
            ["Role:", session_info.get("target_role", "N/A")],
            ["Session ID:", session_info.get("session_id", "N/A")],
        ]
        info_table = Table(info_data, colWidths=[1.5*inch, 4.5*inch])
        info_table.setStyle(TableStyle([
            ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'),
            ('FONTNAME', (1, 0), (1, -1), 'Helvetica'),
            ('FONTSIZE', (0, 0), (-1, -1), 10),
            ('TEXTCOLOR', (0, 0), (0, -1), BRAND_WARM_MID),
            ('TEXTCOLOR', (1, 0), (1, -1), BRAND_PRIMARY),
            ('VALIGN', (0, 0), (-1, -1), 'TOP'),
            ('BOTTOMPADDING', (0, 0), (-1, -1), 6),
        ]))
        story.append(info_table)
        story.append(Spacer(1, 0.3*inch))
    
    # Overall Score
    story.append(Paragraph("Overall Performance", heading_style))
    overall_grade = report_data.get("overall_grade", "N/A")
    overall_score = report_data.get("overall_score_pct", 0)
    
    grade_color = colors.green if overall_score >= 70 else colors.orange if overall_score >= 50 else colors.red
    
    score_data = [
        ["Grade", "Score", "Questions"],
        [overall_grade, f"{overall_score}%", str(report_data.get("total_questions", 0))],
    ]
    score_table = Table(score_data, colWidths=[2*inch, 2*inch, 2*inch])
    score_table.setStyle(TableStyle([
        ('BACKGROUND', (0, 0), (-1, 0), BRAND_TABLE_HEADER),
        ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
        ('FONTSIZE', (0, 0), (-1, 0), 11),
        ('TEXTCOLOR', (0, 0), (-1, 0), BRAND_PRIMARY),
        ('FONTNAME', (0, 1), (-1, 1), 'Helvetica-Bold'),
        ('FONTSIZE', (0, 1), (-1, 1), 14),
        ('TEXTCOLOR', (0, 1), (0, 1), grade_color),
        ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
        ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'),
        ('GRID', (0, 0), (-1, -1), 1, BRAND_GRID),
        ('ROWBACKGROUNDS', (0, 1), (-1, -1), [colors.white]),
        ('TOPPADDING', (0, 0), (-1, -1), 10),
        ('BOTTOMPADDING', (0, 0), (-1, -1), 10),
    ]))
    story.append(score_table)
    story.append(Spacer(1, 0.2*inch))
    
    # Hiring Recommendation
    recommendation = report_data.get("hiring_recommendation", "")
    if recommendation:
        story.append(Paragraph(f"<b>Recommendation:</b> {recommendation}", body_style))
        story.append(Spacer(1, 0.2*inch))
    
    # Phase Breakdown
    phase_breakdown = report_data.get("phase_breakdown", {})
    if phase_breakdown:
        story.append(Paragraph("Phase-wise Performance", heading_style))
        
        phase_data = [["Phase", "Grade", "Score", "Questions"]]
        for phase, data in phase_breakdown.items():
            phase_data.append([
                phase.capitalize(),
                data.get("grade", "N/A"),
                f"{data.get('score_pct', 0)}%",
                str(data.get("questions_asked", 0)),
            ])
        
        phase_table = Table(phase_data, colWidths=[1.5*inch, 1.5*inch, 1.5*inch, 1.5*inch])
        phase_table.setStyle(TableStyle([
            ('BACKGROUND', (0, 0), (-1, 0), BRAND_TABLE_HEADER),
            ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
            ('TEXTCOLOR', (0, 0), (-1, 0), BRAND_PRIMARY),
            ('FONTSIZE', (0, 0), (-1, -1), 10),
            ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
            ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'),
            ('GRID', (0, 0), (-1, -1), 1, BRAND_GRID),
            ('ROWBACKGROUNDS', (0, 1), (-1, -1), [colors.white, BRAND_ALT_ROW]),
            ('TOPPADDING', (0, 0), (-1, -1), 8),
            ('BOTTOMPADDING', (0, 0), (-1, -1), 8),
        ]))
        story.append(phase_table)
        story.append(Spacer(1, 0.2*inch))
    
    # Strengths and Improvements
    strengths = report_data.get("top_strengths", [])
    improvements = report_data.get("top_improvements", [])
    
    if strengths:
        story.append(Paragraph("Key Strengths", subheading_style))
        for strength in strengths:
            story.append(Paragraph(f"• {strength}", body_style))
        story.append(Spacer(1, 0.15*inch))
    
    if improvements:
        story.append(Paragraph("Areas for Improvement", subheading_style))
        for improvement in improvements:
            story.append(Paragraph(f"• {improvement}", body_style))
        story.append(Spacer(1, 0.2*inch))
    
    # Behavioral Analysis
    behavioral = report_data.get("behavioral_summary", {})
    if behavioral and behavioral.get("total_data_points", 0) > 0:
        story.append(Paragraph("Behavioral Analysis", heading_style))
        
        behavioral_data = [
            ["Metric", "Value"],
            ["Avg Confidence Score", f"{behavioral.get('avg_confidence_score', 0)}/100"],
            ["Avg Speaking Rate", f"{behavioral.get('avg_speaking_rate', 0)} wpm"],
            ["Avg Filler Rate", f"{behavioral.get('avg_filler_rate', 0)}%"],
            ["Dominant Emotion", behavioral.get("dominant_emotion", "N/A").capitalize()],
            ["Dominant Sentiment", behavioral.get("dominant_sentiment", "N/A").capitalize()],
            ["Posture Issues", str(behavioral.get("posture_issues", 0))],
            ["Gaze Issues", str(behavioral.get("gaze_issues", 0))],
        ]
        
        behavioral_table = Table(behavioral_data, colWidths=[3*inch, 3*inch])
        behavioral_table.setStyle(TableStyle([
            ('BACKGROUND', (0, 0), (-1, 0), BRAND_TABLE_HEADER),
            ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
            ('TEXTCOLOR', (0, 0), (-1, 0), BRAND_PRIMARY),
            ('FONTSIZE', (0, 0), (-1, -1), 10),
            ('ALIGN', (0, 0), (0, -1), 'LEFT'),
            ('ALIGN', (1, 0), (1, -1), 'RIGHT'),
            ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'),
            ('GRID', (0, 0), (-1, -1), 1, BRAND_GRID),
            ('ROWBACKGROUNDS', (0, 1), (-1, -1), [colors.white, BRAND_ALT_ROW]),
            ('TOPPADDING', (0, 0), (-1, -1), 8),
            ('BOTTOMPADDING', (0, 0), (-1, -1), 8),
        ]))
        story.append(behavioral_table)
        story.append(Spacer(1, 0.2*inch))
    
    # Proctoring Summary
    proctoring = report_data.get("proctoring_summary", {})
    if proctoring and proctoring.get("total_violations", 0) > 0:
        story.append(Paragraph("Proctoring Summary", heading_style))
        
        flagged_text = "Yes" if proctoring.get("session_flagged") else "No"
        flagged_color = colors.red if proctoring.get("session_flagged") else colors.green
        
        proctoring_data = [
            ["Metric", "Count"],
            ["Total Violations", str(proctoring.get("total_violations", 0))],
            ["High Severity", str(proctoring.get("high_severity_count", 0))],
            ["Medium Severity", str(proctoring.get("medium_severity_count", 0))],
            ["Low Severity", str(proctoring.get("low_severity_count", 0))],
            ["Session Flagged", flagged_text],
        ]
        
        proctoring_table = Table(proctoring_data, colWidths=[3*inch, 3*inch])
        proctoring_table.setStyle(TableStyle([
            ('BACKGROUND', (0, 0), (-1, 0), BRAND_TABLE_HEADER),
            ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
            ('TEXTCOLOR', (0, 0), (-1, 0), BRAND_PRIMARY),
            ('FONTSIZE', (0, 0), (-1, -1), 10),
            ('ALIGN', (0, 0), (0, -1), 'LEFT'),
            ('ALIGN', (1, 0), (1, -1), 'RIGHT'),
            ('VALIGN', (0, 0), (-1, -1), 'MIDDLE'),
            ('GRID', (0, 0), (-1, -1), 1, BRAND_GRID),
            ('ROWBACKGROUNDS', (0, 1), (-1, -1), [colors.white, BRAND_ALT_ROW]),
            ('TEXTCOLOR', (1, 5), (1, 5), flagged_color),
            ('FONTNAME', (1, 5), (1, 5), 'Helvetica-Bold'),
            ('TOPPADDING', (0, 0), (-1, -1), 8),
            ('BOTTOMPADDING', (0, 0), (-1, -1), 8),
        ]))
        story.append(proctoring_table)
        story.append(Spacer(1, 0.2*inch))
    
    # Cross-section Insights
    insights = report_data.get("cross_section_insights", [])
    if insights:
        story.append(Paragraph("Cross-section Insights", subheading_style))
        for insight in insights:
            story.append(Paragraph(f"• {insight}", body_style))
    
    # Build PDF
    doc.build(story)
    pdf_bytes = buffer.getvalue()
    buffer.close()
    
    return pdf_bytes



# ── Demo Mode Endpoints ───────────────────────────────────────────────────────

@router.post("/demo/{phase}/start-stream")
async def start_demo_interview_stream(
    phase: str,
    user_id: str = Depends(require_auth),
    graph=Depends(get_graph),
    storage: BodhiStorage = Depends(get_storage),
    cache: BodhiCache | None = Depends(get_cache),
    sarvam_key: str = Depends(get_sarvam_key),
):
    """Start a demo interview locked to a specific phase.
    
    Available phases: intro, technical, behavioral, dsa, project
    
    Uses GrowthX as the default company for context.
    """
    from src.state import PHASES, DEMO_PHASE_CONFIG
    import json
    
    # Validate phase
    valid_demo_phases = ["intro", "technical", "behavioral", "dsa", "project"]
    if phase not in valid_demo_phases:
        raise HTTPException(400, f"Invalid phase. Must be one of: {', '.join(valid_demo_phases)}")
    
    loop = asyncio.get_event_loop()
    session_id = f"demo-{phase}-{uuid.uuid4().hex[:8]}"
    
    # Use GrowthX as default company
    company = "GrowthX"
    role = "Software Engineer"
    candidate_name = "Demo User"
    
    _stream_log.info("[DEMO-START] Session %s: phase=%s", session_id, phase)
    
    # Load GrowthX context
    entity_context = await loop.run_in_executor(
        None, lambda: _load_entity_context(company, role, cache, storage)
    )
    suggested_topics = await loop.run_in_executor(
        None, lambda: _load_suggested_topics(company, role, cache)
    )
    
    # Generate curriculum for the specific phase only
    curriculum = {}
    if phase in ["technical", "dsa"]:
        full_curriculum = await loop.run_in_executor(
            None, lambda: generate_interview_curriculum(company, role, "Mid-Level", storage)
        )
        if phase in full_curriculum:
            curriculum[phase] = full_curriculum[phase]
    
    if cache and curriculum:
        for p, questions in curriculum.items():
            cache.set_question_queue(session_id, p, questions)
    
    # Create session in database
    try:
        await loop.run_in_executor(
            None,
            lambda: storage.create_session(
                session_id,
                candidate_name,
                company,
                role,
                clerk_user_id=user_id,
            ),
        )
    except Exception:
        pass
    
    # Build initial state with demo mode enabled
    graph_config = {"configurable": {"thread_id": session_id}}
    
    # Phase-specific greeting prompts
    phase_greetings = {
        "intro": "Hello! I'm ready to introduce myself.",
        "technical": "Hello! I'm ready for technical questions.",
        "behavioral": "Hello! I'm ready for behavioral questions.",
        "dsa": "Hello! I'm ready for coding and algorithm questions.",
        "project": "Hello! I'm ready to discuss my projects.",
    }
    
    initial_state = {
        "messages": [HumanMessage(content=phase_greetings.get(phase, "Hello!"))],
        "session_id": session_id,
        "candidate_name": candidate_name,
        "target_company": company,
        "target_role": role,
        "current_phase": phase,
        "difficulty_level": 3,
        "interviewer_persona": "bodhi",
        "phase_scores": {},
        "entity_context": entity_context,
        "suggested_topics": suggested_topics,
        "should_end": False,
        "queued_questions": curriculum,
        "target_question": "",
        "interview_mode": "standard",
        "candidate_profile": {},
        "jd_context": "",
        "gap_map": {},
        "demo_mode": True,
        "demo_phase": phase,
        "phase_question_count": 0,
        "phase_start_time": datetime.now(timezone.utc).isoformat(),
    }
    
    _stream_log.info("[DEMO-START] Session %s: invoking graph for greeting...", session_id)
    result = await loop.run_in_executor(
        None, lambda: graph.invoke(initial_state, config=graph_config)
    )
    greeting = _extract_text(
        result["messages"][-1].content
        if result["messages"] and hasattr(result["messages"][-1], "content")
        else ""
    )
    _stream_log.info("[DEMO-START] Session %s: greeting ready (%d chars)", session_id, len(greeting))
    
    if not sarvam_key or not greeting:
        raise HTTPException(500, "TTS not available")
    
    # Get phase config for frontend
    phase_config = DEMO_PHASE_CONFIG.get(phase, {})
    curriculum_json = json.dumps({
        "phase": phase,
        "max_questions": phase_config.get("max_questions", 3),
        "demo_mode": True,
    })
    
    headers = _stream_headers(
        Session=session_id,
        Text=greeting,
        Phase=phase,
        End="false",
        Curriculum=curriculum_json,
    )
    
    return StreamingResponse(
        _tts_stream_generator(greeting, sarvam_key, speaker="shubh"),
        media_type="audio/mpeg",
        headers=headers,
    )