"""结果驱动视频级切分的自包含两阶段 CLI 入口。

把整条离线管线的编排从 shell 搬进 Python：一次调用内联串起
    Phase 1 离线诊断（run_baseline_diagnosis，LLM 重活，断点续跑幂等）→
    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 <ISO>` 显式固定，以对
      manifest 做字节级复现比对。write_manifest 库内不调 datetime.now，时间戳一律由
      本 CLI 传入。

设计要点：
    - 诊断口径指纹 = (诊断 prompt 版本, 模型名, git 短 SHA) 三分量合成，隔离不同
      诊断配置的信号；换 prompt / 模型 / 代码实现即换指纹，旧信号不被覆盖。
    - 真实依赖组装参考 app/harness/runner.py::_run_diagnosis：GovernedLLMClient
      (search llm, thinking=True) + RunLogImpl(harness.db) + VersionedSkillStore +
      DiagnosePrompts(项目根 prompts/) + tree_data 按 wrong_ids 涉及 video 预加载
      （store/videos/<vid>/tree.json 展平）。
    - 缺 .env / config 关键项一律 fail loud（P5），绝不静默兜底。
    - `--dry-run` 用假 deps 跑通两阶段 wiring 不真调 LLM，打印将执行的步骤 + 指纹，
      用于校验装配正确性（对齐 CLAUDE.md §2.5 smoke test）。

编排函数（run_pipeline）通过依赖注入接收 DiagnosisDeps / signal_store / wrong_ids /
questions / canonical_preds，便于单测用假实现替换、不触真实 LLM 与 harness.db。
其中 canonical_preds 供 Phase 0 补 INFRA / 空预测错题的 T0 信号（这些题不进诊断）。
"""

from __future__ import annotations

import argparse
import asyncio
import datetime
import os
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any

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,
    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
    from core.evolution.protocols import DiagnosisSignalStore
    from core.types import GeneratedQuestion

# 与 core.evolution.diagnose._INFRA_STOP_REASONS 对齐：执行/解析层失败排除出可诊断错题。
_INFRA_STOP_REASONS: frozenset[str] = frozenset({"error", "parse_error"})

# 工程路径默认值（少变；可经 CLI 单次覆盖）。诊断信号表建在 harness.db。
_DEFAULT_HARNESS_DB = Path("workspaces/default/harness.db")
_DEFAULT_QUESTIONS_DIR = Path("store/questions/benchmarks/Video-MME")
_DEFAULT_OUT_DIR = Path("workspaces/video-split")
_DEFAULT_STORE_DIR = Path("store")  # tree.json 在 store/videos/<vid>/


# ---------------------------------------------------------------------------
# 配置解析（fail loud）
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class VideoSplitConfig:
    """结果驱动视频级切分的科研旋钮快照（从 config/video_split.yaml 解析）。

    字段:
        baseline_run_id: 基线 run 标识（错题诊断与切分依据）。
        n_trainval: trainval 目标视频数（多样性阶段填充上限）。
        epsilon: test 相对全局最大允许分布偏差（题型 / 难度两维）。
        report_floor: per-type 报告门限，题数 ≥ 此值的 task_type 才入 ε 约束。
        val_wrong_min: validation 池最少错题数（McNemar 功效阈；0=不检查）。
        val_ratio: validation 占 trainval 视频组总数的比例。
        seed: 贪心选择器预洗牌 + 视频组题级切分种子。
        floor_k: 各高信号 task_type 的 T2 defect 下限（硬约束）。
        prompt_version: 诊断 prompt 版本标识（指纹分量）。
        model: 执行诊断的模型名（指纹分量）。
    """

    baseline_run_id: str
    n_trainval: int
    epsilon: float
    report_floor: int
    val_wrong_min: int
    val_ratio: float
    seed: int
    floor_k: dict[str, int]
    prompt_version: str
    model: str


def _require(section: dict[str, Any], keys: tuple[str, ...], where: str) -> None:
    """校验 section 含全部必填键，缺任一即 fail loud（P5，不静默兜底）。

    参数:
        section: 待校验的配置子字典。
        keys: 必填键元组。
        where: 出错信息中标注的段名（如 "video_split"）。

    异常:
        SystemExit: 存在缺失键。
    """
    missing = [k for k in keys if k not in section]
    if missing:
        raise SystemExit(f"config {where} 段缺关键项 {missing}，无法运行（P5 fail loud）")


def parse_config(raw: dict[str, Any]) -> VideoSplitConfig:
    """把 yaml 原始字典解析为 VideoSplitConfig，缺关键项 fail loud。

    参数:
        raw: yaml.safe_load 的顶层字典，需含 video_split / diag 两段。

    返回:
        VideoSplitConfig 冻结快照。

    异常:
        SystemExit: 缺 video_split / diag 段或段内关键项。
    """
    if "video_split" not in raw or "diag" not in raw:
        raise SystemExit("config 缺 video_split / diag 段，无法运行（P5 fail loud）")
    vs = raw["video_split"]
    dg = raw["diag"]
    _require(
        vs,
        (
            "baseline_run_id",
            "n_trainval",
            "epsilon",
            "report_floor",
            "val_wrong_min",
            "val_ratio",
            "seed",
            "floor_k",
        ),
        "video_split",
    )
    _require(dg, ("prompt_version", "model"), "diag")
    return VideoSplitConfig(
        baseline_run_id=vs["baseline_run_id"],
        n_trainval=vs["n_trainval"],
        epsilon=vs["epsilon"],
        report_floor=vs["report_floor"],
        val_wrong_min=vs["val_wrong_min"],
        val_ratio=vs["val_ratio"],
        seed=vs["seed"],
        floor_k=dict(vs["floor_k"]),
        prompt_version=dg["prompt_version"],
        model=dg["model"],
    )


def load_config(config_path: Path) -> VideoSplitConfig:
    """读取并解析 video_split yaml 配置文件（缺文件 / 关键项 fail loud）。

    参数:
        config_path: yaml 配置路径。

    返回:
        VideoSplitConfig。

    异常:
        SystemExit: 文件不存在或缺关键项。
    """
    if not config_path.exists():
        raise SystemExit(f"config 文件不存在: {config_path}（P5 fail loud）")
    raw = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    return parse_config(raw)


def git_short_sha() -> str:
    """取当前 git 短 SHA 作为诊断口径指纹的代码分量（诊断代码变则指纹变）。

    返回:
        git rev-parse --short HEAD 输出（去空白）。

    异常:
        SystemExit: 非 git 仓库或 git 不可用（fail loud，指纹不可缺分量）。
    """
    try:
        out = subprocess.run(
            ["git", "rev-parse", "--short", "HEAD"],
            capture_output=True,
            text=True,
            check=True,
        )
    except (subprocess.CalledProcessError, FileNotFoundError) as exc:
        raise SystemExit(f"无法获取 git 短 SHA 作为诊断代码版本: {exc}（P5 fail loud）") from exc
    sha = out.stdout.strip()
    if not sha:
        raise SystemExit("git rev-parse --short HEAD 返回空，诊断指纹缺代码分量（P5 fail loud）")
    return sha


# ---------------------------------------------------------------------------
# 真实依赖组装（参考 runner.py::_run_diagnosis）
# ---------------------------------------------------------------------------


class _DiagLLMSettings:
    """诊断 LLM 的工程配置（从 .env 读取 search llm 凭证 + 韧性旋钮）。

    仅承载诊断所需字段（搜索 LLM = 诊断 judge），不复用 main.InfraSettings 以免
    构造整套适配器（embed / vlm）的重活；缺关键凭证 fail loud。
    """

    def __init__(self) -> None:
        from pydantic_settings import BaseSettings, SettingsConfigDict

        class _Settings(BaseSettings):
            model_config = SettingsConfigDict(env_file=".env", extra="ignore")

            search_llm_model: str = ""
            search_llm_base_url: str = ""
            search_llm_api_key: str = ""
            redis_url: str = ""
            redis_cache_ttl: int = 86400
            llm_timeout: float = 300.0
            llm_max_retries: int = 3
            llm_retry_base_delay: float = 20.0
            llm_retry_max_delay: float = 120.0
            llm_circuit_breaker_threshold: int = 48
            llm_circuit_breaker_cooldown: float = 60.0
            llm_ttft_timeout: float = 30.0
            llm_inter_token_timeout: float = 15.0

        self._s = _Settings()

    def __getattr__(self, name: str) -> Any:
        return getattr(self._s, name)


def _build_redis_cache(settings: Any) -> Any | None:
    """按 .env redis_url 构建响应缓存（不可用则降级 None，与 main 一致）。"""
    if not settings.redis_url:
        return None
    try:
        import redis.asyncio as aioredis

        from adapters.redis_cache import RedisResponseCache

        redis_client = aioredis.from_url(settings.redis_url, decode_responses=True)
        ttl_s = settings.redis_cache_ttl if settings.redis_cache_ttl > 0 else None
        return RedisResponseCache(redis=redis_client, ttl_s=ttl_s)
    except Exception:
        logger.warning("Redis 缓存不可用，诊断降级为无缓存模式")
        return None


def build_diagnosis_deps(
    *,
    harness_db: Path,
    store_dir: Path,
    video_ids: list[str],
    concurrency: int,
    expected_model: str,
) -> DiagnosisDeps:
    """组装 Phase 1 诊断的真实依赖束（GovernedLLMClient + RunLogImpl + prompts）。

    与 runner.py::_run_diagnosis 对齐：search LLM（thinking=True）作诊断 judge，
    RunLogImpl 只读读取 harness.db 的 predictions/traces，VersionedSkillStore 读技能，
    DiagnosePrompts 从项目根 prompts/ 加载，tree_data 按 video_ids 从
    store/videos/<vid>/tree.json 展平预加载（诊断需真实树，缺失即 fail-loud）。

    参数:
        harness_db: harness.db 路径（诊断读预测 + 信号落库同库）。
        store_dir: store 根目录（含 videos/<vid>/tree.json）。
        video_ids: wrong_ids 涉及的 video 标识列表（可含重复，内部去重加载树）。
        concurrency: 诊断并发上限。
        expected_model: config.diag.model（诊断口径指纹的模型分量）。必须与 .env
            SEARCH_LLM_MODEL 一致——指纹里的 model 与实际诊断所用 model 不一致会让
            信号以错误模型指纹落库，破坏可复现 / resume / 口径隔离，故此处 fail loud。

    返回:
        DiagnosisDeps 冻结依赖束。

    异常:
        SystemExit: .env 缺 search LLM 凭证（model / base_url / api_key 任一为空），
            或 config.diag.model 与 .env SEARCH_LLM_MODEL 不一致（指纹漂移防护）。
        FileNotFoundError / ValueError: video_ids 中任一 video 的 tree.json 缺失或
            无效（load_tree_data_for_videos fail-loud，诊断需真实树）。
    """
    from adapters.breaker import CircuitBreaker
    from adapters.llm import GovernedLLMClient
    from adapters.telemetry import SQLiteTelemetryRecorder
    from app.harness.log import RunLogImpl
    from app.harness.workspace import VersionedSkillStore

    settings = _DiagLLMSettings()
    if not (
        settings.search_llm_model and settings.search_llm_base_url and settings.search_llm_api_key
    ):
        raise SystemExit(
            "诊断 LLM 凭证缺失：.env 需配置 SEARCH_LLM_MODEL / SEARCH_LLM_BASE_URL / "
            "SEARCH_LLM_API_KEY（P5 fail loud，不静默兜底）"
        )
    if expected_model != settings.search_llm_model:
        raise SystemExit(
            "诊断模型指纹漂移：config.diag.model="
            f"{expected_model!r} 与 .env SEARCH_LLM_MODEL={settings.search_llm_model!r} "
            "不一致；指纹里的 model 必须等于实际诊断所用 model（P5 fail loud，"
            "请对齐 config/video_split.yaml diag.model 与 .env SEARCH_LLM_MODEL）"
        )

    telemetry_db = Path("logs/telemetry.db")
    telemetry_db.parent.mkdir(parents=True, exist_ok=True)
    telemetry = SQLiteTelemetryRecorder(telemetry_db)
    breaker = CircuitBreaker(
        fail_threshold=max(settings.llm_circuit_breaker_threshold, 1),
        cooldown_s=settings.llm_circuit_breaker_cooldown,
    )
    llm = GovernedLLMClient(
        model=settings.search_llm_model,
        base_url=settings.search_llm_base_url,
        api_key=settings.search_llm_api_key,
        provider=settings.search_llm_model.split("-")[0],
        thinking=True,
        breaker=breaker,
        cache=_build_redis_cache(settings),
        telemetry=telemetry,
        timeout_s=settings.llm_timeout,
        ttft_timeout_s=settings.llm_ttft_timeout,
        inter_token_timeout_s=settings.llm_inter_token_timeout,
        max_retries=settings.llm_max_retries,
        retry_base_delay_s=settings.llm_retry_base_delay,
        retry_max_delay_s=settings.llm_retry_max_delay,
    )
    from app.harness.tree_nodes import load_tree_data_for_videos

    return DiagnosisDeps(
        run_log=RunLogImpl(str(harness_db)),
        llm=llm,
        skill_store=VersionedSkillStore(_diagnosis_skills_dir()),
        prompts=_load_diagnose_prompts(),
        tree_data=load_tree_data_for_videos(store_dir, video_ids),
        concurrency=concurrency,
    )


def _diagnosis_skills_dir() -> Path:
    """诊断用技能目录：种子 store 的当前技能版本（诊断读技能遵从判定）。

    诊断只读技能内容判断"是否遵从技能"，用 store 种子 v1 即可（与基线 run 一致）。
    """
    return Path("store/skills/v1")


def _load_diagnose_prompts() -> Any:
    """加载诊断模板束（从项目根 prompts/ 读取；与 runner._load_diagnose_prompts 一致）。"""
    from core.evolution.types import DiagnosePrompts

    def _read(name: str) -> str:
        p = Path("prompts") / name
        if not p.exists():
            raise FileNotFoundError(f"缺进化/诊断模板: {p}（请从 TRM4 迁移或检查 prompts/）")
        return p.read_text(encoding="utf-8")

    return DiagnosePrompts(
        defect_vs_lapse=_read("defect_vs_lapse.md"),
        reasoning_sub=_read("reasoning_sub.md"),
        span_eval_system=_read("span_eval_system.md"),
        missed_nodes=_read("missed_nodes.md"),
        skill_adherence=_read("skill_adherence.md"),
        confirmation_bias=_read("confirmation_bias.md"),
        evidence_sufficiency=_read("evidence_sufficiency.md"),
    )


def select_diagnosable_wrong_ids(preds: list[dict]) -> list[str]:
    """从 canonical 预测筛出可诊断错题 question_id（保序）。

    可诊断错题判据：预测非空 且 stop_reason 非 INFRA（error / parse_error）且
    归一后预测 != 答案。INFRA / 空预测错题不进 wrong_ids——它们改由
    persist_infra_t0_rows 直接落 T0（run_diagnosis 内部也会二次排除同类题）。

    参数:
        preds: load_canonical_predictions 产出的 canonical 预测行（已按 qid 去重）。

    返回:
        可诊断错题 question_id 列表（保 preds 顺序）。
    """
    wrong_ids: list[str] = []
    for pred in preds:
        prediction = (pred["prediction"] or "").strip()
        if not prediction or pred["stop_reason"] in _INFRA_STOP_REASONS:
            continue
        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 映射。

    覆盖 wrong_ids 与 run_diagnosis 返回的全部 infra/degraded 题（取 video_id/task_type）。

    参数:
        questions_dir: benchmark 题库目录。

    返回:
        question_id → GeneratedQuestion 映射。
    """
    return {q.question_id: q for q in load_benchmark(questions_dir)}


# ---------------------------------------------------------------------------
# McNemar 功效护栏
# ---------------------------------------------------------------------------


def check_mcnemar_power(pools: Pools, val_wrong_min: int) -> int:
    """校验 validation 池错题数达 McNemar 功效阈，不足即 fail loud。

    build_split 按契约用 split_by_video_assignment(val_wrong_min=0)（不破契约），
    故功效护栏在 capstone 层单独核验：val 错题数 < 阈 → 验证信号不足以支撑可靠比较。

    参数:
        pools: 冻结三池（含 validation 与 correctness）。
        val_wrong_min: 最少错题数阈（0 = 不检查）。

    返回:
        validation 池实际错题数（供日志）。

    异常:
        SystemExit: val_wrong_min > 0 且 val 错题数 < 阈（P5 fail loud，不静默放行）。
    """
    val_wrong = sum(1 for q in pools.validation if not pools.correctness[q.question_id])
    if val_wrong_min > 0 and val_wrong < val_wrong_min:
        raise SystemExit(
            f"validation 池错题数 {val_wrong} < val_wrong_min={val_wrong_min}，"
            "验证信号不足以支撑可靠比较（McNemar 检验功效不够）。"
            "请放大 val_ratio / 调整旋钮后重跑，勿静默放行。"
        )
    return val_wrong


# ---------------------------------------------------------------------------
# 两阶段编排（依赖注入，便于单测）
# ---------------------------------------------------------------------------


async def run_pipeline(
    *,
    config: VideoSplitConfig,
    fingerprint: str,
    diagnosis_deps: DiagnosisDeps,
    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 0 INFRA T0 补录 → Phase 1 诊断 → Phase 2 冻结切分 → McNemar 护栏。

    参数:
        config: 科研旋钮快照。
        fingerprint: 诊断口径指纹（已合成，作诊断信号主键之一）。
        diagnosis_deps: Phase 1 诊断依赖束（真实或假实现）。
        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 字符串，由调用方传入；见模块 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)
    )
    await run_baseline_diagnosis(
        baseline_run_id=config.baseline_run_id,
        diag_fingerprint=fingerprint,
        wrong_ids=wrong_ids,
        questions=questions,
        store=signal_store,
        deps=diagnosis_deps,
    )

    # Phase 2: 冻结切分（读诊断信号 → 贪心选择 → 视频组原子切三池 → 冻结 + 六条断言）。
    out_dir.mkdir(parents=True, exist_ok=True)
    logger.info("Phase 2 冻结切分：out={}", out_dir)
    result = build_split(
        db_path=harness_db,
        baseline_run_id=config.baseline_run_id,
        signal_store=signal_store,
        diag_fingerprint=fingerprint,
        questions_dir=questions_dir,
        config=SplitBuildConfig(
            n_trainval=config.n_trainval,
            floor_k=config.floor_k,
            epsilon=config.epsilon,
            report_floor=config.report_floor,
            select_seed=config.seed,
            val_ratio=config.val_ratio,
            split_seed=config.seed,
        ),
        out_path=out_dir / "pools.json",
        manifest_path=out_dir / "split_manifest.json",
        generated_at=generated_at,
    )

    # McNemar 功效护栏（build_split 契约外的 capstone 层校验）。
    val_wrong = check_mcnemar_power(result.pools, config.val_wrong_min)
    logger.info(
        "切分冻结完成：pools={} manifest={} val错题={}/{}(阈)",
        out_dir / "pools.json",
        out_dir / "split_manifest.json",
        val_wrong,
        config.val_wrong_min,
    )
    return result


# ---------------------------------------------------------------------------
# 真实执行 / dry-run 入口
# ---------------------------------------------------------------------------


def _resolve_paths(args: argparse.Namespace) -> tuple[Path, Path, Path, Path]:
    """解析 harness_db / questions_dir / out_dir / store_dir（CLI 覆盖默认工程路径）。"""
    harness_db = args.harness_db or _DEFAULT_HARNESS_DB
    questions_dir = args.questions_dir or _DEFAULT_QUESTIONS_DIR
    out_dir = args.out_dir or _DEFAULT_OUT_DIR
    store_dir = args.store_dir or _DEFAULT_STORE_DIR
    return harness_db, questions_dir, out_dir, store_dir


def _execute_real(config: VideoSplitConfig, fingerprint: str, args: argparse.Namespace) -> None:
    """真实执行两阶段管线：组装真实 deps、读错题、跑诊断 + 冻结切分。"""
    harness_db, questions_dir, out_dir, store_dir = _resolve_paths(args)
    if not harness_db.exists():
        raise SystemExit(f"harness.db 不存在: {harness_db}（P5 fail loud）")
    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)
    video_ids: list[str] = []
    for qid in wrong_ids:
        q = questions.get(qid)
        if q is None:
            raise SystemExit(
                f"wrong_id {qid!r} 不在 questions_dir 题库中"
                "（baseline predictions 与题库不匹配，P5 fail loud）"
            )
        video_ids.append(q.video_id)
    deps = build_diagnosis_deps(
        harness_db=harness_db,
        store_dir=store_dir,
        video_ids=video_ids,
        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))
    try:
        asyncio.run(
            run_pipeline(
                config=config,
                fingerprint=fingerprint,
                diagnosis_deps=deps,
                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=generated_at,
            )
        )
    finally:
        store.close()


class _DryRunLLM:
    """dry-run 假 LLM：被真实调用即报错，保证不真调 LLM。"""

    async def complete(self, *args: Any, **kwargs: Any) -> Any:
        raise AssertionError("dry-run 不应真调 LLM.complete")


class _DryRunLog:
    """dry-run 假 RunLog：predictions/traces 均返回空，诊断不真正执行。"""

    async def get_predictions(self, run_id: str, *, question_ids: list[str] | None = None) -> list:
        return []

    async def get_traces(self, run_id: str, *, question_ids: list[str] | None = None) -> list:
        return []


def _execute_dry_run(config: VideoSplitConfig, fingerprint: str, args: argparse.Namespace) -> None:
    """dry-run：用假 deps 跑通 Phase 1 wiring（空错题 → 诊断早返回），打印步骤 + 指纹。

    Phase 2 build_split 需真实诊断信号方能冻结，dry-run 不真实冻结，仅打印其计划；
    Phase 1 用空 wrong_ids 走 run_baseline_diagnosis 早返回路径，验证装配可调用而不触 LLM。
    """
    harness_db, questions_dir, out_dir, _store_dir = _resolve_paths(args)
    logger.info("=== dry-run：校验两阶段装配（不真调 LLM / 不冻结产物）===")
    logger.info(
        "诊断口径指纹 diag_fingerprint={} (prompt={} model={})",
        fingerprint,
        config.prompt_version,
        config.model,
    )
    logger.info(
        "解析路径：harness_db={} questions_dir={} out_dir={}", harness_db, questions_dir, out_dir
    )
    logger.info(
        "旋钮：n_trainval={} epsilon={} report_floor={} val_ratio={} seed={} "
        "val_wrong_min={} floor_k={}",
        config.n_trainval,
        config.epsilon,
        config.report_floor,
        config.val_ratio,
        config.seed,
        config.val_wrong_min,
        config.floor_k,
    )

    fake_deps = DiagnosisDeps(
        run_log=_DryRunLog(),
        llm=_DryRunLLM(),
        skill_store=object(),
        prompts=object(),
        tree_data={},
        concurrency=args.concurrency,
    )

    from adapters.baseline_diagnosis_store import SqliteDiagnosisSignalStore

    dry_db = out_dir / "_dry_run_signals.db"
    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(
                baseline_run_id=config.baseline_run_id,
                diag_fingerprint=fingerprint,
                wrong_ids=[],
                questions={},
                store=store,
                deps=fake_deps,
            )
        )
    finally:
        store.close()
        dry_db.unlink(missing_ok=True)
    logger.info(
        "Phase 2 装配 OK：真实执行将调 build_split 冻结 pools.json + manifest（dry-run 跳过）"
    )
    logger.info("=== dry-run 通过：两阶段装配可调用，指纹已算出 ===")


def build_arg_parser() -> argparse.ArgumentParser:
    """构建 CLI 参数解析器。"""
    parser = argparse.ArgumentParser(description="结果驱动视频级切分两阶段 CLI（诊断 → 冻结切分）")
    parser.add_argument("--config", type=Path, default=Path("config/video_split.yaml"))
    parser.add_argument("--dry-run", action="store_true", dest="dry_run")
    parser.add_argument("--gpu", type=str, default=None, help="可选：设置 CUDA_VISIBLE_DEVICES")
    parser.add_argument("--concurrency", type=int, default=8, help="诊断并发上限")
    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("--store-dir", type=Path, default=None, dest="store_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


def main(argv: list[str] | None = None) -> None:
    """CLI 入口：解析参数 → 载配置 → 算指纹 → dry-run 或真实两阶段执行。

    参数:
        argv: 可选参数列表（默认 sys.argv[1:]），便于测试注入。
    """
    from dotenv import load_dotenv

    load_dotenv()
    args = build_arg_parser().parse_args(argv)
    if args.gpu is not None:
        os.environ["CUDA_VISIBLE_DEVICES"] = args.gpu
        logger.info("CUDA_VISIBLE_DEVICES={}", args.gpu)

    config = load_config(args.config)
    fingerprint = diag_fingerprint(config.prompt_version, config.model, git_short_sha())

    if args.dry_run:
        _execute_dry_run(config, fingerprint, args)
        return
    _execute_real(config, fingerprint, args)


if __name__ == "__main__":
    main()