Video-Tree-TRM5/tests/integration/test_build_split_e2e.py

"""build_split 端到端集成测试：真实 infer_adhoc 预测 + 构造诊断信号 → 冻结三池。

用 workspaces/default/harness.db 的真实 infer_adhoc 基线预测（900 题 / 300 视频，
每视频 3 题）驱动结果驱动视频级切分，构造一份 T2 诊断信号 store 喂给选择器，
验证冻结的 pools.json 满足全部防御断言（三池视频互斥、覆盖 900 题、每视频 3 题、
内容指纹一致），确保 capstone 编排把前面所有组件正确串联。
"""

from __future__ import annotations

import hashlib
import sqlite3
from pathlib import Path

import pytest

from adapters.baseline_diagnosis_store import SqliteDiagnosisSignalStore
from app.harness.build_split import SplitBuildConfig, build_split
from app.harness.split_selection import evolution_target_of
from core.evolution.types import DiagnosisSignalRow

_HARNESS_DB = Path("workspaces/default/harness.db")
_BENCHMARK_DIR = Path("store/questions/benchmarks/Video-MME")
_BASELINE_RUN_ID = "infer_adhoc"
_DIAG_FINGERPRINT = "diag_test_v1"
# 四类错误类别轮转，确定性铺满多样性格子（cell = task_type × error_type）。
_ERROR_TYPES = ("extraction_failure", "search_failure", "reasoning_failure", "mixed")


def _normalize(choice: str | None) -> str:
    """选项归一：strip → 大写 → 取首字母（None 归一为空串）。"""
    return (choice or "").strip().upper()[:1]


def _read_canonical_predictions() -> list[dict]:
    """从真实 harness.db 读 infer_adhoc 每题首行（ORDER BY rowid）作为 canonical 预测。"""
    conn = sqlite3.connect(f"file:{_HARNESS_DB}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
    rows = conn.execute(
        "SELECT question_id, video_id, task_type, prediction, answer "
        "FROM predictions WHERE run_id = ? ORDER BY rowid",
        (_BASELINE_RUN_ID,),
    ).fetchall()
    conn.close()
    seen: dict[str, dict] = {}
    for row in rows:
        qid = row["question_id"]
        if qid in seen:
            continue
        seen[qid] = {
            "question_id": qid,
            "video_id": row["video_id"],
            "task_type": row["task_type"],
            "correct": _normalize(row["prediction"]) == _normalize(row["answer"]),
        }
    return list(seen.values())


def _populate_signal_store(store: SqliteDiagnosisSignalStore, preds: list[dict]) -> None:
    """为全部错题写入 T2 诊断信号（error_type 轮转），构造可训练缺陷多样性。"""
    wrong = [p for p in preds if not p["correct"]]
    for idx, pred in enumerate(wrong):
        error_type = _ERROR_TYPES[idx % len(_ERROR_TYPES)]
        store.upsert(
            DiagnosisSignalRow(
                question_id=pred["question_id"],
                video_id=pred["video_id"],
                baseline_run_id=_BASELINE_RUN_ID,
                diag_fingerprint=_DIAG_FINGERPRINT,
                task_type=pred["task_type"],
                error_type=error_type,
                cause_category="defect",
                tier="T2",
                evolution_target=evolution_target_of(error_type),
                degraded=False,
                infra=False,
                session_id=None,
            )
        )


@pytest.mark.skipif(
    not _HARNESS_DB.exists() or not _BENCHMARK_DIR.exists(),
    reason="需要真实 workspaces/default/harness.db 与 Video-MME benchmark",
)
def test_end_to_end_freezes_valid_pools(tmp_path: Path) -> None:
    """真实基线预测 + 构造 T2 信号 → 冻结 pools.json，校验六条防御断言。"""
    out = tmp_path / "pools.json"
    manifest_path = tmp_path / "split_manifest.json"
    signal_db = tmp_path / "signals.db"

    preds = _read_canonical_predictions()
    store = SqliteDiagnosisSignalStore(str(signal_db))
    _populate_signal_store(store, preds)

    config = SplitBuildConfig(
        n_trainval=100,
        floor_k={"Object Reasoning": 3},
        epsilon=0.1,
        report_floor=30,
        select_seed=7,
        val_ratio=0.3,
        split_seed=7,
    )

    result = build_split(
        db_path=_HARNESS_DB,
        baseline_run_id=_BASELINE_RUN_ID,
        signal_store=store,
        diag_fingerprint=_DIAG_FINGERPRINT,
        questions_dir=_BENCHMARK_DIR,
        config=config,
        out_path=out,
        manifest_path=manifest_path,
        generated_at="2026-07-15T00:00:00Z",
    )
    store.close()

    pools = result["pools"]

    # 防御① 三池视频集两两不相交
    diag_v = {q.video_id for q in pools.diagnosis}
    val_v = {q.video_id for q in pools.validation}
    test_v = {q.video_id for q in pools.test}
    assert diag_v & test_v == set()
    assert val_v & test_v == set()
    assert diag_v & val_v == set()

    # 防御② 三池覆盖 900 题（按 distinct question）
    all_q = pools.diagnosis + pools.validation + pools.test
    assert len({q.question_id for q in all_q}) == 900
    assert len(all_q) == 900

    # 防御③ 每 video 恰 3 题
    per_video: dict[str, int] = {}
    for q in all_q:
        per_video[q.video_id] = per_video.get(q.video_id, 0) + 1
    assert set(per_video.values()) == {3}

    # 防御⑤ manifest 的 pools_sha256 == sha256(out 文件内容)
    assert (
        result["manifest"]["pools_sha256"]
        == hashlib.sha256(out.read_text(encoding="utf-8").encode("utf-8")).hexdigest()
    )

    # assignment 覆盖全部 300 视频且取值合法
    assignment = result["assignment"]
    assert set(assignment.values()) <= {"trainval", "test"}
    assert len(assignment) == 300

    # manifest 覆盖报告含关键指标
    coverage = result["manifest"]["coverage_report"]
    assert coverage["grid_total"] == 48
    assert "tier_distribution" in coverage
    assert manifest_path.exists()

    # I-1：evolution_target_distribution 出现在 coverage_report，且 T2 信号按 tool/skill/system 计数。
    target_dist = coverage["evolution_target_distribution"]
    assert set(target_dist) <= {"tool", "skill", "system"}
    # 全部错题构造为 T2（error_type 四类轮转），进化目标覆盖三层且计数为正。
    assert sum(target_dist.values()) > 0
    assert set(target_dist) == {"tool", "skill", "system"}