from __future__ import annotations

import json
import statistics
import time
import uuid
from pathlib import Path
from typing import Any, Callable

from .context_budget import adaptive_context_budget, adaptive_retrieval_top_k, focused_context_budget
from .evidence_compression import focused_evidence_compression
from .eval_metrics import (
    answer_key_match,
    citation_metrics,
    mean,
    missing_answer_match,
    percentile,
    safe_div,
    scalar_value_match,
    source_metrics,
)
from .llm import GeminiGateway, RequestPacer
from .pipeline import RAGEngine
from .schemas import PipelineConfig, QueryPlan
from .security import prompt_injection_score
from .stress_eval import scale_stress_retrieval_eval
from .workspace import Workspace

ROOT = Path(__file__).resolve().parents[2]
BENCHMARK_PATH = ROOT / "evals" / "demo_benchmark.json"


def _load_benchmark() -> dict[str, Any]:
    return json.loads(BENCHMARK_PATH.read_text(encoding="utf-8"))


def demo_benchmark_metadata() -> dict[str, Any]:
    benchmark = _load_benchmark()
    return {
        "version": benchmark.get("version"),
        "description": benchmark.get("description", ""),
        "cases": {
            "focused_qa": len(benchmark.get("qa_cases", [])),
            "semantic_planner": len(benchmark.get("planner_cases", [])),
            "corpus_overview": len(benchmark.get("overview_cases", [])),
            "text2sql": len(benchmark.get("sql_cases", [])),
            "hard_mode": len(benchmark.get("hard_mode_cases", [])),
            "lifecycle_abstention": 2,
        },
        "levels": {
            "Quick": "Small deployment smoke test",
            "Standard": "Full deterministic benchmark, hard-mode robustness, retrieval/context/compression ablations, synthetic scale stress and release-readiness checks",
            "Deep": "Calibrated Gemini judge layered onto Standard; reuses a compatible saved Standard baseline when available",
        },
        "default_target_rpm": 12,
        "deep_judge_cases": sum(
            1 for case in benchmark.get("qa_cases", []) + benchmark.get("overview_cases", []) if case.get("deep_judge")
        ),
        "zero_gemini_ablations": ["reranker", "adaptive-context-budget", "evidence-compression", "scale-stress"],
        "cache_policy": (
            "RAG response cache is bypassed during fresh benchmark execution. Completed Quick/Standard/Deep reports "
            "can be saved per workspace, and Deep can reuse a compatible Standard deterministic baseline."
        ),
    }


def _document_sources(result_sources: list[dict[str, Any]], k: int = 5) -> list[str]:
    return [
        str(source.get("title", ""))
        for source in result_sources
        if source.get("type") == "document"
    ][:k]


def _evidence_text(result_sources: list[dict[str, Any]]) -> str:
    blocks = []
    for source in result_sources:
        sid = source.get("id", "?")
        title = source.get("title", "Source")
        snippet = source.get("snippet", "")
        url = source.get("url")
        blocks.append(f"[{sid}] {title}\n{('URL: ' + url + chr(10)) if url else ''}{snippet}")
    return "\n\n".join(blocks)


def _trace_efficiency(trace: dict[str, Any]) -> dict[str, Any]:
    metrics = trace.get("metrics", {})
    nodes = trace.get("nodes", [])
    names = [str(node.get("node", "")) for node in nodes]
    retrieve = next((node for node in reversed(nodes) if node.get("node") == "retrieve"), {})
    generate = next((node for node in reversed(nodes) if node.get("node") == "generate"), {})
    return {
        "node_count": int(metrics.get("node_count", len(nodes)) or 0),
        "llm_calls_estimate": int(metrics.get("llm_calls_estimate", 0) or 0),
        "web_used": bool(metrics.get("web_used", "web" in names)),
        "correction_used": bool(metrics.get("correction_used", "correct" in names)),
        "abstained": bool(metrics.get("abstained", "abstain" in names)),
        "cache_hit": bool(trace.get("cache_hit", False)),
        "context_pruning_used": bool(retrieve.get("context_pruning_used", False)),
        "context_chunks_before": int(retrieve.get("context_chunks_before", 0) or 0),
        "context_chunks_after": int(retrieve.get("context_chunks_after", 0) or 0),
        "context_tokens_est_before": int(retrieve.get("context_tokens_est_before", 0) or 0),
        "context_tokens_est_after": int(retrieve.get("context_tokens_est_after", 0) or 0),
        "context_reduction_pct": float(retrieve.get("context_reduction_pct", 0.0) or 0.0),
        "manifest_included": bool(generate.get("manifest_included", False)),
        "generation_prompt_tokens_est": int(generate.get("generation_prompt_tokens_est", 0) or 0),
        "generation_output_tokens_est": int(generate.get("generation_output_tokens_est", 0) or 0),
        "generation_total_tokens_est": int(generate.get("generation_total_tokens_est", 0) or 0),
        "evidence_source_utilization_rate": float(generate.get("evidence_source_utilization_rate", 0.0) or 0.0),
        "context_budget_target_chunks": int(retrieve.get("context_budget_target_chunks", 0) or 0),
        "context_budget_policy": str(retrieve.get("context_budget_policy", "")),
        "corpus_scale": str(retrieve.get("corpus_scale", "")),
        "retrieval_top_k": int(retrieve.get("retrieval_top_k", 0) or 0),
        "retrieval_confidence": float(retrieve.get("retrieval_confidence", 0.0) or 0.0),
        "retrieval_score_gap": float(retrieve.get("retrieval_score_gap", 0.0) or 0.0),
        "evidence_compression_used": bool(retrieve.get("evidence_compression_used", False)),
        "evidence_compression_reduction_pct": float(retrieve.get("evidence_compression_reduction_pct", 0.0) or 0.0),
        "evidence_tokens_est_after_compression": int(retrieve.get("evidence_tokens_est_after_compression", 0) or 0),
    }


def _trace_node_times(trace: dict[str, Any], pacing_wait_ms: float = 0.0) -> dict[str, float]:
    """Return approximate service-node time with deliberate eval pacing removed.

    RequestPacer sleeps happen inside the LLM node that is about to issue a
    provider request. The trace records node wall time, so evaluation subtracts
    the query-level deliberate pacing proportionally across nodes according to
    their recorded LLM-call counts. Raw trace JSON remains unchanged.
    """
    nodes = list(trace.get("nodes", []))
    total_calls = sum(int(node.get("llm_calls", 0) or 0) for node in nodes)
    wait_per_call = (max(0.0, float(pacing_wait_ms)) / total_calls) if total_calls else 0.0
    out: dict[str, float] = {}
    for node in nodes:
        name = str(node.get("node", ""))
        if not name:
            continue
        raw = float(node.get("ms", 0.0) or 0.0)
        node_wait = wait_per_call * int(node.get("llm_calls", 0) or 0)
        service = max(0.0, raw - node_wait)
        out[name] = out.get(name, 0.0) + service
    return out


def _chunk_rank_metrics(hits: list[Any], case: dict[str, Any]) -> tuple[float | None, float | None]:
    terms = [str(x).lower() for x in case.get("chunk_must_contain", [])]
    if not terms:
        return None, None
    first_rank = 0
    for rank, hit in enumerate(hits[:5], start=1):
        text = hit.chunk.text.lower()
        if all(term in text for term in terms):
            first_rank = rank
            break
    return (float(first_rank == 1), (1.0 / first_rank if first_rank else 0.0))


def _node_latency_summary(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    buckets: dict[str, list[float]] = {}
    for row in rows:
        for node, ms in (row.get("_node_times") or {}).items():
            buckets.setdefault(node, []).append(float(ms))
    out = []
    for node, values in buckets.items():
        out.append({
            "node": node,
            "mean_ms": round(mean(values), 1),
            "p50_ms": round(percentile(values, 0.50), 1),
            "p95_ms": round(percentile(values, 0.95), 1),
            "samples": len(values),
        })
    return sorted(out, key=lambda row: float(row["mean_ms"]), reverse=True)


def _retrieval_ablation(workspace: Workspace, qa_cases: list[dict[str, Any]]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for use_reranker in (False, True):
        metric_buckets: dict[str, list[float]] = {
            "source_recall@5": [],
            "source_hit@1": [],
            "source_mrr": [],
            "source_ap@5": [],
            "source_ndcg@5": [],
            "source_duplicate_rate@5": [],
            "chunk_hit@1": [],
            "chunk_mrr": [],
        }
        latencies: list[float] = []
        for case in qa_cases:
            started = time.perf_counter()
            hits = workspace.retriever.search(case["question"], top_k=5, use_reranker=use_reranker)
            latency = (time.perf_counter() - started) * 1000
            returned = [hit.chunk.source for hit in hits]
            metrics = source_metrics(returned, case.get("relevant_sources", []))
            chunk_hit, chunk_mrr = _chunk_rank_metrics(hits, case)
            if chunk_hit is not None:
                metrics["chunk_hit@1"] = chunk_hit
                metrics["chunk_mrr"] = chunk_mrr
            for key in metric_buckets:
                if key in metrics:
                    metric_buckets[key].append(float(metrics[key]))
            latencies.append(latency)
        row = {
            "configuration": "Hybrid + reranker" if use_reranker else "Hybrid RRF",
            **{key: round(mean(values), 3) for key, values in metric_buckets.items()},
            "median_retrieval_ms": round(statistics.median(latencies), 1) if latencies else 0.0,
        }
        rows.append(row)
    return rows



def _context_budget_ablation(workspace: Workspace, qa_cases: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Zero-Gemini comparison of full, fixed, and adaptive context budgets."""
    labels = ("Full top-k", "Fixed 3-chunk budget", "Adaptive budget")
    buckets: dict[str, list[dict[str, float]]] = {label: [] for label in labels}
    for case in qa_cases:
        plan = QueryPlan(
            route="documents",
            knowledge_scope="corpus",
            task_type="fact_lookup",
            retrieval_strategy="semantic",
            web_relevance="irrelevant",
            rewritten_query=case["question"],
            document_queries=[case["question"]],
        )
        cfg = PipelineConfig(profile="Balanced", top_k=6, use_context_pruning=True, use_adaptive_top_k=True)
        effective_k = adaptive_retrieval_top_k(
            cfg, plan, corpus_chunks=len(workspace.chunks), corpus_sources=len(workspace.source_profiles)
        )
        hits = workspace.retriever.search(case["question"], top_k=effective_k, use_reranker=False)
        fixed = focused_context_budget(hits, plan, cfg)
        adaptive = adaptive_context_budget(
            hits, plan, cfg, corpus_chunks=len(workspace.chunks), corpus_sources=len(workspace.source_profiles)
        )
        variants = {
            "Full top-k": (hits, len(hits), "full"),
            "Fixed 3-chunk budget": (fixed.hits, fixed.target_chunks, fixed.reason),
            "Adaptive budget": (adaptive.hits, adaptive.target_chunks, adaptive.reason),
        }
        full_chars = max(1, sum(len(hit.chunk.text or "") + len(hit.chunk.source or "") + 24 for hit in hits))
        for label, (variant, target, reason) in variants.items():
            returned = [hit.chunk.source for hit in variant]
            metrics = source_metrics(returned, case.get("relevant_sources", []))
            chars = sum(len(hit.chunk.text or "") + len(hit.chunk.source or "") + 24 for hit in variant)
            buckets[label].append({
                "source_precision@5": float(metrics["source_precision@5"]),
                "source_recall@5": float(metrics["source_recall@5"]),
                "source_hit@1": float(metrics["source_hit@1"]),
                "source_mrr": float(metrics["source_mrr"]),
                "context_chunks": float(len(variant)),
                "target_chunks": float(target),
                "context_sources": float(len(set(returned))),
                "context_chars": float(chars),
                "context_tokens_est": float((chars + 3) // 4),
                "context_reduction_pct": float(max(0.0, 1.0 - chars / full_chars) * 100.0),
                "adaptive_used": float(label == "Adaptive budget" and len(variant) < len(hits)),
            })

    rows: list[dict[str, Any]] = []
    for label in labels:
        values = buckets[label]
        rows.append({
            "configuration": label,
            "source_precision@5": round(mean([v["source_precision@5"] for v in values]), 3),
            "source_recall@5": round(mean([v["source_recall@5"] for v in values]), 3),
            "source_hit@1": round(mean([v["source_hit@1"] for v in values]), 3),
            "source_mrr": round(mean([v["source_mrr"] for v in values]), 3),
            "median_context_chunks": round(statistics.median([v["context_chunks"] for v in values]), 1) if values else 0.0,
            "median_target_chunks": round(statistics.median([v["target_chunks"] for v in values]), 1) if values else 0.0,
            "median_context_sources": round(statistics.median([v["context_sources"] for v in values]), 1) if values else 0.0,
            "median_context_chars": round(statistics.median([v["context_chars"] for v in values]), 1) if values else 0.0,
            "median_context_tokens_est": round(statistics.median([v["context_tokens_est"] for v in values]), 1) if values else 0.0,
            "median_context_reduction_pct": round(statistics.median([v["context_reduction_pct"] for v in values]), 1) if values else 0.0,
        })
    return rows


def _compression_signal(case: dict[str, Any], text: str) -> bool:
    lower = (text or "").lower()
    required = [str(term).lower() for term in case.get("chunk_must_contain", [])]
    if required:
        return all(term in lower for term in required)
    expected = [str(term).lower() for term in case.get("expected_any", [])]
    return any(term in lower for term in expected) if expected else True


def _evidence_compression_ablation(workspace: Workspace, qa_cases: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Zero-Gemini test that focused sentence compression retains labeled answer signals."""
    full_tokens: list[float] = []
    compressed_tokens: list[float] = []
    retention: list[float] = []
    reductions: list[float] = []
    for case in qa_cases:
        plan = QueryPlan(
            route="documents", knowledge_scope="corpus", task_type="fact_lookup", retrieval_strategy="semantic",
            web_relevance="irrelevant", rewritten_query=case["question"], document_queries=[case["question"]],
        )
        cfg = PipelineConfig(profile="Balanced", top_k=6, use_context_pruning=True, use_adaptive_top_k=True)
        effective_k = adaptive_retrieval_top_k(
            cfg, plan, corpus_chunks=len(workspace.chunks), corpus_sources=len(workspace.source_profiles)
        )
        hits = workspace.retriever.search(case["question"], top_k=effective_k, use_reranker=False)
        budget = adaptive_context_budget(
            hits, plan, cfg, corpus_chunks=len(workspace.chunks), corpus_sources=len(workspace.source_profiles)
        )
        compression = focused_evidence_compression(budget.hits, plan, query=case["question"], enabled=True)
        before_text = "\n".join(hit.chunk.text or "" for hit in budget.hits)
        after_text = "\n".join(compression.texts.get(hit.chunk.id, hit.chunk.text or "") for hit in budget.hits)
        full_tokens.append(float((len(before_text) + 3) // 4))
        compressed_tokens.append(float((len(after_text) + 3) // 4))
        retention.append(float(_compression_signal(case, after_text)))
        reductions.append(float(compression.reduction_ratio * 100.0))

    return [
        {
            "configuration": "Adaptive context only",
            "answer_signal_retention": 1.0,
            "median_evidence_tokens_est": round(statistics.median(full_tokens), 1) if full_tokens else 0.0,
            "median_additional_reduction_pct": 0.0,
            "cases": len(qa_cases),
        },
        {
            "configuration": "Adaptive + sentence compression",
            "answer_signal_retention": round(mean(retention), 3),
            "median_evidence_tokens_est": round(statistics.median(compressed_tokens), 1) if compressed_tokens else 0.0,
            "median_additional_reduction_pct": round(statistics.median(reductions), 1) if reductions else 0.0,
            "cases": len(qa_cases),
        },
    ]


def _readiness_rows(summary: dict[str, Any], scale_rows: list[dict[str, Any]], compression_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    level = str(summary.get("evaluation_level", "Standard"))
    largest_scale = scale_rows[-1] if scale_rows else {}
    compressed = compression_rows[-1] if compression_rows else {}
    checks = [
        ("Answer accuracy", float(summary.get("answer_accuracy", 0.0)), 0.95, True),
        ("Source recall", float(summary.get("source_recall@5", 0.0)), 0.95, True),
        ("Citation validity", float(summary.get("citation_validity", 0.0)), 0.95, True),
        ("Citation coverage", float(summary.get("citation_coverage", 0.0)), 0.90, True),
        ("Planner route", float(summary.get("planner_route_accuracy", 0.0)), 0.90, True),
        ("Planner task", float(summary.get("planner_task_accuracy", 0.0)), 0.90, True),
        ("Planner strategy", float(summary.get("planner_strategy_accuracy", 0.0)), 0.90, True),
        ("Web precision", float(summary.get("web_use_precision", 0.0)), 0.90, True),
        ("Text2SQL", float(summary.get("text2sql_pass_rate", 0.0)), 0.90, True),
        ("Hard mode", float(summary.get("hard_mode_pass_rate", 0.0)), 0.85, True),
        ("Corpus overview", float(summary.get("overview_pass_rate", 0.0)), 0.90, True),
    ]
    # Quick intentionally skips local ablations and scale stress. Do not award
    # implicit PASS results for tests that were not executed. Standard/Deep
    # require them and missing rows therefore fail the corresponding gate.
    if level != "Quick":
        checks.extend([
            ("Adaptive-budget recall", float(summary.get("context_pruning_recall@5", 0.0)), 0.95, True),
            ("Compression signal retention", float(compressed.get("answer_signal_retention", 0.0)), 0.95, True),
            ("20x stress recall", float(largest_scale.get("source_recall@5", 0.0)), 0.95, True),
            ("20x stress pruned recall", float(largest_scale.get("adaptive_pruned_recall@5", 0.0)), 0.95, True),
        ])
    rows = []
    for name, value, threshold, critical in checks:
        rows.append({
            "check": name,
            "value": round(value, 3),
            "threshold": threshold,
            "status": "PASS" if value >= threshold else "FAIL",
            "critical": critical,
        })
    return rows


def _readiness_summary(rows: list[dict[str, Any]]) -> tuple[str, float]:
    critical = [row for row in rows if row.get("critical")]
    passed = sum(1 for row in critical if row.get("status") == "PASS")
    score = safe_div(passed, len(critical)) if critical else 0.0
    if score >= 1.0:
        return "READY", score
    if score >= 0.85:
        return "WATCH", score
    return "NOT READY", score


def _planner_eval(
    workspace: Workspace,
    cases: list[dict[str, Any]],
    gateway: GeminiGateway,
    progress: Callable[[float, str], None],
    start: float,
    span: float,
) -> list[dict[str, Any]]:
    manifest = workspace.manifest()
    rows: list[dict[str, Any]] = []
    total = max(1, len(cases))
    for idx, case in enumerate(cases, start=1):
        progress(start + span * (idx - 1) / total, f"Planner case {idx}/{len(cases)}")
        wait_before = gateway.request_pacer.total_sleep_seconds() if gateway.request_pacer else 0.0
        began = time.perf_counter()
        plan = gateway.analyze_query(case["question"], manifest, history=None, profile="Balanced")
        wall_latency = (time.perf_counter() - began) * 1000
        wait_after = gateway.request_pacer.total_sleep_seconds() if gateway.request_pacer else wait_before
        pacing_wait = max(0.0, wait_after - wait_before) * 1000
        latency = max(0.0, wall_latency - pacing_wait)
        planned_web = plan.web_relevance != "irrelevant" or plan.route in {"web", "hybrid"}
        rows.append(
            {
                "id": case["id"],
                "question": case["question"],
                "expected_route": case["route"],
                "route": plan.route,
                "route_correct": plan.route == case["route"],
                "expected_task": case["task"],
                "task": plan.task_type,
                "task_correct": plan.task_type == case["task"],
                "expected_strategy": case["strategy"],
                "strategy": plan.retrieval_strategy,
                "strategy_correct": plan.retrieval_strategy in set(case.get("strategy_any", [case["strategy"]])),
                "expected_web": bool(case["web_expected"]),
                "planned_web": planned_web,
                "latency_ms": round(latency, 1),
                "wall_latency_ms": round(wall_latency, 1),
                "pacing_wait_ms": round(pacing_wait, 1),
            }
        )
    return rows


def _judge_row(
    judge: GeminiGateway,
    case: dict[str, Any],
    answer: str,
    sources: list[dict[str, Any]],
    citations: dict[str, float | int],
) -> dict[str, Any]:
    wait_before = judge.request_pacer.total_sleep_seconds() if judge.request_pacer else 0.0
    began = time.perf_counter()
    judgement = judge.evaluate_rag_answer(
        case["question"],
        answer,
        _evidence_text(sources),
        case.get("reference_answer", ""),
        citation_validity=float(citations["citation_validity"]),
        citation_coverage=float(citations["citation_coverage"]),
    )
    judge_wall_latency = (time.perf_counter() - began) * 1000
    wait_after = judge.request_pacer.total_sleep_seconds() if judge.request_pacer else wait_before
    judge_pacing_wait = max(0.0, wait_after - wait_before) * 1000
    judge_latency = max(0.0, judge_wall_latency - judge_pacing_wait)
    return {
        "judge_faithfulness": round(judgement.faithfulness, 3),
        "judge_answer_relevance": round(judgement.answer_relevance, 3),
        "judge_completeness": round(judgement.completeness, 3),
        "judge_citation_support": round(judgement.citation_support, 3),
        "judge_overall": round(judgement.overall, 3),
        "judge_pass": judgement.pass_,
        "judge_reason": judgement.reason,
        "judge_latency_ms": round(judge_latency, 1),
        "judge_wall_latency_ms": round(judge_wall_latency, 1),
        "judge_pacing_wait_ms": round(judge_pacing_wait, 1),
    }


def _qa_eval(
    workspace: Workspace,
    cases: list[dict[str, Any]],
    api_key: str | None,
    model: str,
    deep_judge: bool,
    request_pacer: RequestPacer,
    progress: Callable[[float, str], None],
    start: float,
    span: float,
) -> list[dict[str, Any]]:
    engine = RAGEngine(workspace, request_pacer=request_pacer)
    judge = GeminiGateway(api_key, model, request_pacer=request_pacer) if deep_judge else None
    cfg = PipelineConfig(
        mode="Documents",
        profile="Fast",
        model=model,
        use_crag=False,
        allow_web_fallback=False,
        use_self_rag=False,
    )
    rows: list[dict[str, Any]] = []
    total = max(1, len(cases))
    for idx, case in enumerate(cases, start=1):
        progress(start + span * (idx - 1) / total, f"Document QA case {idx}/{len(cases)}")
        wait_before = request_pacer.total_sleep_seconds()
        began = time.perf_counter()
        result = engine.ask(case["question"], cfg, api_key, use_cache=False, record_history=False)
        wall_latency = (time.perf_counter() - began) * 1000
        pacing_wait = max(0.0, request_pacer.total_sleep_seconds() - wait_before) * 1000
        latency = max(0.0, wall_latency - pacing_wait)
        returned_sources = _document_sources(result.sources, 5)
        retrieval = source_metrics(returned_sources, case.get("relevant_sources", []))
        citations = citation_metrics(result.answer, result.sources)
        efficiency = _trace_efficiency(result.trace)
        row: dict[str, Any] = {
            "id": case["id"],
            "question": case["question"],
            "answer_key_match": answer_key_match(result.answer, case),
            **{key: round(float(value), 3) for key, value in retrieval.items()},
            "citation_count": citations["citation_count"],
            "citation_validity": round(float(citations["citation_validity"]), 3),
            "citation_coverage": round(float(citations["citation_coverage"]), 3),
            "confidence": round(result.confidence, 3),
            "latency_ms": round(latency, 1),
            "wall_latency_ms": round(wall_latency, 1),
            "pacing_wait_ms": round(pacing_wait, 1),
            **efficiency,
            "_answer": result.answer,
            "_sources": result.sources,
            "_citations": citations,
            "_node_times": _trace_node_times(result.trace, pacing_wait),
        }
        if judge and bool(case.get("deep_judge", False)):
            row.update(_judge_row(judge, case, result.answer, result.sources, citations))
        rows.append(row)
    return rows


def _overview_eval(
    workspace: Workspace,
    cases: list[dict[str, Any]],
    api_key: str | None,
    model: str,
    deep_judge: bool,
    request_pacer: RequestPacer,
    progress: Callable[[float, str], None],
    start: float,
    span: float,
) -> list[dict[str, Any]]:
    engine = RAGEngine(workspace, request_pacer=request_pacer)
    judge = GeminiGateway(api_key, model, request_pacer=request_pacer) if deep_judge else None
    cfg = PipelineConfig(
        mode="Auto",
        profile="Balanced",
        model=model,
        allow_web_fallback=True,
        use_crag=True,
        use_self_rag=False,
    )
    rows: list[dict[str, Any]] = []
    total = max(1, len(cases))
    for idx, case in enumerate(cases, start=1):
        progress(start + span * (idx - 1) / total, f"Corpus overview case {idx}/{len(cases)}")
        wait_before = request_pacer.total_sleep_seconds()
        began = time.perf_counter()
        result = engine.ask(case["question"], cfg, api_key, use_cache=False, record_history=False)
        wall_latency = (time.perf_counter() - began) * 1000
        pacing_wait = max(0.0, request_pacer.total_sleep_seconds() - wait_before) * 1000
        latency = max(0.0, wall_latency - pacing_wait)
        plan = result.trace.get("query_plan", {})
        evidence = result.trace.get("evidence", {})
        efficiency = _trace_efficiency(result.trace)
        coverage = float(evidence.get("source_coverage", 0.0) or 0.0)
        citations = citation_metrics(result.answer, result.sources)
        doc_sources = set(_document_sources(result.sources, 20))
        expected_task = case["expected_task"]
        expected_strategy = case["expected_strategy"]
        actual_task = plan.get("task_type")
        actual_strategy = plan.get("retrieval_strategy")
        # A generic collection summary can legitimately be made richer as an
        # insight synthesis. Treat overview/global and insight/analytical as the
        # same broad-collection family for this suite, while still requiring
        # local routing, breadth and no unnecessary web usage.
        task_ok = actual_task == expected_task or (
            expected_task == "overview" and actual_task == "insight_synthesis"
        )
        strategy_ok = actual_strategy == expected_strategy or (
            expected_task == "overview"
            and actual_task == "insight_synthesis"
            and actual_strategy == "analytical"
        )
        passed = (
            plan.get("route") == case["expected_route"]
            and task_ok
            and strategy_ok
            and efficiency["web_used"] == bool(case["web_expected"])
            and coverage >= float(case.get("min_source_coverage", 0.0))
            and float(citations.get("citation_validity", 0.0)) >= 0.80
        )
        row: dict[str, Any] = {
            "id": case["id"],
            "question": case["question"],
            "route": plan.get("route"),
            "task": plan.get("task_type"),
            "strategy": plan.get("retrieval_strategy"),
            "task_semantic_match": task_ok,
            "strategy_semantic_match": strategy_ok,
            "web_used": efficiency["web_used"],
            "source_coverage": round(coverage, 3),
            "document_sources_returned": len(doc_sources),
            "citation_validity": round(float(citations["citation_validity"]), 3),
            "citation_coverage": round(float(citations["citation_coverage"]), 3),
            "latency_ms": round(latency, 1),
            "wall_latency_ms": round(wall_latency, 1),
            "pacing_wait_ms": round(pacing_wait, 1),
            "pass": passed,
            **efficiency,
            "_answer": result.answer,
            "_sources": result.sources,
            "_citations": citations,
            "_node_times": _trace_node_times(result.trace, pacing_wait),
        }
        if judge and bool(case.get("deep_judge", False)):
            row.update(_judge_row(judge, case, result.answer, result.sources, citations))
        rows.append(row)
    return rows


def _sql_eval(
    workspace: Workspace,
    cases: list[dict[str, Any]],
    api_key: str | None,
    model: str,
    request_pacer: RequestPacer,
    progress: Callable[[float, str], None],
    start: float,
    span: float,
) -> list[dict[str, Any]]:
    """Evaluate Text2SQL generation/execution with one model call per case.

    SQL routing is already measured in the semantic-planner benchmark. Keeping
    this component test route-independent avoids spending two extra Gemini
    calls per case just to duplicate planner and answer-generation coverage.
    """
    gateway = GeminiGateway(api_key, model, request_pacer=request_pacer)
    rows: list[dict[str, Any]] = []
    total = max(1, len(cases))
    for idx, case in enumerate(cases, start=1):
        progress(start + span * (idx - 1) / total, f"Text2SQL case {idx}/{len(cases)}")
        wait_before = request_pacer.total_sleep_seconds()
        began = time.perf_counter()
        try:
            sql, result = workspace.sql.benchmark_query(case["question"], gateway)
            preview = result.head(200)
            result_text = preview.to_markdown(index=False) if len(preview) else "(no rows)"
            observed_scalar = preview.iloc[0, 0] if len(preview) and len(preview.columns) else None
            if "expected_scalar" in case:
                matched = scalar_value_match(observed_scalar, case.get("expected_scalar"))
                match_method = "typed_scalar"
            else:
                matched = answer_key_match(result_text, case)
                match_method = "rendered_answer_key"
            error = ""
            readonly_validated = True
        except Exception as exc:
            sql = ""
            result = None
            result_text = ""
            matched = False
            match_method = "error"
            observed_scalar = None
            error = f"{type(exc).__name__}: {exc}"
            readonly_validated = False
        wall_latency = (time.perf_counter() - began) * 1000
        pacing_wait = max(0.0, request_pacer.total_sleep_seconds() - wait_before) * 1000
        latency = max(0.0, wall_latency - pacing_wait)
        rows.append(
            {
                "id": case["id"],
                "question": case["question"],
                "component": "Text2SQL",
                "answer_key_match": matched,
                "match_method": match_method,
                "observed_value": (
                    observed_scalar.item() if hasattr(observed_scalar, "item") else observed_scalar
                ),
                "expected_value": case.get("expected_scalar", ""),
                "readonly_validated": readonly_validated,
                "sql": sql,
                "rows": int(len(result)) if result is not None else 0,
                "latency_ms": round(latency, 1),
                "wall_latency_ms": round(wall_latency, 1),
                "pacing_wait_ms": round(pacing_wait, 1),
                "llm_calls_estimate": 1,
                "error": error,
            }
        )
    return rows


def _abstention_eval() -> list[dict[str, Any]]:
    empty = Workspace(f"eval-empty-{uuid.uuid4().hex[:10]}")
    engine = RAGEngine(empty)
    cases = [
        ("Documents", "What are the documents about?", "workspace_empty_documents"),
        ("Data (SQL)", "Which row has the highest value?", "workspace_empty_tables"),
    ]
    rows = []
    for mode, question, expected_reason in cases:
        result = engine.ask(
            question,
            PipelineConfig(mode=mode, profile="Fast"),
            api_key=None,
            use_cache=False,
            record_history=False,
        )
        nodes = result.trace.get("nodes", [])
        abstain_node = next((node for node in nodes if node.get("node") == "abstain"), {})
        rows.append(
            {
                "mode": mode,
                "question": question,
                "abstained": bool(abstain_node),
                "reason": abstain_node.get("reason"),
                "expected_reason": expected_reason,
                "pass": abstain_node.get("reason") == expected_reason,
                "llm_calls_estimate": result.trace.get("metrics", {}).get("llm_calls_estimate", 0),
            }
        )
    return rows


def _planner_summary(rows: list[dict[str, Any]]) -> dict[str, float]:
    tp = fp = fn = 0
    for row in rows:
        expected = bool(row["expected_web"])
        planned = bool(row["planned_web"])
        if expected and planned:
            tp += 1
        elif not expected and planned:
            fp += 1
        elif expected and not planned:
            fn += 1
    return {
        "planner_route_accuracy": mean([float(row["route_correct"]) for row in rows]),
        "planner_task_accuracy": mean([float(row["task_correct"]) for row in rows]),
        "planner_strategy_accuracy": mean([float(row["strategy_correct"]) for row in rows]),
        "web_use_precision": safe_div(tp, tp + fp),
        "web_use_recall": safe_div(tp, tp + fn),
        "unnecessary_web_rate": safe_div(fp, sum(1 for row in rows if not row["expected_web"])),
    }


def _base_grade(score: float) -> str:
    if score >= 0.90:
        return "A"
    if score >= 0.80:
        return "B"
    if score >= 0.70:
        return "C"
    if score >= 0.60:
        return "D"
    return "Needs work"


def _grade_with_gates(score: float, metrics: dict[str, float]) -> tuple[str, list[str]]:
    """Prevent a weighted average from hiding a badly failing subsystem."""
    order = ["Needs work", "D", "C", "B", "A"]
    grade = _base_grade(score)
    gates: list[str] = []

    def cap(max_grade: str, reason: str) -> None:
        nonlocal grade
        if order.index(grade) > order.index(max_grade):
            grade = max_grade
        gates.append(reason)

    if metrics["planner_route_accuracy"] < 0.90 or metrics["web_use_precision"] < 0.90:
        cap("C", "Critical routing/web-policy accuracy is below 90%.")
    if metrics["text2sql_pass_rate"] < 0.75:
        cap("B", "Text2SQL pass rate is below 75%.")
    if metrics["citation_validity"] < 0.90:
        cap("B", "Citation validity is below 90%.")
    if metrics["citation_coverage"] < 0.80:
        cap("B", "Citation coverage is below 80%.")
    if metrics["planner_task_accuracy"] < 0.75:
        cap("B", "Planner task taxonomy accuracy is below 75%.")
    if metrics.get("hard_mode_pass_rate", 1.0) < 0.50:
        cap("C", "Hard-mode robustness pass rate is below 50%.")
    elif metrics.get("hard_mode_pass_rate", 1.0) < 0.75:
        cap("B", "Hard-mode robustness pass rate is below 75%.")
    return grade, gates


def _hard_mode_eval(
    workspace: Workspace,
    cases: list[dict[str, Any]],
    api_key: str | None,
    model: str,
    request_pacer: RequestPacer,
    progress: Callable[[float, str], None],
    start: float,
    span: float,
) -> list[dict[str, Any]]:
    engine = RAGEngine(workspace, request_pacer=request_pacer)
    gateway = GeminiGateway(api_key, model, request_pacer=request_pacer)
    rows: list[dict[str, Any]] = []
    total = max(1, len(cases))
    for idx, case in enumerate(cases, start=1):
        progress(start + span * (idx - 1) / total, f"Hard-mode case {idx}/{len(cases)}")
        kind = case.get("kind", "qa")
        row: dict[str, Any] = {"id": case["id"], "kind": kind, "question": case.get("question", "")}
        if kind == "security":
            score = prompt_injection_score(case.get("text", ""))
            row.update({"injection_score": round(score, 3), "pass": score >= float(case.get("min_injection_score", 0.5)), "gemini_calls": 0})
            rows.append(row)
            continue
        if kind == "sql":
            wait_before = request_pacer.total_sleep_seconds()
            began = time.perf_counter()
            try:
                sql, result = workspace.sql.benchmark_query(case["question"], gateway)
                observed = result.iloc[0, 0] if len(result) and len(result.columns) else None
                passed = scalar_value_match(observed, case.get("expected_scalar"))
                error = ""
            except Exception as exc:
                sql, observed, passed = "", None, False
                error = f"{type(exc).__name__}: {exc}"
            wall = (time.perf_counter() - began) * 1000
            pace = max(0.0, request_pacer.total_sleep_seconds() - wait_before) * 1000
            row.update({"observed_value": observed.item() if hasattr(observed, "item") else observed, "expected_value": case.get("expected_scalar"), "sql": sql, "pass": passed, "latency_ms": round(max(0.0, wall-pace),1), "error": error, "gemini_calls": 1})
            rows.append(row)
            continue
        if kind == "planner":
            wait_before = request_pacer.total_sleep_seconds()
            began = time.perf_counter()
            plan = gateway.analyze_query(case["question"], workspace.manifest(), history=None, profile="Balanced")
            wall = (time.perf_counter() - began) * 1000
            pace = max(0.0, request_pacer.total_sleep_seconds() - wait_before) * 1000
            passed = plan.route == case["route"] and plan.task_type == case["task"] and plan.retrieval_strategy == case["strategy"] and ((plan.web_relevance != "irrelevant") == bool(case["web_expected"]))
            row.update({"route":plan.route,"task":plan.task_type,"strategy":plan.retrieval_strategy,"web_relevance":plan.web_relevance,"pass":passed,"latency_ms":round(max(0.0,wall-pace),1),"gemini_calls":1})
            rows.append(row)
            continue

        case_mode = str(case.get("mode") or "Documents")
        case_profile = str(case.get("profile") or "Fast")
        cfg = PipelineConfig(
            mode=case_mode,
            profile=case_profile,
            model=model,
            use_crag=bool(case.get("use_crag", False)),
            allow_web_fallback=False,
            use_self_rag=False,
        )
        if kind == "insight":
            cfg = PipelineConfig(mode="Auto", profile="Balanced", model=model, use_crag=True, allow_web_fallback=False, use_self_rag=False)
        wait_before = request_pacer.total_sleep_seconds()
        began = time.perf_counter()
        result = engine.ask(case["question"], cfg, api_key, use_cache=False, record_history=False)
        wall = (time.perf_counter() - began) * 1000
        pace = max(0.0, request_pacer.total_sleep_seconds() - wait_before) * 1000
        plan = result.trace.get("query_plan", {})
        citations = citation_metrics(result.answer, result.sources)
        returned = _document_sources(result.sources, 5)
        retrieval = source_metrics(returned, case.get("relevant_sources", [])) if case.get("relevant_sources") else {}
        if kind == "missing":
            grounded_absence = bool(result.trace.get("metrics", {}).get("grounded_absence", False))
            passed = missing_answer_match(result.answer, case) or grounded_absence
        elif kind == "insight":
            table_cited = any(str(src.get("id", "")).startswith("T") for src in result.sources) and "[T" in result.answer
            evidence = result.trace.get("evidence", {})
            passed = plan.get("task_type") == case.get("expected_task") and plan.get("retrieval_strategy") == case.get("expected_strategy") and float(evidence.get("source_coverage", 0.0) or 0.0) >= float(case.get("min_source_coverage", 0.0)) and (table_cited or not case.get("requires_table_citation"))
        else:
            passed = answer_key_match(result.answer, case) and float(retrieval.get("source_recall@5", 1.0)) >= 1.0
            expected_route = case.get("expected_route")
            expected_task = case.get("expected_task")
            expected_strategy = case.get("expected_strategy")
            if expected_route:
                passed = passed and plan.get("route") == expected_route
            if expected_task:
                passed = passed and plan.get("task_type") == expected_task
            if expected_strategy:
                passed = passed and plan.get("retrieval_strategy") == expected_strategy
        row.update({
            "evaluation_mode": cfg.mode if kind not in {"sql", "planner", "security"} else None,
            "evaluation_profile": cfg.profile if kind not in {"sql", "planner", "security"} else None,
            "route": plan.get("route"),
            "task": plan.get("task_type"),
            "strategy": plan.get("retrieval_strategy"),
            "answer_key_match": answer_key_match(result.answer, case) if kind == "qa" else None,
            "missing_answer_match": missing_answer_match(result.answer, case) if kind == "missing" else None,
            "grounded_absence": bool(result.trace.get("metrics", {}).get("grounded_absence", False)) if kind == "missing" else None,
            "citation_validity": round(float(citations["citation_validity"]), 3),
            "citation_coverage": round(float(citations["citation_coverage"]), 3),
            "source_recall@5": round(float(retrieval.get("source_recall@5", 1.0)), 3),
            "latency_ms": round(max(0.0, wall - pace), 1),
            "pass": passed,
            "gemini_calls": int(result.trace.get("metrics", {}).get("llm_calls_estimate", 0) or 0),
            "_answer": result.answer,
            "_sources": result.sources,
            "_node_times": _trace_node_times(result.trace, pace),
        })
        rows.append(row)
    return rows


def _profile_benchmark(
    workspace: Workspace,
    cases: list[dict[str, Any]],
    api_key: str | None,
    model: str,
    request_pacer: RequestPacer,
    progress: Callable[[float, str], None],
    start: float,
    span: float,
) -> list[dict[str, Any]]:
    selected = [cases[0]] if cases else []
    cross = next((case for case in cases if len(case.get("relevant_sources", [])) > 1), None)
    if cross and cross not in selected:
        selected.append(cross)
    elif len(cases) > 1:
        selected.append(cases[1])
    rows: list[dict[str, Any]] = []
    combinations = [(profile, case) for profile in ("Fast", "Balanced", "Agentic") for case in selected]
    total = max(1, len(combinations))
    for idx, (profile, case) in enumerate(combinations, start=1):
        progress(start + span * (idx-1)/total, f"Profile benchmark {idx}/{len(combinations)}")
        cfg = PipelineConfig(mode="Documents", profile=profile, model=model, allow_web_fallback=False, use_crag=True, use_self_rag=True)
        engine = RAGEngine(workspace, request_pacer=request_pacer)
        wait_before=request_pacer.total_sleep_seconds(); began=time.perf_counter()
        result=engine.ask(case["question"],cfg,api_key,use_cache=False,record_history=False)
        wall=(time.perf_counter()-began)*1000; pace=max(0.0,request_pacer.total_sleep_seconds()-wait_before)*1000
        citations=citation_metrics(result.answer,result.sources); metrics=result.trace.get("metrics",{})
        rows.append({"profile":profile,"case":case["id"],"answer_key_match":answer_key_match(result.answer,case),"citation_validity":round(float(citations["citation_validity"]),3),"citation_coverage":round(float(citations["citation_coverage"]),3),"latency_ms":round(max(0.0,wall-pace),1),"llm_calls_estimate":int(metrics.get("llm_calls_estimate",0) or 0),"reranker_used":bool(metrics.get("reranker_used",False)),"correction_used":bool(metrics.get("correction_used",False))})
    return rows


def _profile_summary(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    summaries: list[dict[str, Any]] = []
    for profile in ("Fast", "Balanced", "Agentic"):
        group = [row for row in rows if row.get("profile") == profile]
        if not group:
            continue
        summaries.append({
            "profile": profile,
            "answer_accuracy": round(mean([float(bool(row.get("answer_key_match"))) for row in group]), 3),
            "citation_validity": round(mean([float(row.get("citation_validity", 0.0) or 0.0) for row in group]), 3),
            "citation_coverage": round(mean([float(row.get("citation_coverage", 0.0) or 0.0) for row in group]), 3),
            "median_latency_ms": round(percentile([float(row.get("latency_ms", 0.0) or 0.0) for row in group], 0.5), 1),
            "mean_llm_calls": round(mean([float(row.get("llm_calls_estimate", 0.0) or 0.0) for row in group]), 2),
            "reranker_rate": round(mean([float(bool(row.get("reranker_used"))) for row in group]), 3),
            "cases": len(group),
        })
    return summaries


def _profile_recommendation(profile_summary: list[dict[str, Any]]) -> str:
    if not profile_summary:
        return ""
    by_name = {row["profile"]: row for row in profile_summary}
    fast = by_name.get("Fast")
    balanced = by_name.get("Balanced")
    agentic = by_name.get("Agentic")
    if fast and balanced:
        quality_close = (
            float(fast.get("answer_accuracy", 0.0)) >= float(balanced.get("answer_accuracy", 0.0)) - 0.01
            and float(fast.get("citation_coverage", 0.0)) >= float(balanced.get("citation_coverage", 0.0)) - 0.05
        )
        faster = float(fast.get("median_latency_ms", 0.0) or 0.0) < float(balanced.get("median_latency_ms", 0.0) or 0.0)
        if quality_close and faster:
            return "Fast matched Balanced quality on the sampled explicit-Documents cases with lower median latency. Keep Balanced as the general Auto default, but prefer Fast for simple local lookups."
    if agentic and balanced and float(agentic.get("median_latency_ms", 0.0) or 0.0) > 2 * max(1.0, float(balanced.get("median_latency_ms", 0.0) or 0.0)):
        return "Agentic was materially slower than Balanced on the sampled cases. Reserve Agentic for difficult or low-confidence work rather than routine lookups."
    return "Profile differences were not large enough on this sample to justify changing the default execution policy."


def _diagnostics(
    summary: dict[str, Any],
    ablation_rows: list[dict[str, Any]],
    planner_rows: list[dict[str, Any]],
    sql_rows: list[dict[str, Any]],
    hard_rows: list[dict[str, Any]] | None = None,
    profile_summary: list[dict[str, Any]] | None = None,
    node_latency_rows: list[dict[str, Any]] | None = None,
    context_budget_rows: list[dict[str, Any]] | None = None,
    compression_rows: list[dict[str, Any]] | None = None,
    scale_stress_rows: list[dict[str, Any]] | None = None,
) -> list[dict[str, str]]:
    findings: list[dict[str, str]] = []

    if int(summary.get("rate_limit_retries", 0) or 0) > 0:
        findings.append(
            {
                "severity": "warning",
                "area": "gemini quota",
                "finding": (
                    f"Gemini surfaced {int(summary.get('rate_limit_retries', 0))} rate-limit retry event(s); "
                    f"provider-directed retry wait was {float(summary.get('rate_limit_sleep_ms', 0.0)) / 1000:.1f}s."
                ),
                "recommendation": "Keep quota-safe pacing enabled or lower the target RPM below the active project limit.",
            }
        )

    context_budget_rows = context_budget_rows or []
    adaptive_row = next(
        (row for row in context_budget_rows if row.get("configuration") == "Adaptive budget"), {}
    )
    full_row = next((row for row in context_budget_rows if row.get("configuration") == "Full top-k"), {})
    if summary.get("source_recall@5", 0.0) >= 0.95 and summary.get("source_precision@5", 1.0) < 0.60:
        if adaptive_row:
            findings.append({
                "severity": "info",
                "area": "context efficiency",
                "finding": (
                    f"Source Recall@5 is {float(summary.get('source_recall@5', 0.0)):.0%}; adaptive focused context now "
                    f"uses a median target of {float(adaptive_row.get('median_target_chunks', 0.0)):.1f} chunks and reduces "
                    f"context by {float(adaptive_row.get('median_context_reduction_pct', 0.0)):.0f}%."
                ),
                "recommendation": "The runtime already applies adaptive budgeting. Use the compression and scale-stress ablations to decide whether further tightening is safe rather than lowering top-k globally.",
            })
        else:
            findings.append({
                "severity": "info",
                "area": "context efficiency",
                "finding": (
                    f"Source Recall@5 is {float(summary.get('source_recall@5', 0.0)):.0%} while source Precision@5 is "
                    f"{float(summary.get('source_precision@5', 0.0)):.0%}."
                ),
                "recommendation": "Use a focused context budget before reducing global retrieval breadth.",
            })

    if adaptive_row and full_row:
        full_precision = float(full_row.get("source_precision@5", 0.0) or 0.0)
        adaptive_precision = float(adaptive_row.get("source_precision@5", 0.0) or 0.0)
        full_recall = float(full_row.get("source_recall@5", 0.0) or 0.0)
        adaptive_recall = float(adaptive_row.get("source_recall@5", 0.0) or 0.0)
        reduction = float(adaptive_row.get("median_context_reduction_pct", 0.0) or 0.0)
        if adaptive_recall >= full_recall - 1e-9 and reduction >= 25.0:
            findings.append({
                "severity": "ok",
                "area": "adaptive context budget",
                "finding": (
                    f"Adaptive budgeting preserved source Recall@5 at {adaptive_recall:.0%}, changed Precision@5 from "
                    f"{full_precision:.0%} to {adaptive_precision:.0%}, and cut median context by {reduction:.0f}%."
                ),
                "recommendation": "Keep adaptive budgeting enabled; corpus-scale stress now provides the guardrail for future budget changes.",
            })
        elif adaptive_recall < full_recall - 1e-9:
            findings.append({
                "severity": "warning",
                "area": "adaptive context budget",
                "finding": f"Adaptive budgeting reduced source Recall@5 from {full_recall:.0%} to {adaptive_recall:.0%}.",
                "recommendation": "Loosen the adaptive budget before shipping this policy broadly.",
            })

    compression_rows = compression_rows or []
    compressed = next(
        (row for row in compression_rows if row.get("configuration") == "Adaptive + sentence compression"), {}
    )
    if compressed:
        retention = float(compressed.get("answer_signal_retention", 0.0) or 0.0)
        reduction = float(compressed.get("median_additional_reduction_pct", 0.0) or 0.0)
        findings.append({
            "severity": "ok" if retention >= 0.95 else "warning",
            "area": "evidence compression",
            "finding": f"Focused sentence compression retained labeled answer signals in {retention:.0%} of cases while cutting selected-evidence tokens by a median {reduction:.0f}% beyond context budgeting.",
            "recommendation": "Keep compression enabled for focused lookups only." if retention >= 0.95 else "Disable or loosen sentence compression until labeled signal retention returns above 95%.",
        })

    scale_stress_rows = scale_stress_rows or []
    if summary.get("scale_stress_error"):
        findings.append({
            "severity": "warning",
            "area": "scale stress",
            "finding": f"The zero-Gemini scale-stress harness did not complete: {summary.get('scale_stress_error')}",
            "recommendation": "Treat release readiness as incomplete until the local scale-stress harness runs successfully; the main RAG benchmark can still be inspected independently.",
        })
    if scale_stress_rows:
        largest = scale_stress_rows[-1]
        recall = float(largest.get("source_recall@5", 0.0) or 0.0)
        pruned_recall = float(largest.get("adaptive_pruned_recall@5", 0.0) or 0.0)
        findings.append({
            "severity": "ok" if min(recall, pruned_recall) >= 0.95 else "warning",
            "area": "scale stress",
            "finding": (
                f"At {int(largest.get('chunks', 0) or 0)} chunks / {int(largest.get('sources', 0) or 0)} sources, "
                f"retrieval Recall@5 was {recall:.0%} and adaptive-pruned recall was {pruned_recall:.0%}."
            ),
            "recommendation": "Treat this as synthetic distractor evidence, then repeat with a real larger upload before changing the reranker policy." if min(recall, pruned_recall) >= 0.95 else "Increase retrieval depth or budget targets for large corpora before relying on the adaptive policy.",
        })

    if summary.get("citation_coverage", 1.0) < 0.90:
        findings.append(
            {
                "severity": "warning",
                "area": "citations",
                "finding": f"Citation coverage is {float(summary['citation_coverage']):.0%}; some factual statements are uncited.",
                "recommendation": "Keep the generation prompt citation requirement and inspect low-coverage cases individually.",
            }
        )
    if summary.get("planner_task_accuracy", 1.0) < 0.90:
        failed = [row["id"] for row in planner_rows if not row.get("task_correct")]
        findings.append(
            {
                "severity": "warning",
                "area": "planner",
                "finding": f"Task classification misses: {', '.join(failed) or 'none'}.",
                "recommendation": "Review task taxonomy labels separately from route/strategy correctness; do not over-penalize equivalent plans.",
            }
        )
    if summary.get("text2sql_pass_rate", 1.0) < 0.90:
        failed = [row["id"] for row in sql_rows if not row.get("answer_key_match")]
        findings.append(
            {
                "severity": "warning",
                "area": "text2sql",
                "finding": f"Text2SQL failed cases: {', '.join(failed) or 'none'}.",
                "recommendation": (
                    "Inspect generated SQL, typed observed values and benchmark expectations. SQL routing is evaluated "
                    "separately in the planner suite, so a component failure should not automatically be blamed on routing."
                ),
            }
        )

    hard_rows = hard_rows or []
    if hard_rows and summary.get("hard_mode_pass_rate", 1.0) < 0.90:
        failed = [row.get("id", "?") for row in hard_rows if not row.get("pass")]
        findings.append({
            "severity": "warning",
            "area": "hard_mode",
            "finding": f"Hard-mode robustness failures: {', '.join(failed) or 'none'}.",
            "recommendation": "Inspect missing-answer, distractor, analytical and adversarial cases before expanding the feature set.",
        })
    if len(ablation_rows) == 2:
        base, rerank = ablation_rows
        base_ms = float(base.get("median_retrieval_ms", 0.0) or 0.0)
        rerank_ms = float(rerank.get("median_retrieval_ms", 0.0) or 0.0)
        base_mrr = float(base.get("source_mrr", 0.0) or 0.0)
        rerank_mrr = float(rerank.get("source_mrr", 0.0) or 0.0)
        multiplier = safe_div(rerank_ms, base_ms) if base_ms else 0.0
        if multiplier >= 3.0 and rerank_mrr <= base_mrr + 0.01:
            findings.append(
                {
                    "severity": "info",
                    "area": "reranker",
                    "finding": f"The reranker ablation is {multiplier:.1f}x slower on the demo benchmark with no material source-MRR gain.",
                    "recommendation": "The runtime already skips reranking on small corpora. Keep this ablation as evidence and re-enable the cross-encoder only when a larger-corpus benchmark shows source- or chunk-level gain.",
                }
            )
    profile_summary = profile_summary or []
    recommendation = _profile_recommendation(profile_summary)
    if recommendation:
        findings.append({
            "severity": "info",
            "area": "profile policy",
            "finding": recommendation,
            "recommendation": "Use the profile benchmark as local evidence only; repeat it on larger user corpora before making a global policy claim.",
        })

    node_latency_rows = node_latency_rows or []
    if node_latency_rows:
        dominant = max(node_latency_rows, key=lambda row: float(row.get("mean_ms", 0.0) or 0.0))
        total_mean = sum(float(row.get("mean_ms", 0.0) or 0.0) for row in node_latency_rows)
        share = safe_div(float(dominant.get("mean_ms", 0.0) or 0.0), total_mean)
        if share >= 0.60:
            findings.append({
                "severity": "info",
                "area": "latency",
                "finding": f"{dominant.get('node', 'generation')} dominates mean node time at approximately {share:.0%} of measured pipeline-node latency.",
                "recommendation": "Prioritize model/generation efficiency before micro-optimizing millisecond-scale retrieval stages.",
            })

    if not findings:
        findings.append(
            {
                "severity": "ok",
                "area": "benchmark",
                "finding": "No configured quality gate produced a diagnostic warning.",
                "recommendation": "Expand the benchmark before treating this as general performance evidence.",
            }
        )
    return findings


def _deep_from_standard_cache(
    base_report: dict[str, Any],
    api_key: str | None,
    model: str,
    *,
    target_rpm: int,
    progress: Callable[[float, str], None],
) -> dict[str, Any]:
    """Upgrade a cached Standard run to Deep with judge calls only.

    The deterministic suites are identical between Standard and Deep. Reusing a
    current Standard baseline avoids spending ~25 repeated Gemini calls merely
    to regenerate metrics the user already computed. Deep then adds the sampled
    calibrated judge layer on top of those exact answers/evidence artifacts.
    """
    started = time.perf_counter()
    report = json.loads(json.dumps(base_report))
    benchmark = _load_benchmark()
    cases = {
        case["id"]: case
        for case in benchmark.get("qa_cases", []) + benchmark.get("overview_cases", [])
        if case.get("deep_judge")
    }
    request_pacer = RequestPacer(target_rpm=max(0, int(target_rpm)))
    judge = GeminiGateway(api_key, model, request_pacer=request_pacer)
    judge_rows: list[dict[str, Any]] = []
    candidate_rows: list[dict[str, Any]] = []
    for section in ("focused_qa", "corpus_overviews"):
        for row in report.get(section, []):
            if row.get("id") in cases and row.get("_answer") is not None and row.get("_sources") is not None:
                candidate_rows.append(row)

    total = max(1, len(candidate_rows))
    for idx, row in enumerate(candidate_rows, start=1):
        progress(0.08 + 0.84 * (idx - 1) / total, f"Deep judge case {idx}/{len(candidate_rows)}")
        case = cases[row["id"]]
        citations = row.get("_citations") or citation_metrics(row.get("_answer", ""), row.get("_sources", []))
        row.update(_judge_row(judge, case, row.get("_answer", ""), row.get("_sources", []), citations))
        judge_rows.append(row)

    summary = report.setdefault("summary", {})
    baseline_wall_ms = float(summary.get("evaluation_wall_ms", 0.0) or 0.0)
    if judge_rows:
        summary.update(
            {
                "judge_faithfulness": round(mean([float(row["judge_faithfulness"]) for row in judge_rows]), 3),
                "judge_answer_relevance": round(mean([float(row["judge_answer_relevance"]) for row in judge_rows]), 3),
                "judge_completeness": round(mean([float(row["judge_completeness"]) for row in judge_rows]), 3),
                "judge_citation_support": round(mean([float(row["judge_citation_support"]) for row in judge_rows]), 3),
                "judge_overall": round(mean([float(row["judge_overall"]) for row in judge_rows]), 3),
                "judge_pass_rate": round(mean([float(row["judge_pass"]) for row in judge_rows]), 3),
                "judge_latency_mean_ms": round(mean([float(row.get("judge_latency_ms", 0.0)) for row in judge_rows]), 3),
            }
        )
    pacing_stats = request_pacer.stats()
    summary.update(
        {
            "evaluation_level": "Deep",
            "evaluation_wall_ms": round((time.perf_counter() - started) * 1000, 1),
            "evaluation_target_rpm": int(pacing_stats["target_rpm"]),
            "gemini_requests": int(pacing_stats["gemini_requests"]),
            "pacing_sleep_ms": float(pacing_stats["pacing_sleep_ms"]),
            "rate_limit_retries": int(pacing_stats["rate_limit_retries"]),
            "rate_limit_sleep_ms": float(pacing_stats["rate_limit_sleep_ms"]),
            "deep_judge_cases": len(judge_rows),
            "reused_standard_baseline": True,
            "deep_incremental": True,
            "deterministic_baseline_wall_ms": round(baseline_wall_ms, 1),
        }
    )
    report["diagnostics"] = _diagnostics(
        summary,
        report.get("retrieval_ablation", []),
        report.get("semantic_planner", []),
        report.get("text2sql", []),
        report.get("hard_mode", []),
        report.get("profile_summary", []),
        report.get("node_latency", []),
        report.get("context_budget_ablation", []),
        report.get("evidence_compression_ablation", []),
        report.get("scale_stress", []),
    )
    report.setdefault("methodology", {})["evaluation_cache"] = (
        "Deep reused the current cached Standard deterministic baseline and issued only sampled judge calls."
    )
    progress(1.0, "Deep evaluation complete")
    return report


def run_demo_eval(
    workspace: Workspace,
    api_key: str | None,
    model: str,
    level: str = "Standard",
    progress_callback: Callable[[float, str], None] | None = None,
    target_rpm: int = 12,
    base_standard_report: dict[str, Any] | None = None,
    include_profile_benchmark: bool = False,
) -> dict[str, Any]:
    """Run the bundled benchmark with response caching disabled.

    Quick: smaller deterministic regression set.
    Standard: full deterministic set + retrieval and context-budget ablations.
    Deep: Standard plus calibrated Gemini LLM-as-judge scores.
    """
    wall_started = time.perf_counter()
    progress = progress_callback or (lambda _value, _message: None)
    benchmark = _load_benchmark()
    level = level if level in {"Quick", "Standard", "Deep"} else "Standard"
    if level == "Deep" and base_standard_report and not include_profile_benchmark:
        artifact_rows = base_standard_report.get("focused_qa", []) + base_standard_report.get("corpus_overviews", [])
        if any(row.get("_answer") is not None and row.get("_sources") is not None for row in artifact_rows):
            return _deep_from_standard_cache(
                base_standard_report,
                api_key,
                model,
                target_rpm=target_rpm,
                progress=progress,
            )
    deep_judge = level == "Deep"
    request_pacer = RequestPacer(target_rpm=max(0, int(target_rpm)))

    qa_cases = benchmark["qa_cases"] if level != "Quick" else benchmark["qa_cases"][:3]
    planner_cases = benchmark["planner_cases"] if level != "Quick" else benchmark["planner_cases"][:5]
    overview_cases = benchmark["overview_cases"] if level != "Quick" else benchmark["overview_cases"][:1]
    sql_cases = benchmark.get("sql_cases", []) if level != "Quick" else benchmark.get("sql_cases", [])[:1]
    hard_cases = benchmark.get("hard_mode_cases", []) if level != "Quick" else benchmark.get("hard_mode_cases", [])[:2]

    progress(0.01, "Preparing evaluation")
    qa_rows = _qa_eval(
        workspace, qa_cases, api_key, model, deep_judge, request_pacer, progress, 0.03, 0.29
    )
    gateway = GeminiGateway(api_key, model, request_pacer=request_pacer)
    planner_rows = _planner_eval(workspace, planner_cases, gateway, progress, 0.34, 0.22)
    overview_rows = _overview_eval(
        workspace, overview_cases, api_key, model, deep_judge, request_pacer, progress, 0.58, 0.16
    )
    sql_rows = (
        _sql_eval(workspace, sql_cases, api_key, model, request_pacer, progress, 0.75, 0.08)
        if sql_cases
        else []
    )
    hard_rows = _hard_mode_eval(
        workspace, hard_cases, api_key, model, request_pacer, progress, 0.83, 0.10
    ) if hard_cases else []
    profile_rows = _profile_benchmark(
        workspace, qa_cases, api_key, model, request_pacer, progress, 0.93, 0.05
    ) if include_profile_benchmark and level != "Quick" else []
    progress(0.98, "Checking abstention, retrieval, context and scale ablations")
    abstention_rows = _abstention_eval()
    ablation_rows = _retrieval_ablation(workspace, qa_cases) if level != "Quick" else []
    context_budget_rows = _context_budget_ablation(workspace, qa_cases) if level != "Quick" else []
    compression_rows = _evidence_compression_ablation(workspace, qa_cases) if level != "Quick" else []
    scale_stress_error = ""
    if level != "Quick":
        try:
            scale_stress_rows = scale_stress_retrieval_eval(workspace, qa_cases)
        except Exception as exc:  # keep the primary benchmark available even if the local stress harness fails
            scale_stress_rows = []
            scale_stress_error = f"{type(exc).__name__}: {exc}"
    else:
        scale_stress_rows = []

    planner_metrics = _planner_summary(planner_rows)
    qa_latencies = [float(row["latency_ms"]) for row in qa_rows]
    overview_latencies = [float(row["latency_ms"]) for row in overview_rows]
    sql_latencies = [float(row["latency_ms"]) for row in sql_rows]
    planner_latencies = [float(row["latency_ms"]) for row in planner_rows]
    all_latencies = qa_latencies + overview_latencies + sql_latencies
    all_runtime_rows = qa_rows + overview_rows + sql_rows + [row for row in hard_rows if row.get("latency_ms") is not None]

    answer_accuracy = mean([float(row["answer_key_match"]) for row in qa_rows])
    source_recall = mean([float(row["source_recall@5"]) for row in qa_rows])
    source_mrr = mean([float(row["source_mrr"]) for row in qa_rows])
    citation_validity = mean([float(row["citation_validity"]) for row in qa_rows + overview_rows])
    citation_coverage = mean([float(row["citation_coverage"]) for row in qa_rows + overview_rows])
    overview_pass = mean([float(row["pass"]) for row in overview_rows])
    abstention_accuracy = mean([float(row["pass"]) for row in abstention_rows])
    sql_accuracy = mean([float(row["answer_key_match"]) for row in sql_rows]) if sql_rows else 1.0

    hard_accuracy = mean([float(row.get("pass", False)) for row in hard_rows]) if hard_rows else 1.0

    deterministic_score = (
        0.19 * answer_accuracy
        + 0.12 * source_recall
        + 0.06 * source_mrr
        + 0.08 * citation_validity
        + 0.07 * citation_coverage
        + 0.11 * planner_metrics["planner_route_accuracy"]
        + 0.07 * planner_metrics["planner_task_accuracy"]
        + 0.07 * planner_metrics["planner_strategy_accuracy"]
        + 0.07 * planner_metrics["web_use_precision"]
        + 0.04 * overview_pass
        + 0.04 * abstention_accuracy
        + 0.03 * sql_accuracy
        + 0.05 * hard_accuracy
    )

    judge_rows = [row for row in qa_rows + overview_rows if "judge_overall" in row]
    judge_summary: dict[str, float] = {}
    if judge_rows:
        judge_summary = {
            "judge_faithfulness": mean([float(row["judge_faithfulness"]) for row in judge_rows]),
            "judge_answer_relevance": mean([float(row["judge_answer_relevance"]) for row in judge_rows]),
            "judge_completeness": mean([float(row["judge_completeness"]) for row in judge_rows]),
            "judge_citation_support": mean([float(row["judge_citation_support"]) for row in judge_rows]),
            "judge_overall": mean([float(row["judge_overall"]) for row in judge_rows]),
            "judge_pass_rate": mean([float(row["judge_pass"]) for row in judge_rows]),
            "judge_latency_mean_ms": mean([float(row.get("judge_latency_ms", 0.0)) for row in judge_rows]),
        }

    pacing_stats = request_pacer.stats()

    metrics_for_gate = {
        "planner_route_accuracy": planner_metrics["planner_route_accuracy"],
        "web_use_precision": planner_metrics["web_use_precision"],
        "planner_task_accuracy": planner_metrics["planner_task_accuracy"],
        "citation_validity": citation_validity,
        "citation_coverage": citation_coverage,
        "text2sql_pass_rate": sql_accuracy,
        "hard_mode_pass_rate": hard_accuracy,
    }
    grade, quality_gates = _grade_with_gates(deterministic_score, metrics_for_gate)

    adaptive_budget_row = next(
        (row for row in context_budget_rows if row.get("configuration") == "Adaptive budget"), {}
    )
    compression_row = next(
        (row for row in compression_rows if row.get("configuration") == "Adaptive + sentence compression"), {}
    )
    largest_scale_row = scale_stress_rows[-1] if scale_stress_rows else {}

    summary: dict[str, Any] = {
        "benchmark_version": benchmark.get("version"),
        "evaluation_level": level,
        "deterministic_quality_score": round(deterministic_score, 3),
        "quality_grade": grade,
        "quality_gate_notes": quality_gates,
        "answer_accuracy": round(answer_accuracy, 3),
        "source_precision@5": round(mean([float(row["source_precision@5"]) for row in qa_rows]), 3),
        "source_recall@5": round(source_recall, 3),
        "source_hit@1": round(mean([float(row["source_hit@1"]) for row in qa_rows]), 3),
        "source_mrr": round(source_mrr, 3),
        "source_ap@5": round(mean([float(row["source_ap@5"]) for row in qa_rows]), 3),
        "source_ndcg@5": round(mean([float(row["source_ndcg@5"]) for row in qa_rows]), 3),
        "source_duplicate_rate@5": round(mean([float(row["source_duplicate_rate@5"]) for row in qa_rows]), 3),
        "context_pruning_precision@5": round(float(adaptive_budget_row.get("source_precision@5", 0.0)), 3),
        "context_pruning_recall@5": round(float(adaptive_budget_row.get("source_recall@5", 0.0)), 3),
        "context_pruning_token_reduction_pct": round(float(adaptive_budget_row.get("median_context_reduction_pct", 0.0)), 1),
        "adaptive_context_target_p50": round(float(adaptive_budget_row.get("median_target_chunks", 0.0)), 1),
        "compression_signal_retention": round(float(compression_row.get("answer_signal_retention", 0.0)), 3),
        "compression_additional_reduction_pct": round(float(compression_row.get("median_additional_reduction_pct", 0.0)), 1),
        "scale_stress_max_chunks": int(largest_scale_row.get("chunks", 0) or 0),
        "scale_stress_recall@5": round(float(largest_scale_row.get("source_recall@5", 0.0)), 3) if largest_scale_row else 0.0,
        "scale_stress_pruned_recall@5": round(float(largest_scale_row.get("adaptive_pruned_recall@5", 0.0)), 3) if largest_scale_row else 0.0,
        "citation_validity": round(citation_validity, 3),
        "citation_coverage": round(citation_coverage, 3),
        **{key: round(value, 3) for key, value in planner_metrics.items()},
        "overview_pass_rate": round(overview_pass, 3),
        "abstention_accuracy": round(abstention_accuracy, 3),
        "text2sql_pass_rate": round(sql_accuracy, 3),
        "hard_mode_pass_rate": round(hard_accuracy, 3),
        "profile_benchmark_enabled": bool(profile_rows),
        "profile_benchmark_cases": len(profile_rows),
        "latency_p50_ms": round(percentile(all_latencies, 0.50), 1),
        "latency_p95_ms": round(percentile(all_latencies, 0.95), 1),
        "planner_latency_p50_ms": round(percentile(planner_latencies, 0.50), 1),
        "planner_latency_p95_ms": round(percentile(planner_latencies, 0.95), 1),
        "mean_llm_calls_estimate": round(
            mean([float(row.get("llm_calls_estimate", 0)) for row in all_runtime_rows]), 2
        ),
        "focused_context_tokens_before_p50": round(percentile([float(row.get("context_tokens_est_before", 0)) for row in qa_rows], 0.50), 1),
        "focused_context_tokens_after_p50": round(percentile([float(row.get("context_tokens_est_after", 0)) for row in qa_rows], 0.50), 1),
        "focused_context_pruning_rate": round(mean([float(bool(row.get("context_pruning_used", False))) for row in qa_rows]), 3),
        "focused_generation_prompt_tokens_p50": round(percentile([float(row.get("generation_prompt_tokens_est", 0)) for row in qa_rows], 0.50), 1),
        "focused_generation_total_tokens_p50": round(percentile([float(row.get("generation_total_tokens_est", 0)) for row in qa_rows], 0.50), 1),
        "focused_evidence_utilization_p50": round(percentile([float(row.get("evidence_source_utilization_rate", 0.0)) for row in qa_rows], 0.50), 3),
        "focused_evidence_compression_rate": round(mean([float(bool(row.get("evidence_compression_used", False))) for row in qa_rows]), 3),
        "focused_evidence_compression_reduction_p50": round(percentile([float(row.get("evidence_compression_reduction_pct", 0.0)) for row in qa_rows], 0.50), 1),
        "correction_rate": round(mean([float(row.get("correction_used", False)) for row in all_runtime_rows]), 3),
        "runtime_web_use_rate": round(mean([float(row.get("web_used", False)) for row in all_runtime_rows]), 3),
        "cache_bypassed": True,
        "evaluation_wall_ms": 0.0,
        "evaluation_target_rpm": int(pacing_stats["target_rpm"]),
        "gemini_requests": int(pacing_stats["gemini_requests"]),
        "pacing_sleep_ms": float(pacing_stats["pacing_sleep_ms"]),
        "rate_limit_retries": int(pacing_stats["rate_limit_retries"]),
        "rate_limit_sleep_ms": float(pacing_stats["rate_limit_sleep_ms"]),
        "deep_judge_cases": len(judge_rows),
        "scale_stress_error": scale_stress_error,
        **{key: round(value, 3) for key, value in judge_summary.items()},
    }

    node_latency_rows = _node_latency_summary(qa_rows + overview_rows + hard_rows)
    profile_summary_rows = _profile_summary(profile_rows)
    profile_recommendation = _profile_recommendation(profile_summary_rows)
    if profile_recommendation:
        summary["profile_recommendation"] = profile_recommendation
    if level == "Quick":
        readiness_rows = []
        summary["release_readiness"] = "NOT RUN"
        summary["release_readiness_score"] = None
    else:
        readiness_rows = _readiness_rows(summary, scale_stress_rows, compression_rows)
        readiness_status, readiness_score = _readiness_summary(readiness_rows)
        summary["release_readiness"] = readiness_status
        summary["release_readiness_score"] = round(readiness_score, 3)
    diagnostics = _diagnostics(
        summary, ablation_rows, planner_rows, sql_rows, hard_rows, profile_summary_rows, node_latency_rows,
        context_budget_rows, compression_rows, scale_stress_rows
    )
    summary["evaluation_wall_ms"] = round((time.perf_counter() - wall_started) * 1000, 1)
    progress(1.0, "Evaluation complete")
    return {
        "summary": summary,
        "diagnostics": diagnostics,
        "focused_qa": qa_rows,
        "semantic_planner": planner_rows,
        "corpus_overviews": overview_rows,
        "text2sql": sql_rows,
        "abstention": abstention_rows,
        "retrieval_ablation": ablation_rows,
        "context_budget_ablation": context_budget_rows,
        "evidence_compression_ablation": compression_rows,
        "scale_stress": scale_stress_rows,
        "release_readiness": readiness_rows,
        "hard_mode": hard_rows,
        "profile_benchmark": profile_rows,
        "profile_summary": profile_summary_rows,
        "node_latency": node_latency_rows,
        "methodology": {
            "deterministic": (
                "Transparent labels for answer terms, relevant source files, route/task/strategy, web-use policy, "
                "citation validity/coverage, abstention and latency. Source metrics deduplicate repeated chunks from "
                "the same file before source-level AP/MRR/nDCG are computed."
            ),
            "latency": (
                "Evaluation bypasses the response cache and does not mutate chat history, so reported pipeline latency "
                "reflects real benchmark execution rather than cached answers. Node-latency summaries subtract deliberate "
                "quota pacing proportionally across nodes that issued model calls; raw traces retain wall-clock node time."
            ),
            "deep_judge": (
                "Optional Gemini judge for a representative labeled subset of benchmark cases, covering focused QA, "
                "NIST, cross-document synthesis and corpus overview. Sampling reduces free-tier request pressure while "
                "citation-support and overall scores remain calibrated against deterministic citation validity/coverage. "
                "When a compatible Standard report is supplied, Deep reuses that deterministic baseline and only runs "
                "the sampled judge layer."
            ),
            "quota_safety": (
                f"All Gemini calls in this run share a rolling request pacer targeting {int(pacing_stats['target_rpm'])} RPM. "
                "The pacer also accounts for recent interactive requests recorded by this process and surfaced 429s "
                "honor provider retry guidance before a bounded retry."
            ),
            "text2sql": (
                "Text2SQL routing is evaluated in the semantic-planner suite. The Text2SQL component suite uses one "
                "model call per case to generate validated read-only SQL, executes it in DuckDB and checks labeled scalar "
                "outputs as typed boolean/numeric/text values when available."
            ),
            "evaluation_cache": (
                "Completed reports can be saved by the workspace with corpus/model/benchmark metadata. Saved evaluation "
                "history is separate from the RAG response cache and can be reused without rerunning the benchmark."
            ),
            "quality_gates": (
                "The letter grade is capped when a critical subsystem is weak, preventing a high weighted average "
                "from hiding poor Text2SQL, routing or citation performance."
            ),
            "hard_mode": "Hard-mode cases cover paraphrase, distractors, missing answers, multi-hop comparison, analytical synthesis, structured filtering, local freshness semantics and prompt-injection detection.",
            "profile_benchmark": "Optional Fast/Balanced/Agentic comparison uses a small labeled subset because it intentionally spends additional Gemini requests.",
            "chunk_ablation": "Retrieval ablation reports source-level metrics plus chunk Hit@1/MRR for cases with explicit chunk-content labels.",
            "context_budget": "Focused-query pruning is evaluated as a zero-Gemini ablation across full top-k, a fixed 3-chunk budget, and the adaptive budget chosen from retrieval confidence, score separation and corpus scale.",
            "evidence_compression": "Focused evidence compression selects query-relevant sentences after context budgeting and is evaluated by deterministic answer-signal retention plus token reduction; it never spends a Gemini request.",
            "scale_stress": "Scale stress reuses existing embedding vectors and clones long-document distractor chunks to exercise the real Qdrant + BM25 path at roughly 1x, 5x and 20x distractor scale without additional Gemini calls.",
            "release_readiness": "A transparent readiness checklist applies explicit thresholds to answer quality, grounding, routing, robustness, adaptive-budget recall, compression retention and largest-scale retrieval recall. It does not replace the underlying metrics.",
            "benchmark_file": "evals/demo_benchmark.json",
        },
    }