From 8fef7ced42cdf69ef91973ed94e203ff68cd7fd6 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 15 Jul 2026 13:39:14 -0400 Subject: [PATCH] fix: address whole-impl review (INFRA T0 rows, reproducible manifest, evolution_target report, dead config, canonical DRY) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit C-1: persist_infra_t0_rows 补 INFRA/空预测错题的 T0 信号行（不进诊断故须单独落库），run_pipeline 加 Phase 0，dry-run 用假数据走通。 C-2: CLI 加 --generated-at，真实运行默认盖真实 UTC now，可显式固定以字节级复现 manifest。 I-1: coverage_report 增 evolution_target_distribution（T2 信号按 tool/skill/system 计数）。 I-2: 删除 PoolConfig 死字段 n_trainval/floor_k/epsilon/report_floor/val_wrong_min（grep 确认无消费者，视频级切分用独立 VideoSplitConfig/SplitBuildConfig/SelectConfig）。 I-3: 抽共享 load_canonical_predictions(db_path, run_id)，CLI 与 build_split 共用；消除 canonical 取行 + correct 判定重复。 M-1: build_split docstring 注明 val_wrong_min-agnostic 契约（McNemar 护栏由 CLI 冻结后执行，Task 11 契约）。 --- app/harness/build_split.py | 80 ++++++++-- app/harness/video_split_cli.py | 170 +++++++++++++++------ core/types.py | 17 +-- tests/integration/test_build_split_e2e.py | 7 + tests/unit/test_pool_config_video_split.py | 59 +------ tests/unit/test_video_split_cli.py | 103 ++++++++++++- 6 files changed, 302 insertions(+), 134 deletions(-) diff --git a/app/harness/build_split.py b/app/harness/build_split.py index bf36a40..5131163 100644 --- a/app/harness/build_split.py +++ b/app/harness/build_split.py @@ -119,6 +119,11 @@ def build_split( test → 加载题库并以视频归属切三池 → 原子冻结 pools.json → 写溯源 manifest → 六条防御断言 fail-fast 校验。 + 契约（Task 11，非疏漏）：build_split 有意保持 val_wrong_min-agnostic——内部调 + split_by_video_assignment 时不传 val_wrong_min（默认 0，不校验 validation 错题 + 数）。McNemar 功效护栏是切分**冻结后**的独立校验，由 CLI 的 check_mcnemar_power + 在 build_split 返回后执行；切分构造本身不因功效阈失败，二者关注点分离。 + 参数: db_path: harness.db 路径（只读读取 predictions，不改动）。 baseline_run_id: 基线 run 标识（如 "infer_adhoc"）。 @@ -138,7 +143,7 @@ def build_split( ValueError: 上游依赖校验失败（如 correctness 缺题、assignment 非法）。 """ # Phase 1: canonical 基线预测 + 诊断信号。 - preds = _read_canonical_predictions(db_path, baseline_run_id) + preds = load_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 = [ @@ -226,31 +231,38 @@ def _normalize_choice(choice: str | None) -> str: return (choice or "").strip().upper()[:1] -def _read_canonical_predictions(db_path: Path, baseline_run_id: str) -> list[dict]: +def load_canonical_predictions(db_path: Path, baseline_run_id: str) -> list[dict]: """从 harness.db 只读取指定 run 每题首行（ORDER BY rowid）为 canonical 预测。 + 共享口径 helper：CLI（可诊断错题筛选 + INFRA T0 补录）与 build_split（切分） + 共用同一"每 qid 取 rowid 最小首行 + 归一化 correct 判定"口径，消除两处重复实现。 同一 question_id 可能有多行（重跑 / 补测），canonical 口径取 rowid 最小的首行， - 保证 distinct question 计数与对错判定确定。correct = 预测与答案归一后逐字符相等。 + 保证 distinct question 计数与对错判定确定。correct = 预测与答案各自归一 + （strip → 大写 → 取首字母）后逐字符相等。 + + 口径边界：旧 build_or_load_pools 的 legacy 池构建路径（app/harness/pools.py）是 + 另一条独立既有链路，不共用本 helper，两者刻意不统一（本次不动 legacy 路径）。 参数: - db_path: harness.db 路径。 + db_path: harness.db 路径（URI mode=ro 只读打开，绝不改动基线 db）。 baseline_run_id: 基线 run 标识。 返回: - canonical 预测行列表，每行 {video_id, question_id, task_type, correct}。 + canonical 预测行列表，每行含 question_id / video_id / task_type / + prediction / answer / stop_reason / correct（bool）。按 rowid 升序去重， + 每 qid 保留首行。 异常: ValueError: 该 run 无任何预测行（fail-fast，不返回空切分）。 实现细节: - 以 URI mode=ro 打开只读连接，绝不改动基线 db；按 rowid 升序遍历， - 首次见到的 question_id 即 canonical 行，后续同 qid 行跳过。 + 按 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 " + "SELECT question_id, video_id, task_type, prediction, answer, stop_reason " "FROM predictions WHERE run_id = ? ORDER BY rowid", (baseline_run_id,), ).fetchall() @@ -266,6 +278,9 @@ def _read_canonical_predictions(db_path: Path, baseline_run_id: str) -> list[dic "question_id": qid, "video_id": row["video_id"], "task_type": row["task_type"], + "prediction": row["prediction"], + "answer": row["answer"], + "stop_reason": row["stop_reason"], "correct": _normalize_choice(row["prediction"]) == _normalize_choice(row["answer"]), } if not canonical: @@ -367,7 +382,13 @@ def _build_coverage_report( 返回: 覆盖报告字典，含 cells_covered / grid_total / floor_satisfied / - test_representativeness_deviation / tier_distribution。 + test_representativeness_deviation / tier_distribution / + evolution_target_distribution。 + + 实现细节: + evolution_target_distribution 只统计 T2 信号（可训练缺陷），按 + tool / skill / system 计数，报告"哪层参数组拿到梯度"；T0/T1/uncertain 行 + evolution_target 恒为 None，不入该分布。 """ by_id = {v.video_id: v for v in videos} trainval = [by_id[vid] for vid in assignment_obj.trainval] @@ -395,13 +416,7 @@ def _build_coverage_report( 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 {} - ) + tier_distribution, evolution_target_distribution = _signal_distributions(signal_rows_raw) return { "cells_covered": len(covered_cells), @@ -413,9 +428,42 @@ def _build_coverage_report( "epsilon": config.epsilon, }, "tier_distribution": tier_distribution, + "evolution_target_distribution": evolution_target_distribution, } +def _signal_distributions( + signal_rows_raw: list[DiagnosisSignalRow], +) -> tuple[dict[str, float], dict[str, int]]: + """由诊断信号行算 tier 占比分布与 T2 进化目标计数分布。 + + 参数: + signal_rows_raw: 诊断信号行。 + + 返回: + (tier_distribution, evolution_target_distribution) 二元组： + - tier_distribution: {tier: 占比}，无信号时为空 dict； + - evolution_target_distribution: 仅统计 T2（可训练缺陷）信号，按 + tool / skill / system 计数，报告哪层参数组拿到梯度；T0/T1/uncertain 行 + evolution_target 恒为 None，不入该分布。 + """ + 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 {} + ) + evolution_target_distribution = dict( + Counter( + row.evolution_target + for row in signal_rows_raw + if row.tier == "T2" and row.evolution_target is not None + ) + ) + return tier_distribution, evolution_target_distribution + + def _assert_split_invariants( *, pools: Pools, diff --git a/app/harness/video_split_cli.py b/app/harness/video_split_cli.py index 1a66813..0695779 100644 --- a/app/harness/video_split_cli.py +++ b/app/harness/video_split_cli.py @@ -5,6 +5,14 @@ Phase 2 冻结切分（build_split，纯 code-controlled，产出 pools.json + manifest）→ McNemar 功效护栏（validation 池错题数达阈校验）。 +复现锚点约定（C-2）： + - pools.json 的内容（+ seed + diag_fingerprint）是切分的**复现锚点**——相同输入 + 产出字节级相同的 pools.json 与 pools_sha256。 + - manifest 的 generated_at 是**溯源元数据**，非复现锚点：真实运行默认盖真实 UTC + now（记录本次切分何时产出），但可用 `--generated-at ` 显式固定，以对 + manifest 做字节级复现比对。write_manifest 库内不调 datetime.now，时间戳一律由 + 本 CLI 传入。 + 设计要点： - 诊断口径指纹 = (诊断 prompt 版本, 模型名, git 短 SHA) 三分量合成，隔离不同 诊断配置的信号；换 prompt / 模型 / 代码实现即换指纹，旧信号不被覆盖。 @@ -16,7 +24,8 @@ 用于校验装配正确性（对齐 CLAUDE.md §2.5 smoke test）。 编排函数（run_pipeline）通过依赖注入接收 DiagnosisDeps / signal_store / wrong_ids / -questions，便于单测用假实现替换、不触真实 LLM 与 harness.db。 +questions / canonical_preds，便于单测用假实现替换、不触真实 LLM 与 harness.db。 +其中 canonical_preds 供 Phase 0 补 INFRA / 空预测错题的 T0 信号（这些题不进诊断）。 """ from __future__ import annotations @@ -25,7 +34,6 @@ import argparse import asyncio import datetime import os -import sqlite3 import subprocess from dataclasses import dataclass from pathlib import Path @@ -35,9 +43,15 @@ import yaml from loguru import logger from app.harness.baseline_diagnosis import DiagnosisDeps, run_baseline_diagnosis -from app.harness.build_split import SplitBuildConfig, SplitBuildResult, build_split +from app.harness.build_split import ( + SplitBuildConfig, + SplitBuildResult, + build_split, + load_canonical_predictions, +) from app.harness.split_selection import diag_fingerprint from app.question_gen.loader import load_benchmark +from core.evolution.types import DiagnosisSignalRow if TYPE_CHECKING: from app.harness.pools import Pools @@ -351,57 +365,81 @@ def _load_diagnose_prompts() -> Any: ) -def _normalize_choice(choice: str | None) -> str: - """选项归一：strip → 大写 → 取首字母（None 归一为空串）。""" - return (choice or "").strip().upper()[:1] +def select_diagnosable_wrong_ids(preds: list[dict]) -> list[str]: + """从 canonical 预测筛出可诊断错题 question_id（保序）。 - -def load_diagnosable_wrong_ids(harness_db: Path, baseline_run_id: str) -> list[str]: - """从 harness.db 读 baseline run 的可诊断错题 question_id（保序、canonical 首行）。 - - 可诊断错题判据：canonical 首行（rowid 最小）预测非空 且 stop_reason 非 INFRA - （error / parse_error）且 归一后预测 != 答案。INFRA / 空预测题不进 wrong_ids - （run_diagnosis 内部也会二次排除，此处前置过滤减少无谓 LLM 调用）。 + 可诊断错题判据：预测非空 且 stop_reason 非 INFRA（error / parse_error）且 + 归一后预测 != 答案。INFRA / 空预测错题不进 wrong_ids——它们改由 + persist_infra_t0_rows 直接落 T0（run_diagnosis 内部也会二次排除同类题）。 参数: - harness_db: harness.db 路径（只读打开）。 - baseline_run_id: 基线 run 标识。 + preds: load_canonical_predictions 产出的 canonical 预测行（已按 qid 去重）。 返回: - 可诊断错题 question_id 列表（按 rowid 升序 canonical 顺序，去重）。 - - 异常: - SystemExit: 该 run 无任何预测行（fail loud）。 + 可诊断错题 question_id 列表（保 preds 顺序）。 """ - conn = sqlite3.connect(f"file:{harness_db}?mode=ro", uri=True) - conn.row_factory = sqlite3.Row - try: - rows = conn.execute( - "SELECT question_id, prediction, answer, stop_reason " - "FROM predictions WHERE run_id = ? ORDER BY rowid", - (baseline_run_id,), - ).fetchall() - finally: - conn.close() - if not rows: - raise SystemExit( - f"run_id={baseline_run_id} 在 {harness_db} 无任何预测行，无法诊断（P5 fail loud）" - ) - seen: set[str] = set() wrong_ids: list[str] = [] - for row in rows: - qid = row["question_id"] - if qid in seen: + for pred in preds: + prediction = (pred["prediction"] or "").strip() + if not prediction or pred["stop_reason"] in _INFRA_STOP_REASONS: continue - seen.add(qid) - prediction = (row["prediction"] or "").strip() - if not prediction or row["stop_reason"] in _INFRA_STOP_REASONS: - continue - if _normalize_choice(row["prediction"]) != _normalize_choice(row["answer"]): - wrong_ids.append(qid) + if not pred["correct"]: + wrong_ids.append(pred["question_id"]) return wrong_ids +def persist_infra_t0_rows( + store: DiagnosisSignalStore, + preds: list[dict], + baseline_run_id: str, + diag_fingerprint: str, +) -> int: + """把非正确且 INFRA / 空预测的错题以 T0 信号行 upsert 落库（幂等）。 + + 这些题（stop_reason ∈ {error, parse_error} 或预测为空）从不进入 run_diagnosis + （筛选时被前置排除），故其 T0 信号必须在此单独补齐——否则 signal store 缺这些行， + tier 分布 / manifest 不完整（计划要求 4 个 INFRA 空预测错题 → T0）。 + + 投影口径与 baseline_diagnosis 的 INFRA 投影一致：infra=True、tier="T0"、 + error_type / cause_category / evolution_target 均 None、degraded=False； + video_id / task_type 从 canonical 预测取。store.upsert 按主键 + (question_id, baseline_run_id, diag_fingerprint) 幂等，重复调用零副作用。 + + 参数: + store: 诊断信号存储端口（与诊断落库同一 store）。 + preds: load_canonical_predictions 产出的 canonical 预测行。 + baseline_run_id: 基线 run 标识（信号行主键之一）。 + diag_fingerprint: 诊断口径指纹（信号行主键之一）。 + + 返回: + 落库的 T0 行数（供日志）。 + """ + count = 0 + for pred in preds: + prediction = (pred["prediction"] or "").strip() + is_infra_or_empty = pred["stop_reason"] in _INFRA_STOP_REASONS or not prediction + if pred["correct"] or not is_infra_or_empty: + continue + 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=None, + cause_category=None, + tier="T0", + evolution_target=None, + degraded=False, + infra=True, + session_id=None, + ) + ) + count += 1 + return count + + def load_questions_by_id(questions_dir: Path) -> dict[str, GeneratedQuestion]: """加载 benchmark 全部题并建 question_id → GeneratedQuestion 映射。 @@ -460,28 +498,34 @@ async def run_pipeline( signal_store: DiagnosisSignalStore, wrong_ids: list[str], questions: dict[str, GeneratedQuestion], + canonical_preds: list[dict], harness_db: Path, questions_dir: Path, out_dir: Path, generated_at: str, ) -> SplitBuildResult: - """内联两阶段：Phase 1 诊断 → Phase 2 冻结切分 → McNemar 护栏。 + """内联三阶段：Phase 0 INFRA T0 补录 → Phase 1 诊断 → Phase 2 冻结切分 → McNemar 护栏。 参数: config: 科研旋钮快照。 fingerprint: 诊断口径指纹（已合成，作诊断信号主键之一）。 diagnosis_deps: Phase 1 诊断依赖束（真实或假实现）。 - signal_store: 诊断信号存储端口（Phase 1 写、Phase 2 读）。 + signal_store: 诊断信号存储端口（Phase 0/1 写、Phase 2 读）。 wrong_ids: 待诊断的可诊断错题 question_id 列表。 questions: question_id → GeneratedQuestion 映射。 + canonical_preds: canonical 预测行（Phase 0 从中筛 INFRA / 空预测错题补 T0）。 harness_db: harness.db 路径（Phase 2 读 canonical 预测）。 questions_dir: benchmark 题库目录（Phase 2 加载题库切池）。 out_dir: 冻结产物目录（pools.json + split_manifest.json）。 - generated_at: 生成时间戳（ISO 字符串，由调用方传入保证可复现）。 + generated_at: 生成时间戳（ISO 字符串，由调用方传入；见模块 C-2 复现锚点约定）。 返回: SplitBuildResult（冻结三池 + manifest + assignment）。 """ + # Phase 0: INFRA / 空预测错题补 T0（这些题不进诊断，须单独落库保证 tier 分布/manifest 完整）。 + n_t0 = persist_infra_t0_rows(signal_store, canonical_preds, config.baseline_run_id, fingerprint) + logger.info("Phase 0 INFRA T0 补录：落库 {} 行（INFRA / 空预测错题不进诊断）", n_t0) + # Phase 1: 离线诊断（断点续跑幂等：done_question_ids 已完成题跳过）。 logger.info( "Phase 1 离线诊断：baseline={} 待诊断错题 {} 题", config.baseline_run_id, len(wrong_ids) @@ -548,12 +592,16 @@ def _execute_real(config: VideoSplitConfig, fingerprint: str, args: argparse.Nam harness_db, questions_dir, out_dir = _resolve_paths(args) if not harness_db.exists(): raise SystemExit(f"harness.db 不存在: {harness_db}（P5 fail loud）") - wrong_ids = load_diagnosable_wrong_ids(harness_db, config.baseline_run_id) + canonical_preds = load_canonical_predictions(harness_db, config.baseline_run_id) + wrong_ids = select_diagnosable_wrong_ids(canonical_preds) questions = load_questions_by_id(questions_dir) deps = build_diagnosis_deps( harness_db=harness_db, concurrency=args.concurrency, expected_model=config.model ) + # generated_at：默认盖真实 UTC now（溯源用），--generated-at 可显式固定以复现（C-2）。 + generated_at = args.generated_at or datetime.datetime.now(datetime.UTC).isoformat() + from adapters.baseline_diagnosis_store import SqliteDiagnosisSignalStore store = SqliteDiagnosisSignalStore(str(harness_db)) @@ -566,10 +614,11 @@ def _execute_real(config: VideoSplitConfig, fingerprint: str, args: argparse.Nam signal_store=store, wrong_ids=wrong_ids, questions=questions, + canonical_preds=canonical_preds, harness_db=harness_db, questions_dir=questions_dir, out_dir=out_dir, - generated_at=datetime.datetime.now(datetime.UTC).isoformat(), + generated_at=generated_at, ) ) finally: @@ -637,6 +686,20 @@ def _execute_dry_run(config: VideoSplitConfig, fingerprint: str, args: argparse. dry_db.parent.mkdir(parents=True, exist_ok=True) store = SqliteDiagnosisSignalStore(str(dry_db)) try: + # Phase 0 装配：用一条假 INFRA 空预测走通 persist_infra_t0_rows（不触 LLM）。 + fake_infra_preds = [ + { + "question_id": "_dry_infra", + "video_id": "_dry_v", + "task_type": "Counting Problem", + "prediction": "", + "answer": "A", + "stop_reason": "error", + "correct": False, + } + ] + n_t0 = persist_infra_t0_rows(store, fake_infra_preds, config.baseline_run_id, fingerprint) + logger.info("Phase 0 装配 OK：persist_infra_t0_rows 落 {} 行 INFRA T0（假数据）", n_t0) logger.info("Phase 1 装配 OK：run_baseline_diagnosis 以空错题走早返回路径（不触 LLM）") asyncio.run( run_baseline_diagnosis( @@ -667,6 +730,17 @@ def build_arg_parser() -> argparse.ArgumentParser: parser.add_argument("--harness-db", type=Path, default=None, dest="harness_db") parser.add_argument("--questions-dir", type=Path, default=None, dest="questions_dir") parser.add_argument("--out-dir", type=Path, default=None, dest="out_dir") + parser.add_argument( + "--generated-at", + type=str, + default=None, + dest="generated_at", + help=( + "manifest generated_at 时间戳（ISO 字符串）；默认盖真实 UTC now（溯源元数据）。" + "复现锚点是 pools.json 内容 + seed + fingerprint；generated_at 可显式传入以" + "对 manifest 做字节级复现比对。" + ), + ) return parser diff --git a/core/types.py b/core/types.py index bb74667..22cd857 100644 --- a/core/types.py +++ b/core/types.py @@ -160,16 +160,12 @@ class PoolConfig: eval_min_per_class: 验证池中每类保底样本数（GlobalStrategy 用）。 train_ratio: train/(train+val) 比例（PerCategoryStrategy 用）。 test_questions_dir: 外部 test 题源路径（PerCategoryStrategy 用）。 - n_trainval: trainval 目标视频数（结果驱动视频级切分用；0 表示不启用）。 - floor_k: 各高信号 task_type 的 T2 defect 下限（视频级切分硬约束；空表示无约束）。 - epsilon: test 相对全局的最大允许分布偏差（视频级切分 test 代表性守护）。 - report_floor: per-type 报告门限，题数 ≥ 此值的 task_type 才入 ε 约束（视频级切分用）。 - val_wrong_min: validation 池最少错题数（McNemar 功效护栏；0 表示不检查）。 实现细节: - 视频级切分五个旋钮均带惰性默认（0 / 空 dict），使现有 GlobalPoolStrategy / - PerCategoryStrategy 的构造点无需改动即可保持行为不变。floor_k 为不可哈希 dict， - 标 hash=False 排除出 frozen dataclass 的自动 __hash__，避免入 set/dict 键时报错。 + 结果驱动视频级切分不复用本配置——它有独立的 VideoSplitConfig / + SplitBuildConfig / SelectConfig（见 app/harness/video_split_cli.py 与 + split_selection.py），故本类不承载 n_trainval / floor_k / epsilon 等视频级 + 切分旋钮，避免死配置面。 """ task_types: tuple[str, ...] | None @@ -184,8 +180,3 @@ class PoolConfig: train_ratio: float test_questions_dir: _Path | None batch_correct_ratio: float | None = None - n_trainval: int = 0 - floor_k: dict[str, int] = field(default_factory=dict, hash=False) - epsilon: float = 0.0 - report_floor: int = 0 - val_wrong_min: int = 0 diff --git a/tests/integration/test_build_split_e2e.py b/tests/integration/test_build_split_e2e.py index 4307af4..47699a6 100644 --- a/tests/integration/test_build_split_e2e.py +++ b/tests/integration/test_build_split_e2e.py @@ -153,3 +153,10 @@ def test_end_to_end_freezes_valid_pools(tmp_path: Path) -> None: 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"} diff --git a/tests/unit/test_pool_config_video_split.py b/tests/unit/test_pool_config_video_split.py index af45db0..7ea16f7 100644 --- a/tests/unit/test_pool_config_video_split.py +++ b/tests/unit/test_pool_config_video_split.py @@ -1,10 +1,12 @@ -"""视频级切分科研旋钮单元测试：诊断指纹 + val_wrong_min 功效护栏 + PoolConfig 新字段。 +"""视频级切分科研旋钮单元测试：诊断指纹 + val_wrong_min 功效护栏。 覆盖： - diag_fingerprint 对 (prompt 版本 / 模型 / 代码版本) 三元组确定且敏感； - split_by_video_assignment 的 val_wrong_min 门控 fail loud（验证信号不足即报错）； - - val_wrong_min 默认 0 时行为与 Task 11 现有调用完全一致（不回归）； - - PoolConfig 能接收视频级切分的五个新旋钮字段（纯 dataclass 装配）。 + - val_wrong_min 默认 0 时行为与 Task 11 现有调用完全一致（不回归）。 + +注：结果驱动视频级切分不复用 PoolConfig——它有独立的 VideoSplitConfig / +SplitBuildConfig / SelectConfig，故 PoolConfig 不承载视频级切分旋钮（无死配置面）。 """ from __future__ import annotations @@ -13,7 +15,7 @@ import pytest from app.harness.pools import InsufficientValSignal, split_by_video_assignment from app.harness.split_selection import diag_fingerprint -from core.types import GeneratedQuestion, PoolConfig +from core.types import GeneratedQuestion def _q(qid: str, vid: str, tt: str = "Counting Problem") -> GeneratedQuestion: @@ -64,52 +66,3 @@ def test_val_wrong_min_default_zero_no_regression(): qs, assignment, correctness=correctness, val_ratio=1.0, seed=0 ) assert len(pools.validation) == 2 # 未抛异常，正常返回 - - -def test_pool_config_accepts_video_split_knobs(): - """PoolConfig 能接收视频级切分五个新旋钮字段（默认惰性，不破坏现有构造点）。""" - cfg = PoolConfig( - task_types=None, - seed=0, - baseline_run_id="infer_adhoc", - diag_size=200, - diag_correct_ratio=0.5, - val_size=30, - val_correct_ratio=0.5, - test_size=60, - eval_min_per_class=2, - train_ratio=0.667, - test_questions_dir=None, - n_trainval=100, - floor_k={"Counting Problem": 3}, - epsilon=0.1, - report_floor=27, - val_wrong_min=20, - ) - assert cfg.n_trainval == 100 - assert cfg.floor_k == {"Counting Problem": 3} - assert cfg.epsilon == 0.1 - assert cfg.report_floor == 27 - assert cfg.val_wrong_min == 20 - - -def test_pool_config_video_split_knobs_default_inert(): - """未传视频级切分字段时默认惰性（0 / 空 dict），不破坏 GlobalPoolStrategy 现有构造。""" - cfg = PoolConfig( - task_types=None, - seed=0, - baseline_run_id="run_1", - diag_size=200, - diag_correct_ratio=0.5, - val_size=30, - val_correct_ratio=0.5, - test_size=60, - eval_min_per_class=2, - train_ratio=0.667, - test_questions_dir=None, - ) - assert cfg.n_trainval == 0 - assert cfg.floor_k == {} - assert cfg.epsilon == 0.0 - assert cfg.report_floor == 0 - assert cfg.val_wrong_min == 0 diff --git a/tests/unit/test_video_split_cli.py b/tests/unit/test_video_split_cli.py index ef7447b..02b2e69 100644 --- a/tests/unit/test_video_split_cli.py +++ b/tests/unit/test_video_split_cli.py @@ -102,10 +102,18 @@ def test_check_mcnemar_power_zero_threshold_skips(): assert cli.check_mcnemar_power(pools, val_wrong_min=0) == 0 -def test_run_pipeline_orders_two_phases(monkeypatch, tmp_path): - """run_pipeline 先跑 Phase 1 诊断、后跑 Phase 2 build_split（按序）。""" +def test_run_pipeline_orders_three_phases(monkeypatch, tmp_path): + """run_pipeline 先补 Phase 0 INFRA T0、再 Phase 1 诊断、后 Phase 2 build_split（按序）。""" calls: list[str] = [] + class _SpyStore: + def __init__(self): + self.t0_rows: list = [] + + def upsert(self, row): + calls.append("t0_upsert") + self.t0_rows.append(row) + async def fake_diag(**kwargs): calls.append("diagnosis") assert kwargs["diag_fingerprint"] == "fp" @@ -120,24 +128,111 @@ def test_run_pipeline_orders_two_phases(monkeypatch, tmp_path): monkeypatch.setattr(cli, "run_baseline_diagnosis", fake_diag) monkeypatch.setattr(cli, "build_split", fake_build_split) + # 一条 INFRA 空预测错题 → Phase 0 应补一行 T0（在诊断/切分之前）。 + canonical_preds = [ + { + "question_id": "q_infra", + "video_id": "v9", + "task_type": "Counting Problem", + "prediction": "", + "answer": "A", + "stop_reason": "error", + "correct": False, + } + ] + store = _SpyStore() + result = asyncio.run( cli.run_pipeline( config=_config(), fingerprint="fp", diagnosis_deps=object(), - signal_store=object(), + signal_store=store, wrong_ids=["q1"], questions={}, + canonical_preds=canonical_preds, harness_db=tmp_path / "h.db", questions_dir=tmp_path, out_dir=tmp_path / "out", generated_at="2026-07-15T00:00:00Z", ) ) - assert calls == ["diagnosis", "build_split"] + assert calls == ["t0_upsert", "diagnosis", "build_split"] # Phase 0 先于诊断与切分 + assert len(store.t0_rows) == 1 + assert store.t0_rows[0].tier == "T0" and store.t0_rows[0].infra is True assert result.pools.validation == [] +def test_select_diagnosable_wrong_ids_excludes_infra_and_correct(): + """可诊断错题筛选：排除 INFRA / 空预测 / 正确题，保留非空非 INFRA 错题（保序）。""" + preds = [ + {"question_id": "ok", "stop_reason": "finished", "prediction": "B", "correct": True}, + {"question_id": "wrong", "stop_reason": "finished", "prediction": "C", "correct": False}, + {"question_id": "infra", "stop_reason": "error", "prediction": "", "correct": False}, + {"question_id": "parse", "stop_reason": "parse_error", "prediction": "x", "correct": False}, + {"question_id": "empty", "stop_reason": "finished", "prediction": "", "correct": False}, + ] + assert cli.select_diagnosable_wrong_ids(preds) == ["wrong"] + + +def test_persist_infra_t0_rows_persists_only_infra_or_empty_wrong(tmp_path): + """INFRA / 空预测错题落 T0（infra=True，error_type/target=None）；正确 / 可诊断错题不落。""" + from adapters.baseline_diagnosis_store import SqliteDiagnosisSignalStore + + preds = [ + { + "question_id": "ok", + "video_id": "v1", + "task_type": "Counting Problem", + "prediction": "A", + "answer": "A", + "stop_reason": "finished", + "correct": True, + }, + { + "question_id": "diag_wrong", + "video_id": "v2", + "task_type": "Counting Problem", + "prediction": "C", + "answer": "A", + "stop_reason": "finished", + "correct": False, + }, + { + "question_id": "infra_err", + "video_id": "v3", + "task_type": "OCR Problems", + "prediction": "", + "answer": "A", + "stop_reason": "error", + "correct": False, + }, + { + "question_id": "parse_err", + "video_id": "v4", + "task_type": "Counting Problem", + "prediction": "", + "answer": "A", + "stop_reason": "parse_error", + "correct": False, + }, + ] + store = SqliteDiagnosisSignalStore(str(tmp_path / "h.db")) + n = cli.persist_infra_t0_rows(store, preds, "infer_adhoc", "fp") + assert n == 2 # 只有两条 INFRA 空预测错题 + rows = {r.question_id: r for r in store.load("infer_adhoc", "fp")} + assert set(rows) == {"infra_err", "parse_err"} + for r in rows.values(): + assert r.tier == "T0" and r.infra is True + assert r.error_type is None and r.evolution_target is None and r.cause_category is None + assert r.degraded is False + + # 幂等：重复调用同 PK 覆盖，行数不变。 + assert cli.persist_infra_t0_rows(store, preds, "infer_adhoc", "fp") == 2 + assert len(store.load("infer_adhoc", "fp")) == 2 + store.close() + + def test_dry_run_computes_fingerprint_without_llm(monkeypatch, tmp_path, capsys): """--dry-run：diag_fingerprint 被调用、Phase 1 走空错题早返回、不真调 LLM。""" fp_calls: list[tuple[str, str, str]] = []