"""Bot observability metrics (read-only) for the ERP admin dashboard page.

Reads the bot's own tables (bot_traces / bot_trace_events — owned and written
by the Bale bot container) with plain SELECTs only. Admin-gated. Calibration,
demo and test traffic (source != 'bale' / chat_id = 0) is excluded from the
aggregates; the traces endpoint can show everything with ?include_synthetic=1.

v2.1: per-MODEL and per-SECTION records (/models, /stages), message text
excerpts on /traces, and media serving (voice/attachment playback+download)
streamed from the bot's MinIO bucket with an on-disk fallback, via a
short-lived signed URL — <audio>/<a> tags cannot send Authorization headers.
"""
import hashlib
import hmac
import os
import re
import time
from pathlib import Path

from fastapi import APIRouter, Depends, Query, Request, Response
from fastapi.responses import FileResponse
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth import get_optional_user, require_role
from app.config import settings
from app.db.engine import get_db
from app.utils.errors import NotFoundError, UnauthorizedError, ValidationError

router = APIRouter(prefix="/bot-metrics", tags=["bot-metrics"])

# real-traffic predicate shared by every aggregate (same rule as the bot's
# tools/monitor.py so the numbers match the weekly digest)
REAL = ("e.trace_id IN (SELECT id FROM bot_traces"
        " WHERE coalesce(source,'bale') NOT IN ('calibration','demo')"
        " AND coalesce(chat_id,0) <> 0)")


async def _rows(db: AsyncSession, sql: str, params: dict | None = None) -> list[dict]:
    res = await db.execute(text(sql), params or {})
    return [dict(r._mapping) for r in res.all()]


@router.get("/summary")
async def summary(
    days: int = Query(7, ge=1, le=90),
    db: AsyncSession = Depends(get_db),
    _ctx=Depends(require_role("admin")),
):
    """Headline KPIs + daily volumes for the dashboard header cards."""
    p = {"days": days}
    kpis = (await _rows(db, f"""
        SELECT
          (SELECT count(*) FROM bot_traces
             WHERE coalesce(source,'bale') NOT IN ('calibration','demo')
               AND created_at >= now() - (:days * interval '1 day')) AS traces,
          (SELECT count(*) FROM bot_trace_events e WHERE {REAL}
             AND e.created_at >= now() - (:days * interval '1 day')) AS events,
          (SELECT count(*) FROM bot_trace_events e WHERE {REAL}
             AND (e.stage = 'jev.failure' OR (e.stage = 'jev.result' AND NOT e.ok)
                  OR e.stage LIKE 'matching.%.failure')
             AND e.created_at >= now() - (:days * interval '1 day')) AS errors,
          (SELECT coalesce(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY e.latency_ms)), 0)
             FROM bot_trace_events e WHERE {REAL} AND e.stage = 'jev.result'
             AND e.latency_ms IS NOT NULL
             AND e.created_at >= now() - (:days * interval '1 day')) AS jev_p95_ms,
          (SELECT count(*) FROM bot_trace_events e WHERE {REAL} AND e.stage = 'draft.commit'
             AND e.created_at >= now() - (:days * interval '1 day')) AS commits,
          (SELECT coalesce(sum((e.payload->'usage'->>'input_tokens')::bigint), 0)
             FROM bot_trace_events e WHERE {REAL}
             AND e.stage IN ('jev.result','matching.arbitration.result','matching.fallback.result')
             AND e.payload->'usage' IS NOT NULL
             AND e.created_at >= now() - (:days * interval '1 day')) AS tokens_in,
          (SELECT coalesce(sum((e.payload->'usage'->>'output_tokens')::bigint), 0)
             FROM bot_trace_events e WHERE {REAL}
             AND e.stage IN ('jev.result','matching.arbitration.result','matching.fallback.result')
             AND e.payload->'usage' IS NOT NULL
             AND e.created_at >= now() - (:days * interval '1 day')) AS tokens_out
    """, p))[0]
    daily = await _rows(db, f"""
        SELECT to_char(date_trunc('day', e.created_at), 'YYYY-MM-DD') AS day,
               count(*) AS events,
               count(DISTINCT e.trace_id) AS traces
        FROM bot_trace_events e WHERE {REAL}
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1 ORDER BY 1
    """, p)
    return {"data": {"kpis": kpis, "daily": daily, "days": days}}


@router.get("/routing")
async def routing(
    days: int = Query(7, ge=1, le=90),
    db: AsyncSession = Depends(get_db),
    _ctx=Depends(require_role("admin")),
):
    """Final mention statuses (arbiter + crisis fallback aggregates, plus the
    legacy per-mention rows) and the arbiter's actions on Jev's suggestions."""
    p = {"days": days}
    statuses = await _rows(db, f"""
        SELECT metric, sum(n)::int AS n FROM (
          SELECT unnest(ARRAY['exact','ambiguity','new']) AS metric, 0 AS n
          UNION ALL
          SELECT 'exact', sum(coalesce((e.payload->'statuses'->>'exact')::int,0))
            FROM bot_trace_events e WHERE {REAL}
            AND e.stage IN ('matching.arbitration.result','matching.fallback.result')
            AND e.created_at >= now() - (:days * interval '1 day')
          UNION ALL
          SELECT 'ambiguity', sum(coalesce((e.payload->'statuses'->>'ambiguity')::int,0))
            FROM bot_trace_events e WHERE {REAL}
            AND e.stage IN ('matching.arbitration.result','matching.fallback.result')
            AND e.created_at >= now() - (:days * interval '1 day')
          UNION ALL
          SELECT 'new', sum(coalesce((e.payload->'statuses'->>'new')::int,0))
            FROM bot_trace_events e WHERE {REAL}
            AND e.stage IN ('matching.arbitration.result','matching.fallback.result')
            AND e.created_at >= now() - (:days * interval '1 day')
          UNION ALL
          SELECT m->>'status', count(*) FROM bot_trace_events e
            JOIN bot_traces t ON t.id = e.trace_id, jsonb_array_elements(e.payload->'mentions') m
          WHERE e.stage='routing.decision' AND e.payload->>'source'='jev'
            AND coalesce(t.source,'bale') NOT IN ('calibration','demo')
            AND coalesce(t.chat_id,0) <> 0
            AND e.created_at >= now() - (:days * interval '1 day')
          GROUP BY 1
        ) x GROUP BY metric
    """, p)
    actions = await _rows(db, f"""
        SELECT
          coalesce(sum((e.payload->'actions'->>'endorsed_jev')::int),0)::int AS endorsed_jev,
          coalesce(sum((e.payload->'actions'->>'vetoed_to_new')::int),0)::int AS vetoed_to_new,
          coalesce(sum((e.payload->'actions'->>'demoted_to_ambiguity')::int),0)::int AS demoted_to_ambiguity,
          coalesce(sum((e.payload->'actions'->>'other')::int),0)::int AS other,
          count(*)::int AS invocations
        FROM bot_trace_events e WHERE {REAL} AND e.stage='matching.arbitration.result'
          AND e.created_at >= now() - (:days * interval '1 day')
    """, p)
    stages = await _rows(db, f"""
        SELECT e.stage, count(*)::int AS n
        FROM bot_trace_events e WHERE {REAL}
          AND e.stage IN ('jev.request','jev.result','jev.failure',
                          'matching.arbitration.result','matching.fallback.result')
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1 ORDER BY n DESC
    """, p)
    return {"data": {"statuses": statuses, "actions": actions[0] if actions else {}, "stages": stages}}


@router.get("/quality")
async def quality(
    days: int = Query(7, ge=1, le=90),
    db: AsyncSession = Depends(get_db),
    _ctx=Depends(require_role("admin")),
):
    """User choices as the real A/B proxy: what the user picked when asked."""
    p = {"days": days}
    amb = await _rows(db, f"""
        SELECT coalesce(e.payload->>'action','?') AS action,
               coalesce(e.payload->>'arg','-') AS arg,
               count(*)::int AS n
        FROM bot_trace_events e WHERE {REAL} AND e.stage='ambiguity.user_choice'
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1,2 ORDER BY n DESC
    """, p)
    picks = await _rows(db, f"""
        SELECT CASE WHEN (e.payload->>'idx')::int = 0 THEN 'top' ELSE 'other' END AS pick,
               count(*)::int AS n
        FROM bot_trace_events e WHERE {REAL} AND e.stage='user_choice.pick'
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1
    """, p)
    drafts = await _rows(db, f"""
        SELECT e.stage, count(*)::int AS n
        FROM bot_trace_events e WHERE {REAL}
          AND e.stage IN ('draft.render','draft.commit','draft.cancel','ambiguity.sent')
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1 ORDER BY n DESC
    """, p)
    return {"data": {"ambiguities": amb, "picks": picks, "drafts": drafts}}


@router.get("/jev")
async def jev_health(
    days: int = Query(7, ge=1, le=90),
    db: AsyncSession = Depends(get_db),
    _ctx=Depends(require_role("admin")),
):
    """Jev health: models seen (drift vs the pinned version), tokens, latency."""
    p = {"days": days}
    models = await _rows(db, f"""
        SELECT coalesce(e.model,'?') AS model, count(*)::int AS n,
               max(e.created_at)::text AS last_seen
        FROM bot_trace_events e WHERE {REAL} AND e.stage='jev.result'
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1 ORDER BY n DESC
    """, p)
    latency = await _rows(db, f"""
        SELECT e.stage,
               coalesce(round(percentile_cont(0.5) WITHIN GROUP (ORDER BY e.latency_ms)),0)::int AS p50,
               coalesce(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY e.latency_ms)),0)::int AS p95,
               count(*)::int AS n
        FROM bot_trace_events e WHERE {REAL}
          AND e.stage IN ('jev.result','nlu.result','matching.arbitration.result','matching.fallback.result')
          AND e.latency_ms IS NOT NULL
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1 ORDER BY n DESC
    """, p)
    tokens_daily = await _rows(db, f"""
        SELECT to_char(date_trunc('day', e.created_at), 'YYYY-MM-DD') AS day,
               coalesce(sum((e.payload->'usage'->>'input_tokens')::bigint),0)::bigint AS tokens_in,
               coalesce(sum((e.payload->'usage'->>'output_tokens')::bigint),0)::bigint AS tokens_out
        FROM bot_trace_events e WHERE {REAL}
          AND e.stage IN ('jev.result','matching.arbitration.result','matching.fallback.result')
          AND e.payload->'usage' IS NOT NULL
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1 ORDER BY 1
    """, p)
    return {"data": {"models": models, "latency": latency, "tokens_daily": tokens_daily}}


@router.get("/stt")
async def stt_health(
    days: int = Query(7, ge=1, le=90),
    db: AsyncSession = Depends(get_db),
    _ctx=Depends(require_role("admin")),
):
    """STT providers and the audio-enhancement layer (ok-rate + p50 latency)."""
    p = {"days": days}
    providers = await _rows(db, f"""
        SELECT coalesce(e.payload->>'provider','?') AS provider,
               count(*) FILTER (WHERE e.ok)::int AS ok,
               count(*)::int AS total,
               coalesce(round(percentile_cont(0.5) WITHIN GROUP (ORDER BY e.latency_ms)),0)::int AS p50_ms
        FROM bot_trace_events e WHERE {REAL} AND e.stage='stt.result'
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1 ORDER BY ok DESC
    """, p)
    enhance = await _rows(db, f"""
        SELECT count(*) FILTER (WHERE e.ok)::int AS ok,
               count(*)::int AS total,
               coalesce(round(percentile_cont(0.5) WITHIN GROUP (ORDER BY e.latency_ms)),0)::int AS p50_ms
        FROM bot_trace_events e WHERE {REAL} AND e.stage='audio.enhance.result'
          AND e.created_at >= now() - (:days * interval '1 day')
    """, p)
    return {"data": {"providers": providers, "enhance": enhance[0] if enhance else None}}


# ---------------------------------------------------------------------------
# per-MODEL records (v2.1) — one row per AI model / engine in the pipeline
# ---------------------------------------------------------------------------
# result stage(s) per model; latency lives in the latency_ms column with the
# payload copy as fallback for pre-v2.1 rows; tokens in payload->'usage'.
MODEL_DEFS: list[dict] = [
    {"id": "jev", "name": "Jev (تطبیق موجودیت)", "kind": "matching",
     "stages": ["jev.request", "jev.result", "jev.failure"], "result_stage": "jev.result",
     "model_stages": ["jev.result"], "pinned": "jev-1.13.0"},  # the bot's pinned version (calibrated)
    {"id": "deepseek-nlu", "name": "DeepSeek — درک پیام (NLU)", "kind": "llm",
     "stages": ["nlu.request", "nlu.result"], "result_stage": "nlu.result",
     "model_stages": ["nlu.result"], "pinned": None},
    {"id": "deepseek-arbiter", "name": "DeepSeek — داور تطبیق", "kind": "llm",
     "stages": ["matching.arbitration.result", "matching.arbitration.failure"], "result_stage": "matching.arbitration.result",
     "model_stages": [], "pinned": None},
    {"id": "deepseek-fallback", "name": "DeepSeek — فالبک بحرانی", "kind": "llm",
     "stages": ["matching.fallback.result", "matching.fallback.failure"], "result_stage": "matching.fallback.result",
     "model_stages": [], "pinned": None},
    {"id": "audio-enhancer", "name": "پیش‌پردازش صدا (Denoise)", "kind": "audio",
     "stages": ["audio.enhance.result"], "result_stage": "audio.enhance.result",
     "model_stages": ["audio.enhance.result"], "pinned": None},
]

_STAGE_SECTION = {
    "audio.enhance.result": ("صوت", "پیش‌پردازش و نویززدایی"),
    "stt.run": ("صوت", "اجرای STT"),
    "stt.result": ("صوت", "تبدیل گفتار به متن"),
    "nlu.request": ("درک پیام", "درخواست NLU"),
    "nlu.result": ("درک پیام", "پاسخ NLU"),
    "entities.extracted": ("درک پیام", "موجودیت‌های استخراج‌شده"),
    "jev.request": ("تطبیق", "درخواست Jev"),
    "jev.result": ("تطبیق", "امتیازدهی Jev"),
    "jev.failure": ("تطبیق", "خطای Jev"),
    "routing.decision": ("تطبیق", "تصمیم مسیریابی"),
    "matching.arbitration.result": ("تطبیق", "داوری DeepSeek"),
    "matching.arbitration.failure": ("تطبیق", "خطای داوری"),
    "matching.fallback.result": ("تطبیق", "فالبک بحرانی"),
    "matching.fallback.failure": ("تطبیق", "خطای فالبک"),
    "ambiguity.sent": ("ابهام و انتخاب", "ارسال پرسش ابهام"),
    "ambiguity.user_choice": ("ابهام و انتخاب", "انتخاب کاربر"),
    "user_choice.pick": ("ابهام و انتخاب", "دکمه‌ی گزینه"),
    "draft.render": ("پیش‌نویس", "نمایش پیش‌نویس"),
    "draft.edit": ("پیش‌نویس", "ویرایش پیش‌نویس"),
    "draft.commit": ("پیش‌نویس", "ثبت نهایی"),
    "draft.cancel": ("پیش‌نویس", "لغو"),
}

_LAT = "coalesce(e.latency_ms, nullif(e.payload->>'latency_ms','')::int)"
_TOK = "(e.payload->'usage'->>'input_tokens')::bigint"
_TOK_OUT = "(e.payload->'usage'->>'output_tokens')::bigint"


async def _model_rows(db: AsyncSession, days: int, stage: str) -> list[dict]:
    return await _rows(db, f"""
        SELECT count(*)::int AS calls,
               count(*) FILTER (WHERE e.ok)::int AS ok,
               coalesce(round(percentile_cont(0.5) WITHIN GROUP (ORDER BY {_LAT})),0)::int AS p50,
               coalesce(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY {_LAT})),0)::int AS p95,
               coalesce(sum({_TOK}),0)::bigint AS tokens_in,
               coalesce(sum({_TOK_OUT}),0)::bigint AS tokens_out,
               max(e.created_at)::text AS last_seen
        FROM bot_trace_events e WHERE {REAL} AND e.stage = :stage
          AND e.created_at >= now() - (:days * interval '1 day')
    """, {"stage": stage, "days": days})


async def _models_seen(db: AsyncSession, days: int, stages: list[str]) -> list[dict]:
    return await _rows(db, f"""
        SELECT DISTINCT ON (coalesce(e.model,'?')) coalesce(e.model,'?') AS model,
               count(*) OVER (PARTITION BY coalesce(e.model,'?'))::int AS n
        FROM bot_trace_events e
        WHERE {REAL} AND e.stage = ANY(:stages)
          AND e.created_at >= now() - (:days * interval '1 day')
    """, {"stages": stages, "days": days})


@router.get("/models")
async def models(
    days: int = Query(7, ge=1, le=90),
    db: AsyncSession = Depends(get_db),
    _ctx=Depends(require_role("admin")),
):
    """One record per model in the pipeline (incl. dynamic STT providers)."""
    out: list[dict] = []
    for d in MODEL_DEFS:
        agg = (await _model_rows(db, days, d["result_stage"]))[0]
        versions: list[str] = []
        if d["model_stages"]:
            versions = [r["model"] for r in await _models_seen(db, days, d["model_stages"])
                        if r["model"] != "?"]
        out.append({
            "id": d["id"], "name": d["name"], "kind": d["kind"],
            "stages": d["stages"], **agg,
            "versions": versions,
            "drift": bool(d["pinned"] and any(v != d["pinned"] for v in versions)),
            "pinned": d["pinned"],
        })
    # dynamic STT models — one record per provider/model id in stt.result
    stt_rows = await _rows(db, f"""
        SELECT coalesce(e.model, e.payload->>'provider', 'stt') AS model,
               count(*)::int AS calls,
               count(*) FILTER (WHERE e.ok)::int AS ok,
               coalesce(round(percentile_cont(0.5) WITHIN GROUP (ORDER BY {_LAT})),0)::int AS p50,
               coalesce(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY {_LAT})),0)::int AS p95,
               0::bigint AS tokens_in, 0::bigint AS tokens_out,
               max(e.created_at)::text AS last_seen
        FROM bot_trace_events e
        WHERE {REAL} AND e.stage = 'stt.result'
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1 ORDER BY calls DESC
    """, {"days": days})
    for r in stt_rows:
        out.append({"id": f"stt:{r['model']}",
                    "name": f"STT — {r['model']}", "kind": "stt",
                    "stages": ["stt.result"], **r,
                    "versions": [], "drift": False, "pinned": None})
    return {"data": out, "days": days}


def _model_def(model_id: str) -> dict | None:
    for d in MODEL_DEFS:
        if d["id"] == model_id:
            return d
    return None


@router.get("/models/{model_id}/daily")
async def model_daily(
    model_id: str,
    days: int = Query(7, ge=1, le=90),
    db: AsyncSession = Depends(get_db),
    _ctx=Depends(require_role("admin")),
):
    """Daily calls / failures / tokens / latency for one model."""
    if model_id.startswith("stt:"):
        stage, model_val = "stt.result", model_id[4:]
        extra = " AND coalesce(e.model, e.payload->>'provider') = :mv"
        params: dict = {"mv": model_val}
    else:
        d = next((x for x in MODEL_DEFS if x["id"] == model_id), None)
        if d is None:
            raise NotFoundError("مدل یافت نشد")
        stage, model_val, extra = d["result_stage"], None, ""
        params = {}
    rows = await _rows(db, f"""
        SELECT to_char(date_trunc('day', e.created_at), 'YYYY-MM-DD') AS day,
               count(*)::int AS calls,
               count(*) FILTER (WHERE NOT e.ok)::int AS fails,
               coalesce(round(percentile_cont(0.5) WITHIN GROUP (ORDER BY {_LAT})),0)::int AS p50,
               coalesce(sum({_TOK}),0)::bigint AS tokens_in,
               coalesce(sum({_TOK_OUT}),0)::bigint AS tokens_out
        FROM bot_trace_events e
        WHERE {REAL} AND e.stage = :stage{extra}
          AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1 ORDER BY 1
    """, {"stage": stage, "days": days, **params})
    return {"data": rows, "model": model_id, "days": days}


@router.get("/stages")
async def stages(
    days: int = Query(7, ge=1, le=90),
    db: AsyncSession = Depends(get_db),
    _ctx=Depends(require_role("admin")),
):
    """One record per pipeline stage, tagged with its Persian section."""
    rows = await _rows(db, f"""
        SELECT e.stage,
               count(*)::int AS n,
               count(*) FILTER (WHERE NOT e.ok)::int AS fails,
               coalesce(round(percentile_cont(0.5) WITHIN GROUP (ORDER BY {_LAT})),0)::int AS p50,
               coalesce(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY {_LAT})),0)::int AS p95,
               coalesce(sum({_TOK}),0)::bigint AS tokens_in,
               coalesce(sum({_TOK_OUT}),0)::bigint AS tokens_out
        FROM bot_trace_events e
        WHERE {REAL} AND e.created_at >= now() - (:days * interval '1 day')
        GROUP BY 1 ORDER BY n DESC
    """, {"days": days})
    for r in rows:
        sec = _STAGE_SECTION.get(r["stage"])
        r["section"] = sec[0] if sec else "سایر"
        r["stage_fa"] = sec[1] if sec else r["stage"]
    return {"data": rows, "days": days}


@router.get("/traces")
async def traces(
    limit: int = Query(50, ge=1, le=200),
    status: str | None = None,
    include_synthetic: bool = False,
    db: AsyncSession = Depends(get_db),
    _ctx=Depends(require_role("admin")),
):
    """Latest message records («پرونده‌ی پیام») with a text excerpt + media flag."""
    sql = """
        SELECT t.id::text AS id, t.created_at::text AS created_at,
               coalesce(t.source,'bale') AS source, t.chat_id::text AS chat_id,
               t.kind, coalesce(t.status,'open') AS status,
               (SELECT count(*) FROM bot_trace_events e WHERE e.trace_id=t.id)::int AS events,
               CASE WHEN EXISTS (SELECT 1 FROM bot_trace_events e
                                  WHERE e.trace_id=t.id AND e.stage='input.voice')
                    THEN 'voice' ELSE 'text' END AS input_kind,
               left(coalesce((SELECT e.payload->>'consensus_text' FROM bot_trace_events e
                               WHERE e.trace_id=t.id AND e.stage='nlu.result'
                               ORDER BY e.created_at LIMIT 1),
                             (SELECT e.payload->>'text' FROM bot_trace_events e
                               WHERE e.trace_id=t.id AND e.stage='input.text'
                               ORDER BY e.created_at LIMIT 1),
                             (SELECT e.payload->>'text' FROM bot_trace_events e
                               WHERE e.trace_id=t.id AND e.stage='stt.result' AND e.ok
                               ORDER BY e.created_at LIMIT 1),
                             ''), 120) AS text_excerpt
        FROM bot_traces t
        WHERE (:include_synthetic
               OR (coalesce(t.source,'bale') NOT IN ('calibration','demo') AND coalesce(t.chat_id,0) <> 0))
    """
    params: dict = {"include_synthetic": include_synthetic}
    if status:
        sql += " AND t.status = :status"
        params["status"] = status
    sql += " ORDER BY t.created_at DESC LIMIT :limit"
    params["limit"] = limit
    return {"data": await _rows(db, sql, params)}


@router.get("/traces/{trace_id}")
async def trace_detail(
    trace_id: str,
    db: AsyncSession = Depends(get_db),
    _ctx=Depends(require_role("admin")),
):
    """Full event list of one trace (the drawer/drill-down view)."""
    events = await _rows(db, """
        SELECT e.created_at::text AS created_at, e.stage, e.ok,
               e.latency_ms, e.model, e.field, e.payload
        FROM bot_trace_events e
        WHERE e.trace_id::text = :tid
        ORDER BY e.created_at ASC, e.id ASC
        LIMIT 500
    """, {"tid": trace_id})
    head = await _rows(db, """
        SELECT id::text AS id, created_at::text AS created_at,
               coalesce(source,'bale') AS source, chat_id::text AS chat_id,
               kind, coalesce(status,'open') AS status,
               draft_id::text AS draft_id, committed_quote_id::text AS committed_quote_id
        FROM bot_traces WHERE id::text = :tid
    """, {"tid": trace_id})
    return {"data": {"trace": head[0] if head else None, "events": events}}


# ---------------------------------------------------------------------------
# media serving (v2.1) — inline playback + download of voices / user files
# ---------------------------------------------------------------------------
# <audio controls> and <a download> cannot attach an Authorization header, so
# the admin page first mints a short-lived signed URL (same idea as the quote
# preview tokens); the GET below accepts that token OR an admin session and
# streams the file from the bot's MinIO bucket, falling back to the read-only
# ./bot/data mount for pre-MinIO files.

_MEDIA_TTL_SECONDS = 600
_MEDIA_RE = re.compile(r"^(voices|attachments)/[A-Za-z0-9_\-./]+$")
_MEDIA_EXT = {".ogg", ".wav", ".m4a", ".mp3", ".pdf", ".png", ".jpg", ".jpeg", ".webp"}
_MEDIA_MIME = {
    ".ogg": "audio/ogg", ".wav": "audio/wav", ".m4a": "audio/mp4",
    ".mp3": "audio/mpeg", ".pdf": "application/pdf", ".png": "image/png",
    ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp",
}


def _check_ref(ref: str) -> str:
    """Validate a media ref (path under ./bot/data, doubles as the MinIO key)."""
    ref = (ref or "").strip()
    if not _MEDIA_RE.match(ref) or ".." in ref or ref.endswith("."):
        raise ValidationError("مسیر فایل نامعتبر است")
    if os.path.splitext(ref)[1].lower() not in _MEDIA_EXT:
        raise ValidationError("نوع فایل پشتیبانی نمی‌شود")
    return ref


def _media_secret() -> bytes:
    base = (settings.CASDOOR_CLIENT_SECRET or settings.MINIO_SECRET_KEY
            or "procurement-preview-fallback")
    return base.encode("utf-8")


def _media_sign(ref: str, expiry: int) -> str:
    return hmac.new(_media_secret(), f"bot-media:{ref}:{expiry}".encode(),
                    hashlib.sha256).hexdigest()


def _media_token(ref: str, ttl: int = _MEDIA_TTL_SECONDS) -> str:
    expiry = int(time.time()) + ttl
    return f"{expiry}.{_media_sign(ref, expiry)}"


def _verify_media_token(token: str, ref: str) -> bool:
    if not token or "." not in token:
        return False
    expiry_raw, sig = token.split(".", 1)
    try:
        expiry = int(expiry_raw)
    except ValueError:
        return False
    if expiry < int(time.time()):
        return False
    return hmac.compare_digest(_media_sign(ref, expiry), sig)


from pydantic import BaseModel  # noqa: E402  (kept next to its only use)


class MediaUrlRequest(BaseModel):
    ref: str


@router.post("/media-url")
async def media_url(
    body: MediaUrlRequest,
    _ctx=Depends(require_role("admin")),
):
    """Mint a short-lived signed URL for inline playback / download."""
    ref = _check_ref(body.ref)
    from urllib.parse import quote
    return {"data": {
        "url": f"/api/v1/bot-metrics/media?ref={quote(ref)}&t={_media_token(ref)}",
        "expires_in": _MEDIA_TTL_SECONDS,
    }}


@router.get("/media")
async def media(
    request: Request,
    ref: str,
    t: str | None = None,
    db: AsyncSession = Depends(get_db),
):
    """Stream one bot media file (admin session OR signed ?t= token).

    Prefers the MinIO copy (bucket bot-media, key = ref); falls back to the
    read-only disk mount for files saved before MinIO mirroring existed.
    """
    ref = _check_ref(ref)
    allowed = False
    if request.headers.get("authorization"):
        user = await get_optional_user(request, db)
        if user is not None and user.role == "admin":
            allowed = True
    if not allowed and t and _verify_media_token(t, ref):
        allowed = True
    if not allowed:
        raise UnauthorizedError("دسترسی به این فایل مجاز نیست")

    name = os.path.basename(ref)
    mime = _MEDIA_MIME.get(os.path.splitext(ref)[1].lower(), "application/octet-stream")

    # 1) MinIO copy (the bot dual-writes everything here)
    try:
        import asyncio

        from app.services.storage import get_client as _minio_client

        def _fetch() -> bytes:
            resp = _minio_client().get_object(settings.BOT_MEDIA_BUCKET, ref)
            try:
                return resp.read()
            finally:
                resp.close()
                resp.release_conn()

        content = await asyncio.to_thread(_fetch)
        return Response(content, media_type=mime, headers={
            "Content-Disposition": f'inline; filename="{name}"',
            "Cache-Control": "private, max-age=300",
        })
    except Exception:
        pass  # not in MinIO (old file) or MinIO down → disk fallback

    # 2) disk fallback (read-only mount of ./bot/data)
    path = (Path("/srv/bot/data") / ref).resolve()
    data_root = Path("/srv/bot/data").resolve()
    if data_root not in path.parents or not path.is_file():
        raise NotFoundError("فایل یافت نشد")
    return FileResponse(path, media_type=mime, filename=name,
                        headers={"Cache-Control": "private, max-age=300"})