"""结果驱动视频级切分的顶层编排：诊断信号 → 冻结 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 json
import sqlite3
from collections import Counter, defaultdict
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: 视频组题级切分的洗牌种子。
        val_wrong_min: validation 池最少错题数，切分时保证功效（不足则从 diag 换入
            低 T2 错题组补足，耗尽 fail-loud）。
    """

    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
    val_wrong_min: 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 _unique_backup_path(path: Path, suffix: str) -> Path:
    """求 path 的唯一 .bak.<suffix> 备份路径，已存在则追加递增序号避免覆盖。

    首选 ``<name>.bak.<suffix>``；若已存在，退化为 ``<name>.bak.<suffix>.2``、
    ``.3`` … 直到找到不存在的名字。保证连续 forced freeze（同 suffix 或都缺
    manifest 用 'prev'）不会静默覆盖此前保留的备份。

    参数:
        path: 待备份的原文件路径。
        suffix: 备份后缀（旧 pools_sha256 前 8 位或 'prev'）。

    返回:
        目录内唯一、尚不存在的备份路径。
    """
    candidate = path.with_name(f"{path.name}.bak.{suffix}")
    counter = 2
    while candidate.exists():
        candidate = path.with_name(f"{path.name}.bak.{suffix}.{counter}")
        counter += 1
    return candidate


def _guard_frozen_products(out_path: Path, manifest_path: Path, *, force: bool) -> None:
    """冻结前的覆盖保护：产物已存在时按 force 决定报错或备份。

    参数:
        out_path: 目标 pools.json 路径。
        manifest_path: 目标 split_manifest.json 路径。
        force: False 时已存在即 FileExistsError；True 时把旧产物重命名为唯一的
            .bak.<旧 pools_sha256 前 8 位或 'prev'>（同名已存在则追加递增序号）再放行。

    异常:
        FileExistsError: force=False 且产物已存在（防静默覆盖冻结锚点）。
        OSError: 备份 rename 失败；已备份的文件先 rollback 回原名再抛出，保证
            要么两文件都备份、要么都不动（原子性，不留半备份的不一致目录）。
    """
    if not out_path.exists() and not manifest_path.exists():
        return
    if not force:
        raise FileExistsError(
            f"已存在冻结产物 {out_path}（或其 manifest）。重跑切分会覆盖训练依赖的"
            "冻结锚点——确认要替换请加 --force（旧产物将备份为 .bak.*）。"
        )
    # 备份后缀取旧 manifest 的 pools_sha256 前 8 位，无则用 'prev'
    suffix = "prev"
    if manifest_path.exists():
        try:
            old = json.loads(manifest_path.read_text(encoding="utf-8"))
            suffix = str(old.get("pools_sha256", "prev"))[:8] or "prev"
        except (json.JSONDecodeError, OSError):
            suffix = "prev"
    # 先为每个存在的文件求唯一备份路径（互不冲突），再逐个 rename；
    # 中途失败则把已备份的 rollback 回原名，保证原子性。
    to_backup = [p for p in (out_path, manifest_path) if p.exists()]
    done: list[tuple[Path, Path]] = []  # (备份路径, 原路径)，供 rollback
    try:
        for p in to_backup:
            dst = _unique_backup_path(p, suffix)
            p.rename(dst)
            done.append((dst, p))
    except OSError:
        for backup_path, original in reversed(done):
            backup_path.rename(original)
        raise


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,
    force: bool = False,
) -> SplitBuildResult:
    """顶层编排结果驱动视频级切分，冻结 pools.json + manifest 并跑防御断言。

    步骤：读 canonical 基线预测 → 读诊断信号 → 构建全视频画像 → 贪心选择 trainval /
    test → 加载题库并以视频归属切三池 → 原子冻结 pools.json → 写溯源 manifest →
    六条防御断言 fail-fast 校验。

    契约（Task 11）：val_wrong_min 前置到切分内保证功效——build_split 计算
    wrong_tier_by_video 并连同 config.val_wrong_min 传入 split_by_video_assignment，
    切分时若 val 错题不足即从 diag 换入低 T2 错题组补足（耗尽 fail-loud）。CLI 的
    check_mcnemar_power 作切分冻结后的冗余最终确认。

    参数:
        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 字符串），由调用方传入以保证可复现。
        force: 覆盖保护开关。False（默认）时若 out_path/manifest_path 已存在即
            FileExistsError（防静默覆盖训练依赖的冻结锚点）；True 时先把旧产物备份为
            .bak.* 再放行覆盖。

    返回:
        SplitBuildResult，含 pools / manifest / assignment，支持字典式访问。

    异常:
        AssertionError: 六条防御断言任一不满足（fail-fast，不静默）。
        ValueError: 上游依赖校验失败（如 correctness 缺题、assignment 非法）。
    """
    # Phase 1: canonical 基线预测 + 诊断信号。
    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 = [
        {
            "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}
    tier_by_q = {row["question_id"]: row["tier"] for row in signal_rows}
    wrong_tier_by_video: dict[str, int] = defaultdict(int)
    for pred in preds:
        if not pred["correct"] and tier_by_q.get(pred["question_id"]) == "T2":
            wrong_tier_by_video[pred["video_id"]] += 1
    pools = split_by_video_assignment(
        questions,
        assignment,
        correctness,
        config.val_ratio,
        config.split_seed,
        baseline_run_id=baseline_run_id,
        val_wrong_min=config.val_wrong_min,
        wrong_tier_by_video=dict(wrong_tier_by_video),
    )
    _guard_frozen_products(out_path, manifest_path, force=force)
    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 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 = 预测与答案各自归一
    （strip → 大写 → 取首字母）后逐字符相等。

    口径边界：旧 build_or_load_pools 的 legacy 池构建路径（app/harness/pools.py）是
    另一条独立既有链路，不共用本 helper，两者刻意不统一（本次不动 legacy 路径）。

    参数:
        db_path: harness.db 路径（URI mode=ro 只读打开，绝不改动基线 db）。
        baseline_run_id: 基线 run 标识。

    返回:
        canonical 预测行列表，每行含 question_id / video_id / task_type /
        prediction / answer / stop_reason / correct（bool）。按 rowid 升序去重，
        每 qid 保留首行。

    异常:
        ValueError: 该 run 无任何预测行（fail-fast，不返回空切分）。

    实现细节:
        按 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, stop_reason "
            "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"],
            "prediction": row["prediction"],
            "answer": row["answer"],
            "stop_reason": row["stop_reason"],
            "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 /
        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]
    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_distribution, evolution_target_distribution = _signal_distributions(signal_rows_raw)

    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,
        "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,
    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}"
        )