feat: wire end-to-end results-driven split with defensive asserts

app/harness/build_split.py

@@ -0,0 +1,504 @@
+"""结果驱动视频级切分的顶层编排：诊断信号 → 冻结 pools.json + manifest。
+
+把已实现的组件串成 capstone 管线：从 harness.db 读 canonical 基线预测、从
+DiagnosisSignalStore 读逐题诊断信号，构建全视频画像、贪心联合约束选择 trainval /
+test，再以视频组为原子切出诊断 / 验证池，原子冻结 pools.json 并写溯源 manifest。
+全程带六条防御断言（P5，任一不满足即 fail-fast，绝不静默兜底）。
+
+只有基线推理与诊断是上游产物；本模块纯 code-controlled，不发起任何 LLM 调用，
+读预测走只读连接，不改动 harness.db。
+"""
+
+from __future__ import annotations
+
+import hashlib
+import sqlite3
+from collections import Counter
+from dataclasses import asdict, dataclass, field
+from typing import TYPE_CHECKING
+
+from loguru import logger
+
+from app.harness.pools import save_pools, split_by_video_assignment
+from app.harness.split_manifest import write_manifest
+from app.harness.split_selection import (
+    SelectConfig,
+    build_video_records,
+    derive_reportable_types,
+    select_split,
+)
+from app.question_gen.loader import load_benchmark
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from app.harness.pools import Pools
+    from app.harness.split_selection import SplitAssignment, VideoRecord
+    from core.evolution.protocols import DiagnosisSignalStore
+    from core.evolution.types import DiagnosisSignalRow
+    from core.types import GeneratedQuestion
+
+# 多样性主格子 = 12 题型 × 4 错误类别 = 48 格，覆盖报告以此为分母。
+_DIVERSITY_GRID_TOTAL = 48
+_QUESTIONS_PER_VIDEO = 3
+
+
+@dataclass(frozen=True)
+class SplitBuildConfig:
+    """结果驱动切分的旋钮快照（科研配置，随实验扫动）。
+
+    承载贪心选择器与视频组题级切分的全部可扫参数；asdict 后直接写入 manifest 的
+    config 快照，保证复现时可比对。floor_k 为不可哈希容器，标 hash=False 排除出
+    自动 __hash__，避免 frozen dataclass 被哈希时报错（本类不作字典键，仅承载配置）。
+
+    字段:
+        n_trainval: trainval 目标视频数（多样性阶段填充上限）。
+        floor_k: 各高信号 task_type 的 T2 defect 下限（select_split 硬约束）。
+        epsilon: test 相对全局的最大允许分布偏差（题型 / 难度两维）。
+        report_floor: per-type 报告门限，题数 ≥ 此值的 task_type 才入 ε 约束。
+        select_seed: 贪心选择器预洗牌种子（打破等增益平局）。
+        val_ratio: validation 占 trainval 视频组总数的比例。
+        split_seed: 视频组题级切分的洗牌种子。
+    """
+
+    n_trainval: int
+    floor_k: dict[str, int] = field(hash=False)
+    epsilon: float
+    report_floor: int
+    select_seed: int
+    val_ratio: float
+    split_seed: int
+
+
+@dataclass(frozen=True)
+class SplitBuildResult:
+    """build_split 的返回结果（冻结产物 + 溯源）。
+
+    字段:
+        pools: 冻结的三池（diagnosis / validation / test）。
+        manifest: 写入 split_manifest.json 的溯源字典（含 pools_sha256）。
+        assignment: video_id -> "trainval" | "test" 归属字典。
+    """
+
+    pools: Pools
+    manifest: dict
+    assignment: dict[str, str]
+
+    def __getitem__(self, key: str) -> object:
+        """兼容字典式访问（result["pools"] / ["manifest"] / ["assignment"]）。
+
+        参数:
+            key: 字段名，取值 pools / manifest / assignment。
+
+        返回:
+            对应字段值。
+
+        异常:
+            KeyError: key 非上述三者之一。
+        """
+        if key not in {"pools", "manifest", "assignment"}:
+            raise KeyError(f"未知字段: {key}")
+        return getattr(self, key)
+
+
+def build_split(
+    *,
+    db_path: Path,
+    baseline_run_id: str,
+    signal_store: DiagnosisSignalStore,
+    diag_fingerprint: str,
+    questions_dir: Path,
+    config: SplitBuildConfig,
+    out_path: Path,
+    manifest_path: Path,
+    generated_at: str,
+) -> SplitBuildResult:
+    """顶层编排结果驱动视频级切分，冻结 pools.json + manifest 并跑防御断言。
+
+    步骤：读 canonical 基线预测 → 读诊断信号 → 构建全视频画像 → 贪心选择 trainval /
+    test → 加载题库并以视频归属切三池 → 原子冻结 pools.json → 写溯源 manifest →
+    六条防御断言 fail-fast 校验。
+
+    参数:
+        db_path: harness.db 路径（只读读取 predictions，不改动）。
+        baseline_run_id: 基线 run 标识（如 "infer_adhoc"）。
+        signal_store: 逐题诊断信号存储端口，读 (run, fingerprint) 下全部信号行。
+        diag_fingerprint: 诊断口径指纹，隔离不同诊断配置的信号。
+        questions_dir: benchmark 题库目录，加载 GeneratedQuestion。
+        config: 切分旋钮快照。
+        out_path: 冻结 pools.json 目标路径（原子写）。
+        manifest_path: 溯源 manifest 目标路径（原子写）。
+        generated_at: 生成时间戳（ISO 字符串），由调用方传入以保证可复现。
+
+    返回:
+        SplitBuildResult，含 pools / manifest / assignment，支持字典式访问。
+
+    异常:
+        AssertionError: 六条防御断言任一不满足（fail-fast，不静默）。
+        ValueError: 上游依赖校验失败（如 correctness 缺题、assignment 非法）。
+    """
+    # Phase 1: canonical 基线预测 + 诊断信号。
+    preds = _read_canonical_predictions(db_path, baseline_run_id)
+    signal_rows_raw = signal_store.load(baseline_run_id, diag_fingerprint)
+    _assert_fingerprint_consistent(signal_rows_raw, diag_fingerprint)
+    signal_rows = [
+        {
+            "question_id": row.question_id,
+            "task_type": row.task_type,
+            "error_type": row.error_type,
+            "tier": row.tier,
+        }
+        for row in signal_rows_raw
+    ]
+
+    # Phase 2: 全视频画像 + 贪心联合约束选择。
+    videos = build_video_records(preds, signal_rows)
+    total_by_type = Counter(pred["task_type"] for pred in preds)
+    reportable_types = derive_reportable_types(dict(total_by_type), config.report_floor)
+    assignment_obj = select_split(
+        videos,
+        config=SelectConfig(
+            n_trainval=config.n_trainval,
+            floor_k=config.floor_k,
+            epsilon=config.epsilon,
+            reportable_types=reportable_types,
+            seed=config.select_seed,
+        ),
+    )
+    assignment = _assignment_to_dict(assignment_obj)
+    logger.info(
+        "视频级切分完成: trainval={} test={} (总 {} 视频)",
+        len(assignment_obj.trainval),
+        len(assignment_obj.test),
+        len(videos),
+    )
+
+    # Phase 3: 加载题库 + 视频归属切三池 + 原子冻结。
+    questions = load_benchmark(questions_dir)
+    correctness = {pred["question_id"]: pred["correct"] for pred in preds}
+    pools = split_by_video_assignment(
+        questions,
+        assignment,
+        correctness,
+        config.val_ratio,
+        config.split_seed,
+        baseline_run_id=baseline_run_id,
+    )
+    save_pools(pools, out_path)
+
+    # Phase 4: 溯源 manifest（pools_sha256 锚定冻结内容）。
+    coverage_report = _build_coverage_report(
+        videos, assignment_obj, signal_rows_raw, reportable_types, config
+    )
+    manifest = write_manifest(
+        manifest_path,
+        baseline_run_id=baseline_run_id,
+        diag_fingerprint=diag_fingerprint,
+        seed=config.split_seed,
+        config=asdict(config),
+        pools_json_text=out_path.read_text(encoding="utf-8"),
+        coverage_report=coverage_report,
+        generated_at=generated_at,
+    )
+
+    # Phase 5: 防御断言 fail-fast。
+    _assert_split_invariants(
+        pools=pools,
+        expected_question_ids={q.question_id for q in questions},
+        signal_rows_raw=signal_rows_raw,
+        diag_fingerprint=diag_fingerprint,
+        out_path=out_path,
+        manifest=manifest,
+    )
+
+    return SplitBuildResult(pools=pools, manifest=manifest, assignment=assignment)
+
+
+def _normalize_choice(choice: str | None) -> str:
+    """选项归一：strip → 大写 → 取首字母，None 归一为空串。
+
+    参数:
+        choice: 原始选项文本（预测或答案），可为 None。
+
+    返回:
+        归一后的单字母（无内容时为空串）。
+    """
+    return (choice or "").strip().upper()[:1]
+
+
+def _read_canonical_predictions(db_path: Path, baseline_run_id: str) -> list[dict]:
+    """从 harness.db 只读取指定 run 每题首行（ORDER BY rowid）为 canonical 预测。
+
+    同一 question_id 可能有多行（重跑 / 补测），canonical 口径取 rowid 最小的首行，
+    保证 distinct question 计数与对错判定确定。correct = 预测与答案归一后逐字符相等。
+
+    参数:
+        db_path: harness.db 路径。
+        baseline_run_id: 基线 run 标识。
+
+    返回:
+        canonical 预测行列表，每行 {video_id, question_id, task_type, correct}。
+
+    异常:
+        ValueError: 该 run 无任何预测行（fail-fast，不返回空切分）。
+
+    实现细节:
+        以 URI mode=ro 打开只读连接，绝不改动基线 db；按 rowid 升序遍历，
+        首次见到的 question_id 即 canonical 行，后续同 qid 行跳过。
+    """
+    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
+    conn.row_factory = sqlite3.Row
+    try:
+        rows = conn.execute(
+            "SELECT question_id, video_id, task_type, prediction, answer "
+            "FROM predictions WHERE run_id = ? ORDER BY rowid",
+            (baseline_run_id,),
+        ).fetchall()
+    finally:
+        conn.close()
+
+    canonical: dict[str, dict] = {}
+    for row in rows:
+        qid = row["question_id"]
+        if qid in canonical:
+            continue
+        canonical[qid] = {
+            "question_id": qid,
+            "video_id": row["video_id"],
+            "task_type": row["task_type"],
+            "correct": _normalize_choice(row["prediction"]) == _normalize_choice(row["answer"]),
+        }
+    if not canonical:
+        raise ValueError(f"run_id={baseline_run_id} 无任何预测行，无法切分")
+    return list(canonical.values())
+
+
+def _assignment_to_dict(assignment_obj: SplitAssignment) -> dict[str, str]:
+    """把 SplitAssignment 展平为 video_id -> "trainval" | "test" 归属字典。
+
+    参数:
+        assignment_obj: 贪心选择器产出的切分归属。
+
+    返回:
+        全部视频的归属字典（trainval 与 test 并集，键互斥）。
+    """
+    assignment = dict.fromkeys(assignment_obj.trainval, "trainval")
+    for vid in assignment_obj.test:
+        assignment[vid] = "test"
+    return assignment
+
+
+def _assert_fingerprint_consistent(
+    signal_rows_raw: list[DiagnosisSignalRow],
+    diag_fingerprint: str,
+) -> None:
+    """防御④：全部诊断信号行的 diag_fingerprint 必须与传入指纹一致。
+
+    参数:
+        signal_rows_raw: store 读回的诊断信号行。
+        diag_fingerprint: 期望的诊断口径指纹。
+
+    异常:
+        AssertionError: 存在指纹不一致的信号行（store 未正确按指纹过滤）。
+    """
+    mismatched = [
+        row.question_id for row in signal_rows_raw if row.diag_fingerprint != diag_fingerprint
+    ]
+    if mismatched:
+        raise AssertionError(
+            f"诊断信号指纹不一致 {len(mismatched)} 行，期望 {diag_fingerprint}: {mismatched[:5]}"
+        )
+
+
+def _fraction_by_type(records: list[VideoRecord], keys: set[str]) -> dict[str, float]:
+    """各 task_type 在给定视频集中的承载占比（含该题型的视频数 / 总视频数）。
+
+    参数:
+        records: 视频记录子集。
+        keys: 需计算占比的 task_type 键集。
+
+    返回:
+        {task_type: 占比}；records 为空时全部记 0.0。
+    """
+    total = len(records)
+    if total == 0:
+        return dict.fromkeys(keys, 0.0)
+    return {key: sum(1 for r in records if key in r.type_set) / total for key in keys}
+
+
+def _fraction_by_difficulty(records: list[VideoRecord], buckets: set[int]) -> dict[int, float]:
+    """各难度桶（错题数）在给定视频集中的占比。
+
+    参数:
+        records: 视频记录子集。
+        buckets: 难度桶键集。
+
+    返回:
+        {难度桶: 占比}；records 为空时全部记 0.0。
+    """
+    total = len(records)
+    if total == 0:
+        return dict.fromkeys(buckets, 0.0)
+    return {bucket: sum(1 for r in records if r.difficulty == bucket) / total for bucket in buckets}
+
+
+def _max_dev(global_dist: dict, subset_dist: dict, keys: set) -> float:
+    """逐键取全局与子集分布的最大绝对偏差（键集为空约定 0.0）。"""
+    if not keys:
+        return 0.0
+    return max(abs(global_dist.get(k, 0.0) - subset_dist.get(k, 0.0)) for k in keys)
+
+
+def _build_coverage_report(
+    videos: list[VideoRecord],
+    assignment_obj: SplitAssignment,
+    signal_rows_raw: list[DiagnosisSignalRow],
+    reportable_types: set[str],
+    config: SplitBuildConfig,
+) -> dict:
+    """组装 manifest 覆盖报告：48 格覆盖 / floor 达标 / test 代表性偏差 / tier 占比。
+
+    参数:
+        videos: 全视频画像记录。
+        assignment_obj: 切分归属（trainval / test）。
+        signal_rows_raw: 诊断信号行（统计 tier 占比）。
+        reportable_types: 参与 ε 代表性校验的题型集。
+        config: 切分旋钮（floor_k / epsilon）。
+
+    返回:
+        覆盖报告字典，含 cells_covered / grid_total / floor_satisfied /
+        test_representativeness_deviation / tier_distribution。
+    """
+    by_id = {v.video_id: v for v in videos}
+    trainval = [by_id[vid] for vid in assignment_obj.trainval]
+    test = [by_id[vid] for vid in assignment_obj.test]
+
+    covered_cells: set[tuple[str, str]] = set()
+    trainval_wrong: Counter[str] = Counter()
+    for video in trainval:
+        covered_cells |= set(video.cells)
+        trainval_wrong.update(video.wrong_by_type)
+    floor_satisfied = {
+        task_type: trainval_wrong.get(task_type, 0) >= floor
+        for task_type, floor in config.floor_k.items()
+    }
+
+    type_dev = _max_dev(
+        _fraction_by_type(videos, reportable_types),
+        _fraction_by_type(test, reportable_types),
+        reportable_types,
+    )
+    diff_buckets = {r.difficulty for r in videos}
+    diff_dev = _max_dev(
+        _fraction_by_difficulty(videos, diff_buckets),
+        _fraction_by_difficulty(test, diff_buckets),
+        diff_buckets,
+    )
+
+    tier_counts = Counter(row.tier for row in signal_rows_raw)
+    total_signals = sum(tier_counts.values())
+    tier_distribution = (
+        {tier: count / total_signals for tier, count in tier_counts.items()}
+        if total_signals
+        else {}
+    )
+
+    return {
+        "cells_covered": len(covered_cells),
+        "grid_total": _DIVERSITY_GRID_TOTAL,
+        "floor_satisfied": floor_satisfied,
+        "test_representativeness_deviation": {
+            "type_max": type_dev,
+            "difficulty_max": diff_dev,
+            "epsilon": config.epsilon,
+        },
+        "tier_distribution": tier_distribution,
+    }
+
+
+def _assert_split_invariants(
+    *,
+    pools: Pools,
+    expected_question_ids: set[str],
+    signal_rows_raw: list[DiagnosisSignalRow],
+    diag_fingerprint: str,
+    out_path: Path,
+    manifest: dict,
+) -> None:
+    """六条防御断言 fail-fast：任一不满足即 AssertionError（P5，不静默不兜底）。
+
+    ① 三池视频集两两不相交；② 三池覆盖全部题（按 distinct question）；
+    ③ 每 video 恰 3 题；④ 诊断信号指纹一致；⑤ manifest pools_sha256 == sha256(冻结内容)；
+    ⑥ question_id 全局唯一（无重复行）。
+
+    参数:
+        pools: 冻结的三池。
+        expected_question_ids: 加载题库的 question_id 全集（覆盖基准）。
+        signal_rows_raw: 诊断信号行（指纹校验）。
+        diag_fingerprint: 期望诊断指纹。
+        out_path: 冻结 pools.json 路径。
+        manifest: 已写入的 manifest 字典。
+
+    异常:
+        AssertionError: 任一防御断言不满足。
+    """
+    all_questions = pools.diagnosis + pools.validation + pools.test
+    _assert_pools_video_disjoint(pools)  # ①
+    _assert_question_ids_unique(all_questions)  # ⑥
+    _assert_question_coverage(all_questions, expected_question_ids)  # ②
+    _assert_three_questions_per_video(all_questions)  # ③
+    _assert_fingerprint_consistent(signal_rows_raw, diag_fingerprint)  # ④
+    _assert_pools_sha256(out_path, manifest)  # ⑤
+
+
+def _assert_pools_video_disjoint(pools: Pools) -> None:
+    """防御①：diagnosis / validation / test 三池视频集两两不相交。"""
+    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}
+    if diag_v & val_v or diag_v & test_v or val_v & test_v:
+        raise AssertionError(
+            f"三池视频集非互斥: diag∩val={diag_v & val_v}, "
+            f"diag∩test={diag_v & test_v}, val∩test={val_v & test_v}"
+        )
+
+
+def _assert_question_ids_unique(all_questions: list[GeneratedQuestion]) -> None:
+    """防御⑥：三池合并后 question_id 全局唯一（无重复行）。"""
+    qids = [q.question_id for q in all_questions]
+    if len(qids) != len(set(qids)):
+        duplicates = [qid for qid, count in Counter(qids).items() if count > 1]
+        raise AssertionError(f"question_id 重复 {len(duplicates)} 个: {duplicates[:5]}")
+
+
+def _assert_question_coverage(
+    all_questions: list[GeneratedQuestion],
+    expected_question_ids: set[str],
+) -> None:
+    """防御②：三池覆盖题库全部题（按 distinct question，缺题 / 多题均 fail-fast）。"""
+    actual = {q.question_id for q in all_questions}
+    if actual != expected_question_ids:
+        missing = expected_question_ids - actual
+        extra = actual - expected_question_ids
+        raise AssertionError(
+            f"三池题目覆盖不完整: 缺 {len(missing)} 多 {len(extra)} (缺样例 {sorted(missing)[:5]})"
+        )
+
+
+def _assert_three_questions_per_video(all_questions: list[GeneratedQuestion]) -> None:
+    """防御③：每 video 恰 3 题（视频组原子切分不应劈裂视频的题）。"""
+    per_video = Counter(q.video_id for q in all_questions)
+    bad_videos = {vid: n for vid, n in per_video.items() if n != _QUESTIONS_PER_VIDEO}
+    if bad_videos:
+        raise AssertionError(
+            f"存在 video 题数 != {_QUESTIONS_PER_VIDEO}: {dict(list(bad_videos.items())[:5])}"
+        )
+
+
+def _assert_pools_sha256(out_path: Path, manifest: dict) -> None:
+    """防御⑤：manifest 的 pools_sha256 == sha256(冻结 pools.json 内容)。"""
+    actual_sha = hashlib.sha256(out_path.read_text(encoding="utf-8").encode("utf-8")).hexdigest()
+    if actual_sha != manifest["pools_sha256"]:
+        raise AssertionError(
+            f"pools_sha256 不一致: manifest={manifest['pools_sha256']} 实际={actual_sha}"
+        )

tests/integration/test_build_split_e2e.py

@@ -0,0 +1,155 @@
+"""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()