Video-Tree-TRM5/app/harness/validate.py

"""async 块序贯验证编排 — CE-Gate 局部验证的唯一独立子编排器。

从 TRM4 core/harness/validate.py (626 行) 迁移，重大重构：
- 同步 → async（run_inference 注入为 async callable）
- _classify_quadrants → core.evolution.classify_quadrants 纯函数
- 配对逻辑 → 复用 core.evolution.pair_block + 本地证据行组装
- _load_run_rows / _candidate_correctness_from_db → 共享 log.query()
- materialize_candidate_skill 保持同步（纯文件操作）

基线与候选在同一阶梯前缀上逐块配对，只数翻转（基线错→候选对 = W，
基线对→候选错 = L），每块结束调 gate_decision 做四出口判定。
基线侧逐题对错走 BaselineCache 内容寻址缓存，miss 才新鲜跑。
判定逻辑全部在 core/evolution/gate，本模块只负责推理编排与证据收集。
"""

from __future__ import annotations

import asyncio
import json
import shutil
import tempfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable

from loguru import logger

from app.harness.gate_ladder import BaselineCache, skill_hash
from app.harness.question_units import build_units, flatten_units, unit_correctness_view
from core.evolution import (
    INFRA_STOP_REASONS,
    GateParams,
    GateVerdict,
    RejectedEdit,
    classify_quadrants,
    gate_decision,
    pair_block,
)

# INFRA_STOP_REASONS 复用 core.evolution.diagnose 的单一定义（M-2）：INFRA 故障
# stop_reason（推理侧基础设施错误，非模型答错）在诊断与 gate 两处必须同口径，
# 避免各自维护副本致未来漂移。

if TYPE_CHECKING:
    from app.harness.inference import InferenceResult
    from app.harness.log import HarnessLog
    from core.types import GeneratedQuestion, QuestionUnit


# gate_decision 的 decision → ValidationOutcome.stop_reason 映射
_STOP_REASON_BY_DECISION: dict[str, str] = {
    "accept_confirmed": "confirmed",
    "reject_directional": "directional",
    "reject_futility": "futility",
    "accept_provisional": "provisional",
    "reject_inertia": "inertia",
}


# ---------------------------------------------------------------------------
# 注入协议
# ---------------------------------------------------------------------------


@runtime_checkable
class RunInferenceFn(Protocol):
    """注入的推理函数协议。

    调用方（runner）负责绑定 llm、tool_dispatch_fn、prompt_builder、
    log、concurrency、max_steps、skill_mode 等共享依赖。
    validate 侧只传 questions、run_id、skills_dir 三个逐块变化的参数。
    """

    async def __call__(
        self,
        questions: list[GeneratedQuestion],
        *,
        run_id: str,
        skills_dir: Path,
    ) -> InferenceResult: ...


# ---------------------------------------------------------------------------
# 数据类型
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class InferenceRunConfig:
    """一次推理运行的配置三元组，把"如何跑推理"内聚成一组。

    字段:
        concurrency: 推理并发度。
        max_steps: 单题最大推理步数。
        skill_mode: 推理 skill 模式（"auto" / "manual" / "none"）。
    """

    concurrency: int
    max_steps: int
    skill_mode: str


@dataclass
class ValidationOutcome:
    """CE-Gate 局部验证结果：三态动作 + e-process 证据（单元口径）+ 逐题溯源对错。

    correctness 二轨语义：W/L、准确率、四象限均按 **unit 口径** 统计
    （AR pair 双向 AND 折叠为一个单元，不被 P/Q 单题计分污染）；
    candidate_correctness 独立保留 **逐题** 对错（只含已观测题，早停后是阶梯前缀
    子集），accept 时由 runner 按 question_id 粒度增量合并进 state.correctness。
    """

    action: str  # accept_confirmed | accept_provisional | reject
    accepted: bool
    stop_reason: str  # confirmed | directional | futility | provisional | inertia
    e_value: float
    w: int
    l: int  # noqa: E741
    n_used: int
    delta_hat: float
    delta_shrunk: float
    baseline_acc: float  # 已观测单元上的基线准确率（unit 口径）
    candidate_acc: float  # 已观测单元上的候选准确率（unit 口径）
    improvements: list[str] = field(default_factory=list)
    regressions: list[str] = field(default_factory=list)
    persistent_fails: list[str] = field(default_factory=list)
    stable_successes: list[str] = field(default_factory=list)
    candidate_correctness: dict[str, bool] = field(default_factory=dict)
    evidence_rows: list[dict] = field(default_factory=list)  # gate_evidence 逐题行，runner 落库


@dataclass
class Probation:
    """一个题型的在途试用账本（每题型至多一个）。

    字段:
        task_type: 题型。
        anchor_skills_version: 锚版本名（最近一个 CONFIRMED 的 skills 版本）——
            回滚时恢复该版本中本题型 skill 文件的内容。
        target_file: 该题型解析后的 skill 文件名。
        correctness_snapshot: 开账时该题型 val 题的对错快照（回滚时恢复）。
        opened_step: 开账时的 global_step（观测用）。
        pending_edits: 试用链上全部候选 edit 的黑名单素材（回滚时整链入黑名单）。
    """

    task_type: str
    anchor_skills_version: str
    target_file: str
    correctness_snapshot: dict[str, bool]
    opened_step: int
    pending_edits: list[RejectedEdit] = field(default_factory=list)


# ---------------------------------------------------------------------------
# 同步辅助函数
# ---------------------------------------------------------------------------


def materialize_candidate_skill(
    workspace_dir: Path,
    base_skills_version: str,
    target_file: str,
    content: str,
) -> Path:
    """将候选 skill 正文物化为 workspace 专用临时目录下唯一命名的候选 skills 目录。

    复制基线 skills 目录到 .cand_tmp/ 下的唯一命名临时目录，然后覆写 target_file。
    构建失败时尽力清理已建临时目录再重抛原始异常。

    参数:
        workspace_dir: Workspace 根目录。基线 skills 从 workspace_dir/skills/<base>
            复制，临时候选落 workspace_dir/.cand_tmp/。
        base_skills_version: 基线 skills 版本名。
        target_file: 被替换的 skill 文件名。
        content: 候选 skill 文件全文。

    返回:
        新建的临时候选目录绝对路径。

    契约:
        构建失败（OSError）时尽力清理已建临时目录再重抛原始异常；
        清理本身失败记 warning。
    """
    cand_tmp_root = workspace_dir / ".cand_tmp"
    cand_tmp_root.mkdir(parents=True, exist_ok=True)
    cand_dir = Path(tempfile.mkdtemp(prefix=f"{base_skills_version}_cand_", dir=cand_tmp_root))
    try:
        base_dir = workspace_dir / "skills" / base_skills_version
        shutil.copytree(base_dir, cand_dir, dirs_exist_ok=True)
        (cand_dir / target_file).write_text(content, encoding="utf-8")
    except OSError:
        try:
            shutil.rmtree(cand_dir)
        except OSError as cleanup_err:
            logger.warning("候选物化失败后清理临时目录也失败 {}: {}", cand_dir, cleanup_err)
        raise
    return cand_dir


def _load_run_rows(
    log: HarnessLog,
    run_id: str,
) -> dict[str, dict[str, Any]]:
    """读取单个 run 的逐题预测行并规范化轨迹字段。

    从 predictions 表读取指定 run 的题目级记录，补充 _correct
    与规范化后的 steps 字段。保持同步（log.query）——仅在推理完成后调用。

    参数:
        log: HarnessLog 共享实例（用 query 方法做只读 SELECT）。
        run_id: 待读取的预测 run_id。

    返回:
        以 question_id 为键的行字典。每行至少包含 prediction、answer、
        _correct、steps 等字段。
    """
    rows = log.query(
        "SELECT question_id, prediction, answer, stop_reason, steps_json "
        "FROM predictions WHERE run_id=?",
        (run_id,),
    )
    normalized: dict[str, dict[str, Any]] = {}
    for row in rows:
        raw_steps = row.get("steps_json")
        parsed_steps: Any = raw_steps
        if isinstance(raw_steps, str):
            try:
                parsed_steps = json.loads(raw_steps)
            except json.JSONDecodeError:
                parsed_steps = []
        steps = parsed_steps if isinstance(parsed_steps, list) else []
        normalized[row["question_id"]] = {
            **row,
            "_correct": row.get("prediction") == row.get("answer"),
            "steps": steps,
        }
    return normalized


def _infra_question_ids_from_db(
    log: HarnessLog,
    run_id: str,
    chunk: list[GeneratedQuestion],
) -> set[str]:
    """从 db 读取一个 run 中 stop_reason 属 INFRA 故障族的 question_id 集合。

    参数:
        log: HarnessLog 共享实例。
        run_id: 推理 run_id。
        chunk: 题目列表。

    返回:
        stop_reason ∈ {"error", "parse_error"} 的 question_id 集合。
    """
    rows = _load_run_rows(log, run_id)
    return {
        q.question_id
        for q in chunk
        if rows.get(q.question_id, {}).get("stop_reason") in INFRA_STOP_REASONS
    }


def _count_infra_units(units: list[QuestionUnit], infra_qids: set[str]) -> int:
    """统计含 INFRA record 的 unit 数（一个 unit 任一题 INFRA 即计 1）。

    使护栏分子与分母（r.total，unit 粒度）同口径：AR pair 一 unit 含两 record，
    逐 record 计数会放大分子致 gate_guard_err 误触发，破坏 unit 粒度一致性
    （核心算法保真 #5/#6）。

    参数:
        units: 当前块的单元列表（single 或 AR pair）。
        infra_qids: 本 run 中 stop_reason 属 INFRA 故障族的 question_id 集合。

    返回:
        含至少一题 INFRA 的 unit 数。
    """
    return sum(1 for u in units if any(q.question_id in infra_qids for q in u.questions))


def _candidate_correctness_from_db(
    log: HarnessLog,
    run_id: str,
    chunk: list[GeneratedQuestion],
) -> dict[str, bool]:
    """从 db 读取候选/基线 run 在指定题目上的逐题对错。

    参数:
        log: HarnessLog 共享实例。
        run_id: 推理 run_id。
        chunk: 题目列表。

    返回:
        question_id -> 是否答对的映射。缺行的题目记为 False。
    """
    rows = _load_run_rows(log, run_id)
    return {q.question_id: rows.get(q.question_id, {}).get("_correct", False) for q in chunk}


# ---------------------------------------------------------------------------
# 块级 async 函数
# ---------------------------------------------------------------------------


async def _resolve_baseline_block(
    units: list[QuestionUnit],
    task_type: str,
    s_hash: str,
    prompts_version: str,
    baseline_cache: BaselineCache,
    base_skills_dir: Path,
    run_inference: RunInferenceFn,
    log: HarnessLog,
    run_id: str,
) -> tuple[dict[str, bool], list[QuestionUnit], int, int]:
    """基线侧处理一个块：缓存优先（unit 键），miss 的单元新鲜跑基线版本并回写缓存。

    缓存以 unit_id 为键、存单元级对错（AR pair 双向 AND 折叠后一个布尔）。
    miss 的单元展开为逐题送推理，读回逐题预测后经 unit_correctness_view 折叠成
    单元级对错再写缓存（核心算法保真 #5）。逐题 predictions 仍逐题落库溯源。

    INFRA 隔离（算法 #6）：miss 单元内**任一题** stop_reason ∈ {error, parse_error}
    即判定该单元为 INFRA 故障——**不写 BaselineCache**（否则瞬时故障永久污染基线
    快照）、**不入 b_units**、并从返回的有效单元集中剔除，避免污染 W/L 翻转与配对。
    命中缓存的单元恒为有效（此前已成功验证过）。

    参数:
        units: 当前块的单元列表（single 或 AR pair）。
        task_type: 当前验证题型（缓存键成分）。
        s_hash: 基线侧生效 skill 的内容哈希（缓存键成分）。
        prompts_version: 当前 prompts 版本（缓存键成分）。
        baseline_cache: 基线侧单元级对错缓存（键含 unit_id）。
        base_skills_dir: 基线 skills 版本目录。
        run_inference: 注入的 async 推理函数。
        log: HarnessLog 共享实例（推理后读预测）。
        run_id: 本块基线 run_id。

    返回:
        (b_units, valid_units, errors_inc, denom_inc)：块内有效 unit_id -> 基线单元
        对错、剔除 INFRA 后的有效单元列表、本块新增的 INFRA error 计数与推理题次
        分母增量（全命中时为 0, 0）。
    """
    miss_units = [
        u
        for u in units
        if baseline_cache.get(task_type, s_hash, prompts_version, u.unit_id) is None
    ]
    errors_inc = 0
    denom_inc = 0
    infra_qids: set[str] = set()
    if miss_units:
        miss_questions = flatten_units(miss_units)
        r_b = await run_inference(miss_questions, run_id=run_id, skills_dir=base_skills_dir)
        infra_qids = _infra_question_ids_from_db(log, r_b.run_id, miss_questions)
        # 护栏分子与分母（r.total，unit 粒度）同口径：含 INFRA record 的 unit 计 1，
        # 避免 AR pair（一 unit 两 record）逐 record 计数放大分子致误触发；仍涵盖
        # error + parse_error（_infra_question_ids_from_db 口径），parse_error 风暴不被绕过。
        errors_inc = _count_infra_units(miss_units, infra_qids)
        denom_inc = r_b.total
        fresh_per_q = _candidate_correctness_from_db(log, r_b.run_id, miss_questions)
        fresh_units = unit_correctness_view(miss_units, fresh_per_q)
        # 只回写非 INFRA 单元；INFRA 单元不入缓存（不永久污染基线快照）
        for u in miss_units:
            if any(q.question_id in infra_qids for q in u.questions):
                continue
            baseline_cache.put(task_type, s_hash, prompts_version, u.unit_id, fresh_units[u.unit_id])

    valid_units = [
        u for u in units if not any(q.question_id in infra_qids for q in u.questions)
    ]

    b_units: dict[str, bool] = {}
    for u in valid_units:
        val = baseline_cache.get(task_type, s_hash, prompts_version, u.unit_id)
        assert val is not None, f"基线缓存补齐后仍有 miss: unit={u.unit_id} run_id={run_id}"
        b_units[u.unit_id] = val
    return b_units, valid_units, errors_inc, denom_inc


async def _run_candidate_block(
    units: list[QuestionUnit],
    cand_dir: Path,
    run_inference: RunInferenceFn,
    log: HarnessLog,
    run_id: str,
) -> tuple[dict[str, bool], int, int]:
    """候选侧处理一个块：单元展开为逐题全块新鲜跑候选版本并从 db 读逐题对错。

    返回逐题对错映射（question_id -> bool），折叠为单元视图交由调用方完成，
    逐题结果同时用于 candidate_correctness 溯源与二轨 correctness 合并。

    参数:
        units: 当前块的单元列表。
        cand_dir: 已物化的候选 skills 目录。
        run_inference: 注入的 async 推理函数。
        log: HarnessLog 共享实例（推理后读预测）。
        run_id: 本块候选 run_id。

    返回:
        (c_per_q, errors_inc, denom_inc)：块内 question_id -> 候选对错。
    """
    questions = flatten_units(units)
    r_c = await run_inference(questions, run_id=run_id, skills_dir=cand_dir)
    c_per_q = _candidate_correctness_from_db(log, r_c.run_id, questions)
    infra_qids = _infra_question_ids_from_db(log, r_c.run_id, questions)
    # 护栏分子与分母（r.total，unit 粒度）同口径：含 INFRA record 的 unit 计 1
    # （见 _count_infra_units），涵盖 error + parse_error。
    errors_inc = _count_infra_units(units, infra_qids)
    return c_per_q, errors_inc, r_c.total


def _build_evidence_rows(
    units: list[QuestionUnit],
    b_units: dict[str, bool],
    c_units: dict[str, bool],
    task_type: str,
    block_idx: int,
) -> list[dict]:
    """组装一个块的 gate_evidence 单元级证据行。

    证据行按 unit 口径（question_id 字段存 unit_id、correct 存单元级对错），
    与 e-process 判定同粒度；逐题预测明细仍在 predictions 表逐题溯源。
    e_value 留 None 待块判定后回填，stop_reason 留空串待终态回填。

    参数:
        units: 当前块的单元列表。
        b_units: 块内 unit_id -> 基线单元对错。
        c_units: 块内 unit_id -> 候选单元对错。
        task_type: 当前验证题型。
        block_idx: 当前块序号。

    返回:
        单元级证据行列表。
    """
    return [
        {
            "question_id": u.unit_id,
            "task_type": task_type,
            # 落库列已更名 ladder_rank(阶梯序号);旧块路径此处值仍为块号,
            # 仅键名对齐 gate_evidence 表结构以保持落库兼容。
            "ladder_rank": block_idx,
            "baseline_correct": b_units[u.unit_id],
            "candidate_correct": c_units[u.unit_id],
            "e_value": None,
            "stop_reason": "",
        }
        for u in units
    ]


# ---------------------------------------------------------------------------
# INFRA 护栏
# ---------------------------------------------------------------------------


def _check_infra_guard(errors: int, infra_denom: int, gate_guard_err: float) -> None:
    """跨块累计 INFRA 错误率护栏：分母 >=10 且超阈值时 raise。

    参数:
        errors: 两侧累计 error 计数。
        infra_denom: 两侧累计推理题次分母。
        gate_guard_err: 错误率阈值。

    异常:
        RuntimeError: 错误率超阈值。
    """
    if infra_denom >= 10 and errors / infra_denom > gate_guard_err:
        raise RuntimeError(f"gate 推理累计错误率过高 {errors / infra_denom:.0%}，中止本轮")


# ---------------------------------------------------------------------------
# 终态组装
# ---------------------------------------------------------------------------


def _finalize_outcome(
    verdict: GateVerdict,
    w: int,
    l: int,  # noqa: E741
    n_used: int,
    n_plan: int,
    base_obs: dict[str, bool],
    cand_obs: dict[str, bool],
    candidate_per_q: dict[str, bool],
    evidence_rows: list[dict],
    task_type: str,
) -> ValidationOutcome:
    """将块循环终态判定组装为 ValidationOutcome。

    四象限/准确率/W/L 均按单元口径（base_obs/cand_obs 为 unit_id -> bool），
    candidate_correctness 独立保留逐题溯源（供 runner 二轨合并进 state.correctness）。

    参数:
        verdict: 最后一块的 gate 判定结果。
        w: 累计 W（基线错→候选对单元翻转）。
        l: 累计 L（基线对→候选错单元翻转）。
        n_used: 已消费的阶梯单元数。
        n_plan: 阶梯总单元数。
        base_obs: 累计基线已观测单元对错（unit_id -> bool）。
        cand_obs: 累计候选已观测单元对错（unit_id -> bool）。
        candidate_per_q: 累计候选逐题对错（question_id -> bool，溯源用）。
        evidence_rows: 单元级证据行。
        task_type: 验证题型（日志用）。

    返回:
        ValidationOutcome。
    """
    action = {
        "accept_confirmed": "accept_confirmed",
        "accept_provisional": "accept_provisional",
    }.get(verdict.decision, "reject")
    stop_reason = _STOP_REASON_BY_DECISION[verdict.decision]
    # 只有终态单元的证据行才携带 stop_reason
    evidence_rows[-1]["stop_reason"] = stop_reason

    quadrants = classify_quadrants({uid: (base_obs[uid], cand_obs[uid]) for uid in base_obs})
    baseline_acc = sum(base_obs.values()) / len(base_obs)
    candidate_acc = sum(cand_obs.values()) / len(cand_obs)
    accepted = action != "reject"

    logger.info(
        "gate 局部验证[{}]: 基线{:.1%} → 候选{:.1%} (W={} L={} E={:.2f} n={}/{} 单元) {}",
        task_type,
        baseline_acc,
        candidate_acc,
        w,
        l,
        verdict.e_value,
        n_used,
        n_plan,
        "接受" if accepted else "回滚",
    )

    return ValidationOutcome(
        action=action,
        accepted=accepted,
        stop_reason=stop_reason,
        e_value=verdict.e_value,
        w=w,
        l=l,
        n_used=n_used,
        delta_hat=verdict.delta_hat,
        delta_shrunk=verdict.delta_shrunk,
        baseline_acc=baseline_acc,
        candidate_acc=candidate_acc,
        improvements=quadrants.improvements,
        regressions=quadrants.regressions,
        persistent_fails=quadrants.persistent_fails,
        stable_successes=quadrants.stable_successes,
        candidate_correctness=candidate_per_q,
        evidence_rows=evidence_rows,
    )


# ---------------------------------------------------------------------------
# 主编排
# ---------------------------------------------------------------------------


def _ladder_units(ladder_items: list[GeneratedQuestion]) -> list[QuestionUnit]:
    """把阶梯题序聚合为单元并保持信息阶梯顺序（按单元最早出现位置排序）。

    build_units 会把 single 与 pair 分组重排（single 先、pair 后），破坏"难题优先"
    的阶梯序；此处按单元内题目在 ladder 中的最早下标重排，恢复原阶梯优先级，
    保证 AR pair 折叠不改变 e-process 的出题顺序（核心算法保真 #5）。非 AR 全 single
    时排序为恒等（unit_id 等于 question_id、位置即原序），与迁移前逐题行为一致。

    参数:
        ladder_items: 阶梯出题序（可混含 single 与 AR pair 成员）。

    返回:
        按阶梯序排列的单元列表。
    """
    units = build_units(ladder_items)
    position = {q.question_id: i for i, q in enumerate(ladder_items)}
    units.sort(key=lambda u: min(position[q.question_id] for q in u.questions))
    return units


async def _run_local_validation(
    workspace_dir: Path,
    cand_dir: Path,
    base_skills_version: str,
    task_type: str,
    base_skill_content: str,
    units: list[QuestionUnit],
    gate_params: GateParams,
    gate_block: int,
    gate_guard_err: float,
    baseline_cache: BaselineCache,
    prompts_version: str,
    run_inference: RunInferenceFn,
    log: HarnessLog,
    gate_run_prefix: str,
) -> ValidationOutcome:
    """块序贯循环主体：逐块基线（缓存优先）/候选按单元配对推理，块间 e-process 判定。

    按 gate_block 切**单元**前缀（AR pair 整锁在同一块，不跨块拆分），每块先补齐
    基线侧缓存 miss（新鲜跑基线版本并按 unit_id 写 BaselineCache），再全块跑候选，
    折叠成单元视图后配对累计 W/L 调 gate_decision；非 continue 即早停。单元尽时
    最后一块的判定即终态（n_remaining=0 走 provisional/inertia 分支），无循环外补判。

    参数:
        workspace_dir: Workspace 根目录。
        cand_dir: 已物化的候选 skills 目录。
        base_skills_version: 基线 skills 版本名。
        task_type: 当前验证题型。
        base_skill_content: 基线侧生效 skill 全文（skill_hash 作缓存键成分）。
        units: 已截断到 gate_n_max 的阶梯单元序（single 或 AR pair）。
        gate_params: e-process 判据阈值组。
        gate_block: 块大小（单位为**单元数**）。
        gate_guard_err: 跨块累计 INFRA 错误率护栏（分母 >=10 才触发）。
        baseline_cache: 基线侧单元级对错缓存（键含 unit_id）。
        prompts_version: 当前 prompts 版本（缓存键成分）。
        run_inference: 注入的 async 推理函数。
        log: HarnessLog 共享实例。
        gate_run_prefix: 块 run_id 前缀（含 "_gate_" 标记）。

    返回:
        ValidationOutcome。

    关键实现:
        INFRA 护栏跨块累计基线+候选两侧的 error 计数，分母（总推理题次，仍逐题计）
        >=10 且错误率超 gate_guard_err 时直接 raise，避免坏批次污染判定。
    """
    w = 0
    l = 0  # noqa: E741
    n_used = 0
    n_excluded = 0  # 累计被 INFRA 隔离剔除的单元数（从阶梯分母扣除）
    errors = 0
    infra_denom = 0
    evidence_rows: list[dict] = []
    base_obs: dict[str, bool] = {}
    cand_obs: dict[str, bool] = {}
    candidate_per_q: dict[str, bool] = {}
    s_hash = skill_hash(base_skill_content)
    base_skills_dir = workspace_dir / "skills" / base_skills_version
    unit_chunks = [units[i : i + gate_block] for i in range(0, len(units), gate_block)]
    n_plan = len(units)
    verdict: GateVerdict | None = None

    for block_idx, unit_chunk in enumerate(unit_chunks):
        # Phase 1: 基线侧（缓存优先，miss 新鲜跑，INFRA 单元剔除）
        b_units, valid_chunk, err_b, den_b = await _resolve_baseline_block(
            units=unit_chunk,
            task_type=task_type,
            s_hash=s_hash,
            prompts_version=prompts_version,
            baseline_cache=baseline_cache,
            base_skills_dir=base_skills_dir,
            run_inference=run_inference,
            log=log,
            run_id=f"{gate_run_prefix}_b{block_idx}_base",
        )
        # 本块全 INFRA：无有效单元可配对——候选无需空跑，仅把基线侧错误计入护栏后
        # 累计剔除数进入下一块（护栏仍能在整轮 INFRA 错误率超阈值时熔断）。
        n_excluded += len(unit_chunk) - len(valid_chunk)
        if not valid_chunk:
            errors += err_b
            infra_denom += den_b
            _check_infra_guard(errors, infra_denom, gate_guard_err)
            continue

        # 候选侧只跑基线侧判定有效（非 INFRA）的单元，保证配对 unit_ids 两侧一致
        c_per_q, err_c, den_c = await _run_candidate_block(
            units=valid_chunk,
            cand_dir=cand_dir,
            run_inference=run_inference,
            log=log,
            run_id=f"{gate_run_prefix}_b{block_idx}_cand",
        )

        # Phase 2: INFRA 护栏（跨块累计，分母 >=10 才触发）——写缓存前置于此已由
        # _resolve_baseline_block 保证 INFRA 单元不落缓存，此处仅做整轮错误率熔断。
        errors += err_b + err_c
        infra_denom += den_b + den_c
        _check_infra_guard(errors, infra_denom, gate_guard_err)

        # Phase 3: 折叠成单元视图 + 配对 + 证据行 + 块间判定（均用有效单元）
        c_units = unit_correctness_view(valid_chunk, c_per_q)
        candidate_per_q.update(c_per_q)
        unit_ids = [u.unit_id for u in valid_chunk]
        pair_result = pair_block(b_units, c_units, unit_ids)
        for uid, (b, c) in pair_result.observed.items():
            base_obs[uid] = b
            cand_obs[uid] = c

        block_rows = _build_evidence_rows(valid_chunk, b_units, c_units, task_type, block_idx)

        w += pair_result.w
        l += pair_result.l  # noqa: E741
        n_used += len(valid_chunk)
        # 阶梯剩余按扣除 INFRA 后的有效分母计：n_remaining = (n_plan - n_excluded) - n_used
        verdict = gate_decision(w, l, n_used, (n_plan - n_excluded) - n_used, params=gate_params)

        for row in block_rows:
            row["e_value"] = verdict.e_value
        evidence_rows.extend(block_rows)

        if verdict.decision != "continue":
            break

    # verdict 仍为 None ⟺ 全部单元被 INFRA 排除（空 ladder 已在入口拒绝）。
    # 明确失败，避免落到误导性的"空阶梯"断言而无法定位为 INFRA 原因。
    if verdict is None:
        raise RuntimeError("gate 阶梯所有 unit 被判为 INFRA 排除，无法验证（检查推理基础设施）")
    # 最后一块判定即终态（n_remaining=0 → provisional/inertia）
    return _finalize_outcome(
        verdict=verdict,
        w=w,
        l=l,
        n_used=n_used,
        n_plan=n_plan,
        base_obs=base_obs,
        cand_obs=cand_obs,
        candidate_per_q=candidate_per_q,
        evidence_rows=evidence_rows,
        task_type=task_type,
    )


async def validate_skill_local(
    workspace_dir: Path,
    base_skills_version: str,
    task_type: str,
    target_file: str,
    candidate_content: str,
    base_skill_content: str,
    ladder_items: list[GeneratedQuestion],
    gate_params: GateParams,
    gate_block: int,
    gate_n_max: int,
    gate_guard_err: float,
    baseline_cache: BaselineCache,
    prompts_version: str,
    run_inference: RunInferenceFn,
    log: HarnessLog,
    gate_run_prefix: str,
) -> ValidationOutcome:
    """块序贯配对验证：阶梯出题，基线/候选逐块配对，e-process 四出口早停。

    参数:
        workspace_dir: workspace 根目录。
        base_skills_version: 基线 skills 版本名（候选物化复制源）。
        task_type: 待验证题型。
        target_file: fallback 解析后该题型的真实生效 skill 文件名
            （record.target_file，可能是共享 default-strategy.md）；
            候选物化写此文件，与 accept 路径同源。
        candidate_content: 候选 skill 全文。
        base_skill_content: 基线侧该题型解析后生效 skill 文件全文
            （skill_hash(base_skill_content) 作 BaselineCache 键成分）。
        ladder_items: 阶梯序题目列表（已排除本 step 案例包题）。
        gate_params: e-process 判据阈值组。
        gate_block: 块大小（单位为**单元数**，AR pair 整锁不跨块拆）。
        gate_n_max: 单 gate 单元数上限（阶梯截断到此数量个单元）。
        gate_guard_err: 跨块累计 INFRA 错误率护栏（分母 >=10 才触发）。
        baseline_cache: 基线侧单元级对错缓存（键含 unit_id）。
        prompts_version: 当前 prompts 版本（缓存键成分）。
        run_inference: 注入的 async 推理函数（RunInferenceFn 协议）。
        log: HarnessLog 共享实例（供 DB 回读逐题对错）。
        gate_run_prefix: gate 内推理 run_id 前缀，必须含 "_gate_"
            （防泄露过滤靠它识别）。块 run_id = f"{prefix}_b{block_idx}_{arm}"。

    返回:
        ValidationOutcome。单元级证据记入 outcome.evidence_rows 随结果返回，
        gate_evidence 落库由调用方（runner）负责。
    """
    if "_gate_" not in gate_run_prefix:
        raise ValueError(f"gate_run_prefix 必须含 '_gate_'（防泄露过滤依赖）: {gate_run_prefix!r}")
    if not ladder_items:
        raise ValueError(f"task_type={task_type} 阶梯为空，无法验证")

    # 阶梯题序聚合为单元并按信息阶梯序截断到 gate_n_max 个单元（AR pair 整锁不拆）
    units = _ladder_units(ladder_items)[:gate_n_max]
    cand_dir = materialize_candidate_skill(
        workspace_dir, base_skills_version, target_file, candidate_content
    )
    try:
        return await _run_local_validation(
            workspace_dir=workspace_dir,
            cand_dir=cand_dir,
            base_skills_version=base_skills_version,
            task_type=task_type,
            base_skill_content=base_skill_content,
            units=units,
            gate_params=gate_params,
            gate_block=gate_block,
            gate_guard_err=gate_guard_err,
            baseline_cache=baseline_cache,
            prompts_version=prompts_version,
            run_inference=run_inference,
            log=log,
            gate_run_prefix=gate_run_prefix,
        )
    finally:
        try:
            shutil.rmtree(cand_dir)
        except OSError as e:
            logger.warning("候选临时目录清理失败 {}: {}", cand_dir, e)


# ---------------------------------------------------------------------------
# 连续并发 gate:数据结构 + 前缀消费(algo #6 语义修订:块序贯 → 阶梯序前缀逐对序贯)
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class GateSpec:
    """单题型 gate 验证规格(runner 装配阶段产物,调度器输入)。

    字段:
        task_type: 题型。
        target_file: 解析后生效 skill 文件名(候选物化写此文件)。
        candidate_content: 候选 skill 全文。
        base_skill_content: 基线侧生效 skill 全文(skill_hash 作缓存键)。
        units: 阶梯序单元元组(已排除案例单元、截断 gate_n_max);元组,装配后
            不可变,防 spec.units 与 run.slots 漂移。
        gate_run_prefix: run_id 前缀,必须含 "_gate_"(防泄露过滤依赖)。
    """

    task_type: str
    target_file: str
    candidate_content: str
    base_skill_content: str
    units: tuple[QuestionUnit, ...]
    gate_run_prefix: str


@dataclass
class _UnitSlot:
    """单个阶梯单元的双臂到达状态。

    base 为单元级对错(AR pair 已折叠);cand_per_q 为逐题对错(折叠交给消费时,
    以复用 unit_correctness_view 并保留逐题溯源)。INFRA 标志与结果互斥。
    """

    unit: QuestionUnit
    base: bool | None = None
    cand_per_q: dict[str, bool] | None = None
    base_infra: bool = False
    cand_infra: bool = False

    def resolved(self) -> bool:
        """双臂均已出结果(含 INFRA 判定)。

        返回:
            base 臂(结果或 INFRA)与 cand 臂(结果或 INFRA)都已到达时为 True。
        """
        base_done = self.base is not None or self.base_infra
        cand_done = self.cand_per_q is not None or self.cand_infra
        return base_done and cand_done

    def excluded(self) -> bool:
        """任一臂 INFRA 即整单元剔除(不入配对)。

        返回:
            base_infra 或 cand_infra 任一为 True 时为 True。
        """
        return self.base_infra or self.cand_infra


@dataclass
class _GateRun:
    """单题型 gate 的运行时状态(计数器 + 前缀指针 + 证据)。"""

    spec: GateSpec
    slots: list[_UnitSlot]
    s_hash: str
    prefix_ptr: int = 0
    w: int = 0
    l: int = 0  # noqa: E741
    n_used: int = 0
    n_excluded: int = 0
    errors: int = 0
    infra_denom: int = 0
    frozen: bool = False
    verdict: GateVerdict | None = None
    base_obs: dict[str, bool] = field(default_factory=dict)
    cand_obs: dict[str, bool] = field(default_factory=dict)
    candidate_per_q: dict[str, bool] = field(default_factory=dict)
    evidence_rows: list[dict] = field(default_factory=list)

    @classmethod
    def from_spec(cls, spec: GateSpec) -> _GateRun:
        """由规格构造初始状态(slots 与阶梯序一一对应)。

        参数:
            spec: 单题型 gate 规格(units 已阶梯序)。

        返回:
            计数器归零、slots 逐单元初始化、s_hash 已计算的 _GateRun。
        """
        return cls(
            spec=spec,
            slots=[_UnitSlot(unit=u) for u in spec.units],
            s_hash=skill_hash(spec.base_skill_content),
        )


def _advance_prefix(run: _GateRun, params: GateParams) -> None:
    """沿阶梯序消费"已配齐前缀",逐单元更新 (W,L) 并判定,过线即冻结。

    统计合法性关键(设计 v3 §1 / Codex C1):严禁按到达序消费——base 臂缓存命中
    瞬间返回、cand 臂必新鲜跑,两臂延迟不对称,若 cand 延迟与对错相关,早到翻转
    对系统性偏向 W 型 → e-值虚高假接受。前缀消费把判定顺序钉回预声明阶梯序,
    anytime-valid 无条件成立;INFRA 单元视为"已解决(剔除)"不阻塞前缀。

    契约:全部单元被剔除时 verdict 保持 None、frozen 保持 False,由调度编排层
    (Task 3 的 validate_skills_concurrent)检测 verdict None 并 raise
    RuntimeError;本函数不负责该终态。

    参数:
        run: 单题型 gate 运行时状态(原地更新计数器/指针/证据)。
        params: e-process 判据阈值组。

    返回:
        无(所有效果原地写入 run;可重复调用,已消费前缀不重复消费)。
    """
    while not run.frozen and run.prefix_ptr < len(run.slots):
        slot = run.slots[run.prefix_ptr]
        if not slot.resolved():
            return
        rank = run.prefix_ptr
        run.prefix_ptr += 1
        if slot.excluded():
            run.n_excluded += 1
            # 剔除使 n_remaining 缩小,必须重判(Codex plan 审 C1):否则尾部全 INFRA
            # 时 verdict 停留在 "continue",绕过题尽第四出口且 _finalize_outcome
            # 查 stop_reason 映射 KeyError。n_used==0(纯前导 INFRA)时无证据可判,跳过。
            if run.n_used > 0:
                n_remaining = (len(run.slots) - run.n_excluded) - run.n_used
                run.verdict = gate_decision(run.w, run.l, run.n_used, n_remaining, params=params)
                if run.verdict.decision != "continue":
                    run.frozen = True
            continue
        uid = slot.unit.unit_id
        assert slot.base is not None and slot.cand_per_q is not None, (
            f"slot 未配齐即被消费: unit={slot.unit.unit_id}"
        )
        c_units = unit_correctness_view([slot.unit], slot.cand_per_q)
        pair_result = pair_block({uid: slot.base}, c_units, [uid])
        run.candidate_per_q.update(slot.cand_per_q)
        for u, (b, c) in pair_result.observed.items():
            run.base_obs[u] = b
            run.cand_obs[u] = c
        run.w += pair_result.w
        run.l += pair_result.l
        run.n_used += 1
        n_remaining = (len(run.slots) - run.n_excluded) - run.n_used
        run.verdict = gate_decision(run.w, run.l, run.n_used, n_remaining, params=params)
        run.evidence_rows.append(
            {
                "question_id": uid,
                "task_type": run.spec.task_type,
                "ladder_rank": rank,
                "baseline_correct": slot.base,
                "candidate_correct": c_units[uid],
                "e_value": run.verdict.e_value,
                "stop_reason": "",
            }
        )
        if run.verdict.decision != "continue":
            run.frozen = True


class _QuestionSlots:
    """按题数计数的共享并发闸:峰值在飞请求恒 ≤ width(设计 v3 §2.4)。

    多槽获取(AR pair 一单元两题)经内部锁串行化,防多任务半持有交错死锁。
    本类只承诺"并发上限 + 多槽获取原子性";公平性由调用方按题型 round-robin
    顺序创建任务实现(实践中 asyncio 等待队列近似先来先服务,但那不是本类契约)。
    """

    def __init__(self, width: int) -> None:
        """初始化题槽闸。

        参数:
            width: 并发宽度(全 gate 同时在飞的题数上限),必须为正。

        返回:
            无。

        关键实现细节:
            _width 供 acquire 做超宽 fail-fast;BoundedSemaphore 使多还立即
            ValueError 而非静默扩容;_acquire_lock 串行化多槽获取防交错死锁。
        """
        assert width > 0, f"并发宽度必须为正: {width}"
        self._width = width
        # BoundedSemaphore:多还立即 ValueError 而非静默扩容(Codex 质量审 3)
        self._sem = asyncio.BoundedSemaphore(width)
        self._acquire_lock = asyncio.Lock()

    async def acquire(self, n: int) -> None:
        """原子获取 n 个题槽。

        fail-fast:n > 宽度时任务持锁等待永不满足的槽位 → 自死锁
        (AR pair 单元 2 题 + width=1 的病态配置,Codex plan 审 C2),直接报错。
        取消安全:半持有自动回滚——逐槽获取途中被取消(或任何 BaseException)
        时,已拿到的 permit 全部归还再重抛,容量不泄漏(Codex 质量审 2)。

        参数:
            n: 申请的题槽数(单元内题目数,single=1 / AR pair=2)。

        返回:
            无(成功返回即持有 n 个槽,须与 release(n) 配对)。

        异常:
            ValueError: n 超过并发宽度(否则自死锁)。
        """
        if n > self._width:
            raise ValueError(f"单次申请题槽 {n} 超过并发宽度 {self._width},将自死锁")
        async with self._acquire_lock:
            got = 0
            try:
                for _ in range(n):
                    await self._sem.acquire()
                    got += 1
            except BaseException:
                for _ in range(got):
                    self._sem.release()
                raise

    def release(self, n: int) -> None:
        """归还 n 个题槽。

        参数:
            n: 与 acquire 对应的题槽数。

        返回:
            无。

        关键实现细节:
            底层为 BoundedSemaphore——多还(release 数超过 acquire)立即
            ValueError 暴露调用方配对错误,属防御性设计。
        """
        for _ in range(n):
            self._sem.release()


async def _run_unit_arm(
    run: _GateRun,
    slot_idx: int,
    arm: str,
    slots: _QuestionSlots,
    run_inference: RunInferenceFn,
    log: HarnessLog,
    baseline_cache: BaselineCache,
    prompts_version: str,
    base_skills_dir: Path,
    cand_dir: Path,
    gate_params: GateParams,
    gate_guard_err: float,
) -> None:
    """执行一个 (单元, 臂) 任务:缓存/推理 → 到达登记 → 前缀消费推进。

    冻结检查三次:启动时(排队任务撤销点)、获得题槽后(获槽期间被冻结)、
    推理返回后(τ 之后的 in-flight 结果不计入,整体丢弃)。
    base 臂缓存命中不占题槽(零推理);INFRA 单元不写缓存(不永久污染基线快照)。
    护栏在每次臂完成时检查(等价迁移自跨块累计,设计 v3 §2.3),超阈值 raise
    中止整轮(与现行行为一致)。

    参数:
        run: 该题型的 gate 运行时状态。
        slot_idx: 单元在阶梯中的下标。
        arm: "base" 或 "cand"。
        slots: 全 gate 共享题槽闸。
        run_inference: 注入推理函数。
        log: HarnessLog 共享实例(推理后读预测)。
        baseline_cache / prompts_version: 基线缓存及键成分。
        base_skills_dir / cand_dir: 两臂各自的 skills 目录。
        gate_params: e-process 判据(前缀消费用)。
        gate_guard_err: INFRA 错误率护栏阈值。

    返回:
        无(结果写入 run.slots[slot_idx] 并触发 _advance_prefix)。

    异常:
        RuntimeError: 累计 INFRA 错误率超护栏阈值(经 _check_infra_guard)。
    """
    assert arm in ("base", "cand")
    if run.frozen:
        return
    slot = run.slots[slot_idx]
    spec = run.spec

    if arm == "base":
        cached = baseline_cache.get(spec.task_type, run.s_hash, prompts_version, slot.unit.unit_id)
        if cached is not None:
            slot.base = cached
            _advance_prefix(run, gate_params)
            return

    questions = list(slot.unit.questions)
    await slots.acquire(len(questions))
    try:
        if run.frozen:
            return
        run_id = f"{spec.gate_run_prefix}_{arm}"
        skills_dir = base_skills_dir if arm == "base" else cand_dir
        r = await run_inference(questions, run_id=run_id, skills_dir=skills_dir)
        # 推理 await 期间该题型可能已被其他任务判定冻结:设计语义是
        # "τ(冻结时刻)之后的 in-flight 结果不计入"——整体丢弃,不写
        # slot/infra_denom/errors,滞后 INFRA 也不得触发护栏 raise 掀翻
        # 整轮 gather(Codex 质量审 1)。
        if run.frozen:
            return
        _register_arm_arrival(
            run=run,
            slot=slot,
            arm=arm,
            questions=questions,
            inference_run_id=r.run_id,
            inference_total=r.total,
            log=log,
            baseline_cache=baseline_cache,
            prompts_version=prompts_version,
        )
        _check_infra_guard(run.errors, run.infra_denom, gate_guard_err)
    finally:
        slots.release(len(questions))
    _advance_prefix(run, gate_params)


def _register_arm_arrival(
    run: _GateRun,
    slot: _UnitSlot,
    arm: str,
    questions: list[GeneratedQuestion],
    inference_run_id: str,
    inference_total: int,
    log: HarnessLog,
    baseline_cache: BaselineCache,
    prompts_version: str,
) -> None:
    """把一次臂推理结果登记进 slot 与 run 计数器(INFRA 判定 + 对错折叠 + 回写缓存)。

    INFRA 臂只标记不写缓存(不永久污染基线快照);正常 base 臂折叠为单元级对错并
    回写 BaselineCache,正常 cand 臂保留逐题对错(折叠交给前缀消费,保留逐题溯源)。

    参数:
        run: 该题型的 gate 运行时状态(errors / infra_denom 原地累加)。
        slot: 本单元的双臂到达状态(结果或 INFRA 标志原地写入)。
        arm: "base" 或 "cand"。
        questions: 本单元展开后的题目列表。
        inference_run_id: 本次推理的 run_id(DB 回读键)。
        inference_total: 本次推理的题次数(护栏分母增量)。
        log: HarnessLog 共享实例(推理后读预测)。
        baseline_cache / prompts_version: 基线缓存及键成分。

    返回:
        无(所有效果原地写入 run 与 slot)。

    关键实现细节:
        errors 按单元级去重(Codex plan 审 I3):同一单元双臂都 INFRA 只计 1 个
        error,与设计 §2.3"分子=INFRA 单元数(任一臂)"及旧块实现口径一致
        (旧实现 cand 不跑 base-INFRA 单元,天然无双计)。
    """
    spec = run.spec
    infra_qids = _infra_question_ids_from_db(log, inference_run_id, questions)
    run.infra_denom += inference_total
    if infra_qids:
        if not slot.excluded():
            run.errors += 1
        if arm == "base":
            slot.base_infra = True
        else:
            slot.cand_infra = True
        return
    per_q = _candidate_correctness_from_db(log, inference_run_id, questions)
    if arm == "base":
        folded = unit_correctness_view([slot.unit], per_q)
        slot.base = folded[slot.unit.unit_id]
        baseline_cache.put(
            spec.task_type, run.s_hash, prompts_version, slot.unit.unit_id, slot.base
        )
    else:
        slot.cand_per_q = per_q


def _validate_gate_specs(specs: list[GateSpec]) -> None:
    """校验各题型 gate 规格,不合法直接报错(不兜底)。

    参数:
        specs: 各题型 gate 规格。

    异常:
        ValueError: 阶梯为空,或 gate_run_prefix 缺 "_gate_"(防泄露过滤依赖
            该标记识别 gate run)。
    """
    for spec in specs:
        if "_gate_" not in spec.gate_run_prefix:
            raise ValueError(f"gate_run_prefix 必须含 '_gate_': {spec.gate_run_prefix!r}")
        if not spec.units:
            raise ValueError(f"task_type={spec.task_type} 阶梯为空,无法验证")


def _cleanup_candidate_dirs(cand_dirs: dict[str, Path]) -> None:
    """尽力清理全部候选临时目录,单个失败只记 warning 不中断其余清理。

    参数:
        cand_dirs: task_type -> 候选临时目录路径。

    返回:
        无。
    """
    for d in cand_dirs.values():
        try:
            shutil.rmtree(d)
        except OSError as e:
            logger.warning("候选临时目录清理失败 {}: {}", d, e)


def _build_launch_order(runs: list[_GateRun]) -> list[tuple[_GateRun, int, str]]:
    """构建 (run, rank, arm) 发射队列:题型 round-robin × 题型内阶梯序。

    交错顺序 = rank 0 各题型 → rank 1 各题型 → ...;同一 (题型, rank) 内
    base 先 cand 后。round-robin 让各题型的阶梯头部同批起跑,配合前缀消费
    使统计推进不因某题型阶梯过长而饿死其他题型。

    参数:
        runs: 各题型 gate 运行时状态(slots 已按阶梯序初始化)。

    返回:
        (run, rank, arm) 三元组列表,即任务创建顺序。
    """
    order: list[tuple[_GateRun, int, str]] = []
    max_rank = max((len(r.slots) for r in runs), default=0)
    for rank in range(max_rank):
        for r in runs:
            if rank < len(r.slots):
                for arm in ("base", "cand"):
                    order.append((r, rank, arm))
    return order


async def validate_skills_concurrent(
    workspace_dir: Path,
    base_skills_version: str,
    specs: list[GateSpec],
    gate_params: GateParams,
    gate_guard_err: float,
    baseline_cache: BaselineCache,
    prompts_version: str,
    run_inference: RunInferenceFn,
    log: HarnessLog,
    concurrency: int,
) -> dict[str, ValidationOutcome]:
    """连续并发 gate:多题型全部臂共享题槽并发,统计按阶梯序前缀有序推进。

    发射顺序 = 题型 round-robin × 题型内阶梯序(base 先 cand 后);题型过线即
    冻结,其排队任务启动时自查冻结标志撤销,in-flight 结果不计入(τ 之后样本,
    合法丢弃)。全部题型判定后统一组装 ValidationOutcome。

    参数:
        workspace_dir: workspace 根目录(候选物化用)。
        base_skills_version: 基线 skills 版本名。
        specs: 各题型 gate 规格(units 已阶梯序 + 截断 n_max)。
        gate_params: e-process 判据阈值组。
        gate_guard_err: INFRA 错误率护栏阈值。
        baseline_cache: 基线侧单元级对错缓存。
        prompts_version: 当前 prompts 版本(缓存键成分)。
        run_inference: 注入推理函数(调用方须绑定共享 HarnessLog)。
        log: HarnessLog 共享实例(推理后读预测,与 run_inference 同库)。
        concurrency: 题槽宽度(峰值在飞题数上限)。

    返回:
        {task_type: ValidationOutcome}。

    异常:
        RuntimeError: INFRA 护栏超阈值,或某题型全部单元被 INFRA 排除。
        ValueError: spec 校验失败(空阶梯 / run_prefix 缺 "_gate_")。
    """
    _validate_gate_specs(specs)
    base_skills_dir = workspace_dir / "skills" / base_skills_version
    runs = [_GateRun.from_spec(s) for s in specs]
    cand_dirs: dict[str, Path] = {}
    slots_gate = _QuestionSlots(concurrency)
    try:
        # 成功一个登记一个:第 N 个题型物化抛 OSError 时,已登记的前 N-1 个
        # 目录仍由 finally 统一清理,不泄漏(Codex 质量审 C001)。
        for r in runs:
            cand_dirs[r.spec.task_type] = materialize_candidate_skill(
                workspace_dir, base_skills_version, r.spec.target_file, r.spec.candidate_content
            )
        tasks = [
            asyncio.ensure_future(
                _run_unit_arm(
                    r,
                    rank,
                    arm,
                    slots_gate,
                    run_inference,
                    log,
                    baseline_cache,
                    prompts_version,
                    base_skills_dir,
                    cand_dirs[r.spec.task_type],
                    gate_params,
                    gate_guard_err,
                )
            )
            for r, rank, arm in _build_launch_order(runs)
        ]
        # 护栏 raise 中止整轮的语义不变(Codex 质量审 C002):首异常先取消其余
        # 任务并排水(return_exceptions 吞取消回报),确保外层 finally 删除候选
        # 目录时已无在飞任务访问该目录、事件循环收尾无 pending task 警告;
        # _run_unit_arm 的题槽获取自带取消回滚,cancel 安全。
        try:
            await asyncio.gather(*tasks)
        except BaseException:
            for t in tasks:
                t.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            raise
    finally:
        _cleanup_candidate_dirs(cand_dirs)

    outcomes: dict[str, ValidationOutcome] = {}
    for r in runs:
        if r.verdict is None:
            raise RuntimeError(
                f"gate[{r.spec.task_type}] 全部 unit 被判为 INFRA 排除,无法验证(检查推理基础设施)"
            )
        outcomes[r.spec.task_type] = _finalize_outcome(
            verdict=r.verdict,
            w=r.w,
            l=r.l,
            n_used=r.n_used,
            n_plan=len(r.slots),
            base_obs=r.base_obs,
            cand_obs=r.cand_obs,
            candidate_per_q=r.candidate_per_q,
            evidence_rows=r.evidence_rows,
            task_type=r.spec.task_type,
        )
    return outcomes