feat(harness): correctness 三口径 + gate 块按 unit 跑

进化引擎与 gate e-process 从 question_id 口径迁至 unit_id 口径，AR pair
双向 AND 折叠为单元、不被 P/Q 单题计分污染；逐题 predictions 仅作溯源。

- question_units: 新增 unit_correctness_view(units, per_q)->dict[unit_id,bool]
  作为逐题→单元折叠的唯一入口（复用 unit_correctness）。
- core/evolution/validate: pair_block/compute_accuracy 参数改 unit_ids、
  分母按单元数（键即 unit_id）。
- app/harness/validate（gate 实际执行路径）：阶梯题序聚合为单元并保持信息
  阶梯序（_ladder_units），gate 块按单元切分（AR pair 整锁不跨块拆）；
  baseline_cache 键含 unit_id、存单元级对错；候选逐题读回后折叠成单元视图；
  n_used/W/L/四象限/准确率均按单元计；证据行按 unit 口径，candidate_correctness
  独立保留逐题对错供 runner 二轨合并。
- runner: probation 结算按 unit 折叠计 W/L（_probation_unit_flips）；quadrant
  四象限 id 承载 unit_id。

核心算法保真 #5（信息阶梯 e-process）：本次仅迁移 correctness 口径，不改冷启动
2:1 / gamma-EMA / 反泄漏算法本身（gate_ladder 迁移见 Task 8）。

app/harness/question_units.py

@@ -111,3 +111,25 @@ def unit_correctness(unit: QuestionUnit, per_q: dict[str, bool]) -> bool:
         强制上游先补齐全部单题结果再计单元正确性。
     """
     return all(per_q[q.question_id] for q in unit.questions)
+
+
+def unit_correctness_view(units: list[QuestionUnit], per_q: dict[str, bool]) -> dict[str, bool]:
+    """把逐题对错折叠成单元级视图：unit_id → 单元是否整体正确。
+
+    进化引擎（gate e-process / quadrant / probation / pair_block / compute_accuracy）
+    统一消费此单元视图，保证 AR pair 双向 AND、非 AR single 单题，混格池中
+    孪生对折叠为一个单元、不被 P/Q 单题计分污染（核心算法保真 #5）。
+
+    参数:
+        units: 目标单元列表（single 或 pair）。
+        per_q: 题目 question_id → 该题是否作答正确（唯一逐题溯源来源）。
+
+    返回:
+        unit_id → 单元级正确性。single 的 unit_id 等于其 question_id，
+        pair 的 unit_id 等于共享 pair_id。
+
+    关键实现:
+        逐单元复用 unit_correctness（内部以 per_q[q.question_id] 取值，缺任一题
+        触发 KeyError），禁静默兜底、强制上游先补齐全部单题结果。
+    """
+    return {u.unit_id: unit_correctness(u, per_q) for u in units}

app/harness/runner.py

@@ -42,6 +42,7 @@ from app.harness.observation import (
     write_shadow_gate,
     write_step_report,
 )
+from app.harness.question_units import build_units, unit_correctness_view
 from app.harness.store import advance_version
 from app.harness.validate import Probation, ValidationOutcome
 from app.harness.workspace import (
@@ -61,6 +62,7 @@ from core.evolution import (
     RejectedEdit,
     edit_budget_at,
     momentum_inner,
+    pair_block,
     probation_verdict,
     replace_momentum,
     resolve_skill_file,
@@ -72,11 +74,12 @@ if TYPE_CHECKING:
     from app.harness.pools import Pools
     from core.evolution.types import (
         EvolutionRecord,
+        PairResult,
         SystemCasePack,
         ToolCasePack,
     )
     from core.protocols import LLMProvider, TelemetryRecorder, VLMProvider
-    from core.types import GeneratedQuestion
+    from core.types import GeneratedQuestion, QuestionUnit
 
 
 class _InterruptError(RuntimeError):
@@ -267,15 +270,57 @@ def _compute_total_steps(pools: Pools, correctness: dict[str, bool], config: Run
     return config.epochs * steps_per_epoch
 
 
+def _probation_unit_flips(
+    val_units: list[QuestionUnit],
+    snapshot: dict[str, bool],
+    rows: dict[str, dict[str, Any]],
+    task_type: str,
+    eval_run_id: str,
+) -> PairResult:
+    """按 unit 折叠锚快照与当前重跑对错，返回单元级翻转统计（W/L）。
+
+    锚快照（开账时逐题对错）与当前全 val 重跑逐题对错各经 unit_correctness_view
+    折叠成单元视图，再走 pair_block 计单元翻转（AR pair 双向 AND，不被 P/Q 单题
+    计分污染，核心算法保真 #5）。
+
+    参数:
+        val_units: 该题型的 val 单元列表。
+        snapshot: 开账时逐题对错快照（question_id -> bool）。
+        rows: 当前全 val 重跑逐题预测行（question_id -> 规范化行）。
+        task_type: 题型（错误信息用）。
+        eval_run_id: 全 val 重跑 run_id（错误信息用）。
+
+    返回:
+        PairResult（单元级 W/L 与 observed）。
+
+    异常:
+        RuntimeError: 重跑缺某 val 题的预测行。
+    """
+    cur_per_q: dict[str, bool] = {}
+    for q in (q for u in val_units for q in u.questions):
+        row = rows.get(q.question_id)
+        if row is None:
+            raise RuntimeError(
+                f"probation 结算缺预测行: {task_type}/{q.question_id}（run={eval_run_id}）"
+            )
+        cur_per_q[q.question_id] = row["_correct"]
+    snap_units = unit_correctness_view(val_units, snapshot)
+    cur_units = unit_correctness_view(val_units, cur_per_q)
+    return pair_block(snap_units, cur_units, [u.unit_id for u in val_units])
+
+
 def _outcome_to_quadrant_pairs(task_type: str, outcome: ValidationOutcome) -> list[dict]:
-    """把 ValidationOutcome 的四象限拍平为逐题 pair（供 quadrant_pair 表落库观测）。
+    """把 ValidationOutcome 的四象限拍平为单元 pair（供 quadrant_pair 表落库观测）。
+
+    四象限 id 为 **unit_id 口径**（single 即 question_id、AR pair 为 pair_id），
+    与 gate e-process 同粒度；question_id 字段承载 unit_id。
 
     参数:
         task_type: 该批 gate 的任务类型。
         outcome: 局部验证决策结果。
 
     返回:
-        每条含 question_id/task_type/prev_correct/curr_correct/category 的 dict 列表。
+        每条含 question_id(=unit_id)/task_type/prev_correct/curr_correct/category 的 dict。
     """
     from app.harness.momentum import (
         IMPROVED,
@@ -1294,7 +1339,7 @@ class Runner:
         self._writeback_val_correctness(eval_r.run_id, pools, state)
 
         # Phase 4: probation 结算
-        self._settle_probations(eval_r.run_id, state)
+        self._settle_probations(eval_r.run_id, pools, state)
 
         # Phase 5: best argmax
         self._maybe_promote_best(
@@ -1382,15 +1427,20 @@ class Runner:
     # 慢更新内部方法
     # -----------------------------------------------------------------------
 
-    def _settle_probations(self, eval_run_id: str, state: _TrainState) -> None:
+    def _settle_probations(self, eval_run_id: str, pools: Pools, state: _TrainState) -> None:
-        """epoch 末试用期一次性结算：全 val 重跑逐题结果与锚快照配对。
+        """epoch 末试用期一次性结算：全 val 重跑结果按 unit 与锚快照配对。
+
+        W/L 按 **unit 口径** 统计（AR pair 双向 AND 折叠，不被 P/Q 单题计分污染，
+        核心算法保真 #5）。锚快照与当前重跑均先经 unit_correctness_view 折叠成单元
+        视图再走 pair_block 计翻转。逐题 predictions 仍逐题落库溯源。
 
         参数:
             eval_run_id: 本 epoch 全 val 重跑（R）的 run_id。
+            pools: 冻结三池（按 task_type 取 val 子集重建单元）。
             state: 训练状态（probations 结算后清空）。
 
         异常:
-            RuntimeError: 重跑缺某快照题的预测行。
+            RuntimeError: 重跑缺某 val 题的预测行。
         """
         if not state.probations:
             return
@@ -1410,20 +1460,12 @@ class Runner:
         )
         for task_type in sorted(state.probations):
             probation = state.probations[task_type]
-            w = l = 0  # noqa: E741
+            val_units = build_units([q for q in pools.validation if q.task_type == task_type])
-            for qid, snap_correct in probation.correctness_snapshot.items():
+            flips = _probation_unit_flips(
-                row = rows.get(qid)
+                val_units, probation.correctness_snapshot, rows, task_type, eval_run_id
-                if row is None:
+            )
-                    raise RuntimeError(
+            verdict = probation_verdict(flips.w, flips.l, params=params)
-                        f"probation 结算缺预测行: {task_type}/{qid}（run={eval_run_id}）"
+            logger.info("probation 结算[{}]: W={} L={} → {}", task_type, flips.w, flips.l, verdict)
-                    )
-                cur = row["_correct"]
-                if not snap_correct and cur:
-                    w += 1
-                elif snap_correct and not cur:
-                    l += 1  # noqa: E741
-            verdict = probation_verdict(w, l, params=params)
-            logger.info("probation 结算[{}]: W={} L={} → {}", task_type, w, l, verdict)
             if verdict == "rollback":
                 self._rollback_probation(probation, state)
         state.probations.clear()

app/harness/validate.py

@@ -25,6 +25,7 @@ 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 (
     GateParams,
     GateVerdict,
@@ -37,7 +38,7 @@ from core.evolution import (
 if TYPE_CHECKING:
     from app.harness.inference import InferenceResult
     from app.harness.log import HarnessLog
-    from core.types import GeneratedQuestion
+    from core.types import GeneratedQuestion, QuestionUnit
 
 
 # gate_decision 的 decision → ValidationOutcome.stop_reason 映射
@@ -95,10 +96,12 @@ class InferenceRunConfig:
 
 @dataclass
 class ValidationOutcome:
-    """CE-Gate 局部验证结果：三态动作 + e-process 证据 + 已观测题逐题对错。
+    """CE-Gate 局部验证结果：三态动作 + e-process 证据（单元口径）+ 逐题溯源对错。
 
-    correctness 二轨语义：candidate_correctness 只含已观测题（早停后是
+    correctness 二轨语义：W/L、准确率、四象限均按 **unit 口径** 统计
-    阶梯前缀子集）；accept 时由 runner 按题粒度增量合并进 state.correctness。
+    （AR pair 双向 AND 折叠为一个单元，不被 P/Q 单题计分污染）；
+    candidate_correctness 独立保留 **逐题** 对错（只含已观测题，早停后是阶梯前缀
+    子集），accept 时由 runner 按 question_id 粒度增量合并进 state.correctness。
     """
 
     action: str  # accept_confirmed | accept_provisional | reject
@@ -110,8 +113,8 @@ class ValidationOutcome:
     n_used: int
     delta_hat: float
     delta_shrunk: float
-    baseline_acc: float  # 已观测题上的基线准确率（观测口径）
+    baseline_acc: float  # 已观测单元上的基线准确率（unit 口径）
-    candidate_acc: float  # 已观测题上的候选准确率（观测口径）
+    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)
@@ -252,7 +255,7 @@ def _candidate_correctness_from_db(
 
 
 async def _resolve_baseline_block(
-    chunk: list[GeneratedQuestion],
+    units: list[QuestionUnit],
     task_type: str,
     s_hash: str,
     prompts_version: str,
@@ -262,102 +265,114 @@ async def _resolve_baseline_block(
     log: HarnessLog,
     run_id: str,
 ) -> tuple[dict[str, bool], int, int]:
-    """基线侧处理一个块：缓存优先，miss 的题新鲜跑基线版本并回写缓存。
+    """基线侧处理一个块：缓存优先（unit 键），miss 的单元新鲜跑基线版本并回写缓存。
+
+    缓存以 unit_id 为键、存单元级对错（AR pair 双向 AND 折叠后一个布尔）。
+    miss 的单元展开为逐题送推理，读回逐题预测后经 unit_correctness_view 折叠成
+    单元级对错再写缓存（核心算法保真 #5）。逐题 predictions 仍逐题落库溯源。
 
     参数:
-        chunk: 当前块的题目列表。
+        units: 当前块的单元列表（single 或 AR pair）。
         task_type: 当前验证题型（缓存键成分）。
         s_hash: 基线侧生效 skill 的内容哈希（缓存键成分）。
         prompts_version: 当前 prompts 版本（缓存键成分）。
-        baseline_cache: 基线侧逐题对错缓存。
+        baseline_cache: 基线侧单元级对错缓存（键含 unit_id）。
         base_skills_dir: 基线 skills 版本目录。
         run_inference: 注入的 async 推理函数。
         log: HarnessLog 共享实例（推理后读预测）。
         run_id: 本块基线 run_id。
 
     返回:
-        (b_map, errors_inc, denom_inc)：块内 question_id -> 基线对错、
+        (b_units, errors_inc, denom_inc)：块内 unit_id -> 基线单元对错、
         本块新增的 INFRA error 计数与推理题次分母增量（全命中时为 0, 0）。
     """
-    misses = [
+    miss_units = [
-        q
+        u
-        for q in chunk
+        for u in units
-        if baseline_cache.get(task_type, s_hash, prompts_version, q.question_id) is None
+        if baseline_cache.get(task_type, s_hash, prompts_version, u.unit_id) is None
     ]
     errors_inc = 0
     denom_inc = 0
-    if misses:
+    if miss_units:
-        r_b = await run_inference(misses, run_id=run_id, skills_dir=base_skills_dir)
+        miss_questions = flatten_units(miss_units)
+        r_b = await run_inference(miss_questions, run_id=run_id, skills_dir=base_skills_dir)
         errors_inc = r_b.stop_reason_counts.get("error", 0)
         denom_inc = r_b.total
-        fresh = _candidate_correctness_from_db(log, r_b.run_id, misses)
+        fresh_per_q = _candidate_correctness_from_db(log, r_b.run_id, miss_questions)
-        for qid, correct in fresh.items():
+        fresh_units = unit_correctness_view(miss_units, fresh_per_q)
-            baseline_cache.put(task_type, s_hash, prompts_version, qid, correct)
+        for uid, correct in fresh_units.items():
+            baseline_cache.put(task_type, s_hash, prompts_version, uid, correct)
 
-    b_map: dict[str, bool] = {}
+    b_units: dict[str, bool] = {}
-    for q in chunk:
+    for u in units:
-        val = baseline_cache.get(task_type, s_hash, prompts_version, q.question_id)
+        val = baseline_cache.get(task_type, s_hash, prompts_version, u.unit_id)
-        assert val is not None, f"基线缓存补齐后仍有 miss: {q.question_id} run_id={run_id}"
+        assert val is not None, f"基线缓存补齐后仍有 miss: unit={u.unit_id} run_id={run_id}"
-        b_map[q.question_id] = val
+        b_units[u.unit_id] = val
-    return b_map, errors_inc, denom_inc
+    return b_units, errors_inc, denom_inc
 
 
 async def _run_candidate_block(
-    chunk: list[GeneratedQuestion],
+    units: list[QuestionUnit],
     cand_dir: Path,
     run_inference: RunInferenceFn,
     log: HarnessLog,
     run_id: str,
 ) -> tuple[dict[str, bool], int, int]:
-    """候选侧处理一个块：全块新鲜跑候选版本并从 db 读逐题对错。
+    """候选侧处理一个块：单元展开为逐题全块新鲜跑候选版本并从 db 读逐题对错。
+
+    返回逐题对错映射（question_id -> bool），折叠为单元视图交由调用方完成，
+    逐题结果同时用于 candidate_correctness 溯源与二轨 correctness 合并。
 
     参数:
-        chunk: 当前块的题目列表。
+        units: 当前块的单元列表。
         cand_dir: 已物化的候选 skills 目录。
         run_inference: 注入的 async 推理函数。
         log: HarnessLog 共享实例（推理后读预测）。
         run_id: 本块候选 run_id。
 
     返回:
-        (c_map, errors_inc, denom_inc)。
+        (c_per_q, errors_inc, denom_inc)：块内 question_id -> 候选对错。
     """
-    r_c = await run_inference(chunk, run_id=run_id, skills_dir=cand_dir)
+    questions = flatten_units(units)
-    c_map = _candidate_correctness_from_db(log, r_c.run_id, chunk)
+    r_c = await run_inference(questions, run_id=run_id, skills_dir=cand_dir)
-    return c_map, r_c.stop_reason_counts.get("error", 0), r_c.total
+    c_per_q = _candidate_correctness_from_db(log, r_c.run_id, questions)
+    return c_per_q, r_c.stop_reason_counts.get("error", 0), r_c.total
 
 
 def _build_evidence_rows(
-    chunk: list[GeneratedQuestion],
+    units: list[QuestionUnit],
-    b_map: dict[str, bool],
+    b_units: dict[str, bool],
-    c_map: dict[str, bool],
+    c_units: dict[str, bool],
     task_type: str,
     block_idx: int,
 ) -> list[dict]:
-    """组装一个块的 gate_evidence 逐题证据行。
+    """组装一个块的 gate_evidence 单元级证据行。
 
+    证据行按 unit 口径（question_id 字段存 unit_id、correct 存单元级对错），
+    与 e-process 判定同粒度；逐题预测明细仍在 predictions 表逐题溯源。
     e_value 留 None 待块判定后回填，stop_reason 留空串待终态回填。
 
     参数:
-        chunk: 当前块的题目列表。
+        units: 当前块的单元列表。
-        b_map: 块内 question_id -> 基线对错。
+        b_units: 块内 unit_id -> 基线单元对错。
-        c_map: 块内 question_id -> 候选对错。
+        c_units: 块内 unit_id -> 候选单元对错。
         task_type: 当前验证题型。
         block_idx: 当前块序号。
 
     返回:
-        逐题证据行列表。
+        单元级证据行列表。
     """
     return [
         {
-            "question_id": q.question_id,
+            "question_id": u.unit_id,
             "task_type": task_type,
             "block_idx": block_idx,
-            "baseline_correct": b_map[q.question_id],
+            "baseline_correct": b_units[u.unit_id],
-            "candidate_correct": c_map[q.question_id],
+            "candidate_correct": c_units[u.unit_id],
             "e_value": None,
             "stop_reason": "",
         }
-        for q in chunk
+        for u in units
     ]
 
 
@@ -394,20 +409,25 @@ def _finalize_outcome(
     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（基线错→候选对翻转）。
+        w: 累计 W（基线错→候选对单元翻转）。
-        l: 累计 L（基线对→候选错翻转）。
+        l: 累计 L（基线对→候选错单元翻转）。
-        n_used: 已消费的阶梯题数。
+        n_used: 已消费的阶梯单元数。
-        n_plan: 阶梯总题数。
+        n_plan: 阶梯总单元数。
-        base_obs: 累计基线已观测对错。
+        base_obs: 累计基线已观测单元对错（unit_id -> bool）。
-        cand_obs: 累计候选已观测对错。
+        cand_obs: 累计候选已观测单元对错（unit_id -> bool）。
-        evidence_rows: 逐题证据行。
+        candidate_per_q: 累计候选逐题对错（question_id -> bool，溯源用）。
+        evidence_rows: 单元级证据行。
         task_type: 验证题型（日志用）。
 
     返回:
@@ -418,16 +438,16 @@ def _finalize_outcome(
         "accept_provisional": "accept_provisional",
     }.get(verdict.decision, "reject")
     stop_reason = _STOP_REASON_BY_DECISION[verdict.decision]
-    # 只有终态题的证据行才携带 stop_reason
+    # 只有终态单元的证据行才携带 stop_reason
     evidence_rows[-1]["stop_reason"] = stop_reason
 
-    quadrants = classify_quadrants({qid: (base_obs[qid], cand_obs[qid]) for qid in base_obs})
+    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={}/{}) {}",
+        "gate 局部验证[{}]: 基线{:.1%} → 候选{:.1%} (W={} L={} E={:.2f} n={}/{} 单元) {}",
         task_type,
         baseline_acc,
         candidate_acc,
@@ -455,7 +475,7 @@ def _finalize_outcome(
         regressions=quadrants.regressions,
         persistent_fails=quadrants.persistent_fails,
         stable_successes=quadrants.stable_successes,
-        candidate_correctness=cand_obs,
+        candidate_correctness=candidate_per_q,
         evidence_rows=evidence_rows,
     )
 
@@ -465,13 +485,33 @@ def _finalize_outcome(
 # ---------------------------------------------------------------------------
 
 
+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,
-    plan: list[GeneratedQuestion],
+    units: list[QuestionUnit],
     gate_params: GateParams,
     gate_block: int,
     gate_guard_err: float,
@@ -481,12 +521,12 @@ async def _run_local_validation(
     log: HarnessLog,
     gate_run_prefix: str,
 ) -> ValidationOutcome:
-    """块序贯循环主体：逐块基线（缓存优先）/候选配对推理，块间 e-process 判定。
+    """块序贯循环主体：逐块基线（缓存优先）/候选按单元配对推理，块间 e-process 判定。
 
-    按 gate_block 切阶梯前缀，每块先补齐基线侧缓存 miss（新鲜跑基线版本
+    按 gate_block 切**单元**前缀（AR pair 整锁在同一块，不跨块拆分），每块先补齐
-    并逐题写 BaselineCache），再全块跑候选，配对累计 W/L 后调 gate_decision；
+    基线侧缓存 miss（新鲜跑基线版本并按 unit_id 写 BaselineCache），再全块跑候选，
-    非 continue 即早停。题尽时最后一块的判定即终态（n_remaining=0 走
+    折叠成单元视图后配对累计 W/L 调 gate_decision；非 continue 即早停。单元尽时
-    provisional/inertia 分支），无循环外补判。
+    最后一块的判定即终态（n_remaining=0 走 provisional/inertia 分支），无循环外补判。
 
     参数:
         workspace_dir: Workspace 根目录。
@@ -494,11 +534,11 @@ async def _run_local_validation(
         base_skills_version: 基线 skills 版本名。
         task_type: 当前验证题型。
         base_skill_content: 基线侧生效 skill 全文（skill_hash 作缓存键成分）。
-        plan: 已截断到 gate_n_max 的阶梯出题序。
+        units: 已截断到 gate_n_max 的阶梯单元序（single 或 AR pair）。
         gate_params: e-process 判据阈值组。
-        gate_block: 块大小。
+        gate_block: 块大小（单位为**单元数**）。
         gate_guard_err: 跨块累计 INFRA 错误率护栏（分母 >=10 才触发）。
-        baseline_cache: 基线侧逐题对错缓存。
+        baseline_cache: 基线侧单元级对错缓存（键含 unit_id）。
         prompts_version: 当前 prompts 版本（缓存键成分）。
         run_inference: 注入的 async 推理函数。
         log: HarnessLog 共享实例。
@@ -508,8 +548,8 @@ async def _run_local_validation(
         ValidationOutcome。
 
     关键实现:
-        INFRA 护栏跨块累计基线+候选两侧的 error 计数，分母（总推理题次）>=10
+        INFRA 护栏跨块累计基线+候选两侧的 error 计数，分母（总推理题次，仍逐题计）
-        且错误率超 gate_guard_err 时直接 raise，避免坏批次污染判定。
+        >=10 且错误率超 gate_guard_err 时直接 raise，避免坏批次污染判定。
     """
     w = 0
     l = 0  # noqa: E741
@@ -519,15 +559,17 @@ async def _run_local_validation(
     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
-    chunks = [plan[i : i + gate_block] for i in range(0, len(plan), gate_block)]
+    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, chunk in enumerate(chunks):
+    for block_idx, unit_chunk in enumerate(unit_chunks):
         # Phase 1: 基线侧（缓存优先，miss 新鲜跑）+ 候选侧（全块新鲜跑）
-        b_map, err_b, den_b = await _resolve_baseline_block(
+        b_units, err_b, den_b = await _resolve_baseline_block(
-            chunk=chunk,
+            units=unit_chunk,
             task_type=task_type,
             s_hash=s_hash,
             prompts_version=prompts_version,
@@ -537,8 +579,8 @@ async def _run_local_validation(
             log=log,
             run_id=f"{gate_run_prefix}_b{block_idx}_base",
         )
-        c_map, err_c, den_c = await _run_candidate_block(
+        c_per_q, err_c, den_c = await _run_candidate_block(
-            chunk=chunk,
+            units=unit_chunk,
             cand_dir=cand_dir,
             run_inference=run_inference,
             log=log,
@@ -550,19 +592,21 @@ async def _run_local_validation(
         infra_denom += den_b + den_c
         _check_infra_guard(errors, infra_denom, gate_guard_err)
 
-        # Phase 3: 配对 + 证据行 + 块间判定
+        # Phase 3: 折叠成单元视图 + 配对 + 证据行 + 块间判定
-        qids = [q.question_id for q in chunk]
+        c_units = unit_correctness_view(unit_chunk, c_per_q)
-        pair_result = pair_block(b_map, c_map, qids)
+        candidate_per_q.update(c_per_q)
-        for qid, (b, c) in pair_result.observed.items():
+        unit_ids = [u.unit_id for u in unit_chunk]
-            base_obs[qid] = b
+        pair_result = pair_block(b_units, c_units, unit_ids)
-            cand_obs[qid] = c
+        for uid, (b, c) in pair_result.observed.items():
+            base_obs[uid] = b
+            cand_obs[uid] = c
 
-        block_rows = _build_evidence_rows(chunk, b_map, c_map, task_type, block_idx)
+        block_rows = _build_evidence_rows(unit_chunk, b_units, c_units, task_type, block_idx)
 
         w += pair_result.w
         l += pair_result.l  # noqa: E741
-        n_used += len(chunk)
+        n_used += len(unit_chunk)
-        verdict = gate_decision(w, l, n_used, len(plan) - n_used, params=gate_params)
+        verdict = gate_decision(w, l, n_used, n_plan - n_used, params=gate_params)
 
         for row in block_rows:
             row["e_value"] = verdict.e_value
@@ -578,9 +622,10 @@ async def _run_local_validation(
         w=w,
         l=l,
         n_used=n_used,
-        n_plan=len(plan),
+        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,
     )
@@ -618,10 +663,10 @@ async def validate_skill_local(
             （skill_hash(base_skill_content) 作 BaselineCache 键成分）。
         ladder_items: 阶梯序题目列表（已排除本 step 案例包题）。
         gate_params: e-process 判据阈值组。
-        gate_block: 块大小。
+        gate_block: 块大小（单位为**单元数**，AR pair 整锁不跨块拆）。
-        gate_n_max: 单 gate 题数上限。
+        gate_n_max: 单 gate 单元数上限（阶梯截断到此数量个单元）。
         gate_guard_err: 跨块累计 INFRA 错误率护栏（分母 >=10 才触发）。
-        baseline_cache: 基线侧逐题对错缓存。
+        baseline_cache: 基线侧单元级对错缓存（键含 unit_id）。
         prompts_version: 当前 prompts 版本（缓存键成分）。
         run_inference: 注入的 async 推理函数（RunInferenceFn 协议）。
         log: HarnessLog 共享实例（供 DB 回读逐题对错）。
@@ -629,7 +674,7 @@ async def validate_skill_local(
             （防泄露过滤靠它识别）。块 run_id = f"{prefix}_b{block_idx}_{arm}"。
 
     返回:
-        ValidationOutcome。逐题证据记入 outcome.evidence_rows 随结果返回，
+        ValidationOutcome。单元级证据记入 outcome.evidence_rows 随结果返回，
         gate_evidence 落库由调用方（runner）负责。
     """
     if "_gate_" not in gate_run_prefix:
@@ -637,7 +682,8 @@ async def validate_skill_local(
     if not ladder_items:
         raise ValueError(f"task_type={task_type} 阶梯为空，无法验证")
 
-    plan = ladder_items[:gate_n_max]
+    # 阶梯题序聚合为单元并按信息阶梯序截断到 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
     )
@@ -648,7 +694,7 @@ async def validate_skill_local(
             base_skills_version=base_skills_version,
             task_type=task_type,
             base_skill_content=base_skill_content,
-            plan=plan,
+            units=units,
             gate_params=gate_params,
             gate_block=gate_block,
             gate_guard_err=gate_guard_err,

core/evolution/validate.py

@@ -1,9 +1,12 @@
 """core/evolution/validate.py — 块验证纯决策函数。
 
-算法 #7（块顺序验证）的局部实现：pair_block 逐题比对基线与候选、
+算法 #7（块顺序验证）的局部实现：pair_block 按 unit 比对基线与候选、
 classify_quadrants 四象限分类、compute_accuracy 纯算术准确率。
 
-三个函数均为纯函数，无副作用、无外部依赖。
+三个函数均为纯函数，无副作用、无外部依赖。输入的对错映射均为 **unit 口径**
+（unit_id → 单元级正确性，AR pair 已在上游经 unit_correctness_view 双向 AND
+折叠），保证 e-process W/L 与准确率分母按单元计、不被 P/Q 单题计分污染
+（核心算法保真 #5：信息阶梯口径从 question_id 迁至 unit_id）。
 """
 
 from core.evolution.types import PairResult, QuadrantClassification
@@ -12,24 +15,24 @@ from core.evolution.types import PairResult, QuadrantClassification
 def pair_block(
     baseline: dict[str, bool],
     candidate: dict[str, bool],
-    question_ids: list[str],
+    unit_ids: list[str],
 ) -> PairResult:
-    """逐题比对基线与候选对错，统计翻转。
+    """按单元比对基线与候选对错，统计翻转。
 
     参数:
-        baseline: 基线臂每题正确性映射。
+        baseline: 基线臂单元级正确性映射（unit_id → bool）。
-        candidate: 候选臂每题正确性映射。
+        candidate: 候选臂单元级正确性映射（unit_id → bool）。
-        question_ids: 参与比对的题目 ID 列表。
+        unit_ids: 参与比对的单元 ID 列表（AR pair 折叠后为单一 unit_id）。
 
     返回:
         PairResult，包含 w（基线错→候选对翻转数）、l（基线对→候选错翻转数）
-        和 observed（每题的 (基线, 候选) 对错记录）。
+        和 observed（每单元的 (基线, 候选) 对错记录）。
     """
     w = l = 0  # noqa: E741 — 数学记号 W/L（win/loss），与 gate.py 一致
     observed: dict[str, tuple[bool, bool]] = {}
-    for qid in question_ids:
+    for uid in unit_ids:
-        b, c = baseline[qid], candidate[qid]
+        b, c = baseline[uid], candidate[uid]
-        observed[qid] = (b, c)
+        observed[uid] = (b, c)
         if not b and c:
             w += 1
         elif b and not c:
@@ -71,15 +74,15 @@ def classify_quadrants(
 
 def compute_accuracy(
     correctness: dict[str, bool],
-    question_ids: list[str],
+    unit_ids: list[str],
 ) -> float:
-    """纯算术：sum(correct) / len(ids)。
+    """纯算术：sum(correct) / len(units)，分母按单元数（非逐题）。
 
     参数:
-        correctness: 每题正确性映射。
+        correctness: 单元级正确性映射（unit_id → bool）。
-        question_ids: 参与计算的题目 ID 列表。
+        unit_ids: 参与计算的单元 ID 列表。
 
     返回:
-        准确率浮点数。question_ids 为空时抛出 ZeroDivisionError。
+        准确率浮点数。unit_ids 为空时抛出 ZeroDivisionError。
     """
-    return sum(correctness[qid] for qid in question_ids) / len(question_ids)
+    return sum(correctness[uid] for uid in unit_ids) / len(unit_ids)

tests/unit/test_correctness_unit_view.py

@@ -0,0 +1,131 @@
+"""tests/unit/test_correctness_unit_view.py — correctness 三对象口径单元测试。
+
+验证 unit_correctness_view（逐题 per_q → unit 折叠：AR pair 双向 AND、非 AR single）
+及 core.evolution 的 pair_block / compute_accuracy 消费 unit 口径时，混格池中
+AR pair 折叠为单元、W/L 与准确率分母不被 P/Q 单题计分污染（核心算法保真 #5）。
+"""
+
+from __future__ import annotations
+
+from app.harness.question_units import build_units, unit_correctness_view
+from core.evolution.validate import classify_quadrants, compute_accuracy, pair_block
+from core.types import GeneratedQuestion
+
+
+def _single(qid: str, task_type: str = "temporal") -> GeneratedQuestion:
+    """构造一条非 AR single 题（unit_id 回填为 question_id）。"""
+    return GeneratedQuestion(
+        question_id=qid,
+        video_id=f"v_{qid}",
+        task_type=task_type,
+        question="Q?",
+        options=("A", "B", "C", "D"),
+        answer="A",
+        source_nodes=(),
+        difficulty="easy",
+    )
+
+
+def _pair(pair_id: str, task_type: str = "temporal") -> tuple[GeneratedQuestion, GeneratedQuestion]:
+    """构造一个 AR 孪生对（original + mirror，共享 pair_id → unit_id=pair_id）。"""
+    common = {
+        "video_id": f"v_{pair_id}",
+        "task_type": task_type,
+        "question": "Q?",
+        "options": ("A", "B", "C", "D"),
+        "answer": "A",
+        "source_nodes": (),
+        "difficulty": "easy",
+        "pair_id": pair_id,
+        "flip_axis": "before_after",
+    }
+    orig = GeneratedQuestion(question_id=f"{pair_id}_o", question_role="pair_original", **common)
+    mirror = GeneratedQuestion(question_id=f"{pair_id}_m", question_role="pair_mirror", **common)
+    return orig, mirror
+
+
+class TestUnitCorrectnessView:
+    """unit_correctness_view：逐题对错折叠成 unit_id → bool。"""
+
+    def test_mixed_pool_and_semantics(self) -> None:
+        """混格：single 直取、AR pair 双向 AND。"""
+        s0 = _single("s0")
+        s1 = _single("s1")
+        p1o, p1m = _pair("p1")
+        p2o, p2m = _pair("p2")
+        units = build_units([s0, s1, p1o, p1m, p2o, p2m])
+        per_q = {
+            "s0": True,
+            "s1": False,
+            "p1_o": True,
+            "p1_m": True,
+            "p2_o": True,
+            "p2_m": False,
+        }
+        view = unit_correctness_view(units, per_q)
+        # single 的 unit_id 等于 question_id；pair 的 unit_id 等于 pair_id
+        assert view == {"s0": True, "s1": False, "p1": True, "p2": False}
+
+    def test_missing_per_q_raises(self) -> None:
+        """任一题缺 per_q → KeyError（禁静默兜底）。"""
+        p1o, p1m = _pair("p1")
+        units = build_units([p1o, p1m])
+        import pytest
+
+        with pytest.raises(KeyError):
+            unit_correctness_view(units, {"p1_o": True})
+
+
+class TestPairBlockUnitFold:
+    """pair_block 消费 unit 口径：混格 W/L 不被 P/Q 单题计分污染。"""
+
+    def test_pair_partial_improvement_not_counted(self) -> None:
+        """AR pair 基线(F,F)→候选(T,F)：单元仍错，不计 W（保真 #5）。"""
+        p1o, p1m = _pair("p1")
+        s0 = _single("s0")
+        units = build_units([p1o, p1m, s0])
+        unit_ids = [u.unit_id for u in units]
+        baseline_per_q = {"p1_o": False, "p1_m": False, "s0": False}
+        candidate_per_q = {"p1_o": True, "p1_m": False, "s0": True}
+        b_units = unit_correctness_view(units, baseline_per_q)
+        c_units = unit_correctness_view(units, candidate_per_q)
+        result = pair_block(b_units, c_units, unit_ids)
+        # 只有 s0 单元发生 F→T 翻转；pair 单元双向 AND 后仍错，不计 W
+        assert result.w == 1
+        assert result.l == 0
+        assert result.observed["p1"] == (False, False)
+        assert result.observed["s0"] == (False, True)
+
+    def test_pair_full_flip_counts_once(self) -> None:
+        """AR pair 两题齐翻(F,F)→(T,T)：单元计 1 次 W（非 2）。"""
+        p1o, p1m = _pair("p1")
+        units = build_units([p1o, p1m])
+        b_units = unit_correctness_view(units, {"p1_o": False, "p1_m": False})
+        c_units = unit_correctness_view(units, {"p1_o": True, "p1_m": True})
+        result = pair_block(b_units, c_units, [u.unit_id for u in units])
+        assert result.w == 1
+        assert result.l == 0
+
+
+class TestComputeAccuracyUnitDenominator:
+    """compute_accuracy 分母按 unit 数（非逐题）。"""
+
+    def test_denominator_is_unit_count(self) -> None:
+        """1 pair(错) + 1 single(对) → 1/2；逐题会误算 2/3。"""
+        p1o, p1m = _pair("p1")
+        s0 = _single("s0")
+        units = build_units([p1o, p1m, s0])
+        unit_ids = [u.unit_id for u in units]
+        view = unit_correctness_view(units, {"p1_o": True, "p1_m": False, "s0": True})
+        assert compute_accuracy(view, unit_ids) == 0.5
+
+
+class TestClassifyQuadrantsUnitKeys:
+    """classify_quadrants 按 unit_id 分桶（pair 单元只出现一次）。"""
+
+    def test_pair_unit_single_bucket(self) -> None:
+        """pair 单元 F→F 落 persistent_fails，只记 unit_id 一次。"""
+        observed = {"p1": (False, False), "s0": (False, True)}
+        qc = classify_quadrants(observed)
+        assert qc.improvements == ["s0"]
+        assert qc.persistent_fails == ["p1"]

tests/unit/test_gate_block_unit.py

@@ -0,0 +1,279 @@
+"""tests/unit/test_gate_block_unit.py — gate 块实际执行路径按 unit 跑。
+
+针对 app/harness/validate.py::validate_skill_local（真实 gate 执行路径），
+断言混格阶梯下 gate 块按 unit 口径运行：baseline_cache 键含 unit_id、
+n_used 按 unit 累加、pair_block 折叠 AR pair、逐题 predictions 仍溯源。
+核心算法保真 #5（信息阶梯 e-process 口径从 question_id 迁至 unit_id）。
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+import pytest
+
+from app.harness.gate_ladder import BaselineCache, skill_hash
+from app.harness.inference import PREDICTIONS_SCHEMA, InferenceResult
+from app.harness.log import HarnessLog
+from app.harness.validate import validate_skill_local
+from core.evolution import GateParams
+from core.types import GeneratedQuestion
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+_DEFAULT_GATE_PARAMS = GateParams(
+    e_confirm=20.0,
+    e_provisional=3.0,
+    w_net_min=2,
+    delta_min=0.05,
+    lambda_dir=-2.0,
+    e_rollback=10.0,
+)
+
+
+def _single(qid: str) -> GeneratedQuestion:
+    """构造非 AR single 题。"""
+    return GeneratedQuestion(
+        question_id=qid,
+        video_id=f"v_{qid}",
+        task_type="temporal",
+        question="Q?",
+        options=("A", "B", "C", "D"),
+        answer="A",
+        source_nodes=(),
+        difficulty="easy",
+    )
+
+
+def _pair(pair_id: str) -> list[GeneratedQuestion]:
+    """构造 AR 孪生对（original + mirror，共享 pair_id）。"""
+    common = {
+        "video_id": f"v_{pair_id}",
+        "task_type": "temporal",
+        "question": "Q?",
+        "options": ("A", "B", "C", "D"),
+        "answer": "A",
+        "source_nodes": (),
+        "difficulty": "easy",
+        "pair_id": pair_id,
+        "flip_axis": "before_after",
+    }
+    return [
+        GeneratedQuestion(question_id=f"{pair_id}_o", question_role="pair_original", **common),
+        GeneratedQuestion(question_id=f"{pair_id}_m", question_role="pair_mirror", **common),
+    ]
+
+
+def _setup_workspace(tmp_path: Path) -> Path:
+    """构建最小 workspace（skills/v1/temporal.md）。"""
+    skills_dir = tmp_path / "skills" / "v1"
+    skills_dir.mkdir(parents=True)
+    (skills_dir / "temporal.md").write_text("baseline skill content", encoding="utf-8")
+    return tmp_path
+
+
+def _make_log(workspace: Path, run_id: str = "test_master") -> HarnessLog:
+    """创建 HarnessLog 并建 predictions 表。"""
+    log = HarnessLog(str(workspace / "harness.db"), run_id)
+    log.create_table("predictions", PREDICTIONS_SCHEMA)
+    return log
+
+
+def _insert_predictions(log: HarnessLog, run_id: str, per_q: dict[str, bool]) -> None:
+    """逐题写 predictions（溯源仍逐题）。"""
+    for qid, correct in per_q.items():
+        log.insert(
+            "predictions",
+            {
+                "run_id": run_id,
+                "video_id": "v0",
+                "question_id": qid,
+                "task_type": "temporal",
+                "prediction": "A" if correct else "Z",
+                "answer": "A",
+                "evidence": "",
+                "reasoning": "",
+                "steps_used": 1,
+                "prompt_tokens": 10,
+                "completion_tokens": 10,
+                "stop_reason": "completed",
+                "steps_json": "[]",
+            },
+        )
+
+
+def _make_mock_run_inference(
+    log: HarnessLog,
+    baseline_correct: dict[str, bool],
+    candidate_correct: dict[str, bool],
+):
+    """构建 mock RunInferenceFn，按 run_id 的 arm 后缀选基线/候选逐题对错。"""
+    call_log: list[dict[str, Any]] = []
+
+    async def mock_fn(
+        questions: list[GeneratedQuestion],
+        *,
+        run_id: str,
+        skills_dir: Path,
+    ) -> InferenceResult:
+        src = baseline_correct if run_id.endswith("_base") else candidate_correct
+        per_q = {q.question_id: src.get(q.question_id, False) for q in questions}
+        _insert_predictions(log, run_id, per_q)
+        call_log.append({"run_id": run_id, "qids": [q.question_id for q in questions]})
+        total = len(questions)
+        return InferenceResult(
+            run_id=run_id,
+            accuracy=sum(per_q.values()) / total if total else 0.0,
+            total=total,
+            correct=sum(per_q.values()),
+            per_task_type={},
+            steps_mean=1.0,
+            token_usage={"prompt_tokens": 10, "completion_tokens": 10},
+            stop_reason_counts={"completed": total},
+        )
+
+    return mock_fn, call_log
+
+
+@pytest.mark.asyncio
+async def test_gate_n_used_counts_units_not_questions(tmp_path: Path) -> None:
+    """混格阶梯（1 pair + 2 single）→ n_used=3 单元，非 4 题。"""
+    workspace = _setup_workspace(tmp_path)
+    log = _make_log(workspace)
+    cache = BaselineCache(workspace / "baseline_cache.json")
+    ladder = [*_pair("p1"), _single("s0"), _single("s1")]
+
+    # 基线全错、候选全对 → 3 单元齐翻 W=3
+    baseline = {"p1_o": False, "p1_m": False, "s0": False, "s1": False}
+    candidate = {"p1_o": True, "p1_m": True, "s0": True, "s1": True}
+    mock_fn, call_log = _make_mock_run_inference(log, baseline, candidate)
+
+    accept_params = GateParams(
+        e_confirm=15.0,
+        e_provisional=3.0,
+        w_net_min=2,
+        delta_min=0.05,
+        lambda_dir=-2.0,
+        e_rollback=10.0,
+    )
+    try:
+        outcome = await validate_skill_local(
+            workspace_dir=workspace,
+            base_skills_version="v1",
+            task_type="temporal",
+            target_file="temporal.md",
+            candidate_content="improved skill",
+            base_skill_content="baseline skill content",
+            ladder_items=ladder,
+            gate_params=accept_params,
+            gate_block=10,
+            gate_n_max=20,
+            gate_guard_err=0.5,
+            baseline_cache=cache,
+            prompts_version="p1",
+            run_inference=mock_fn,
+            log=log,
+            gate_run_prefix="step1_gate_test",
+        )
+        # n_used 按 unit 计（3），W 按 unit 计（3）
+        assert outcome.n_used == 3
+        assert outcome.w == 3
+        assert outcome.l == 0
+        # 证据行按 unit 口径（3 行）
+        assert len(outcome.evidence_rows) == 3
+        # baseline_cache 键含 unit_id：pair 用 pair_id、single 用 question_id
+        s_hash = skill_hash("baseline skill content")
+        assert cache.get("temporal", s_hash, "p1", "p1") is False
+        assert cache.get("temporal", s_hash, "p1", "s0") is False
+        # 逐题 question_id 不作为 baseline_cache 键（pair 成员未单独缓存）
+        assert cache.get("temporal", s_hash, "p1", "p1_o") is None
+        # 逐题 predictions 仍溯源：候选 per-q 含两 pair 成员
+        assert set(outcome.candidate_correctness) >= {"p1_o", "p1_m", "s0", "s1"}
+    finally:
+        log.close()
+
+
+@pytest.mark.asyncio
+async def test_gate_pair_partial_flip_not_counted(tmp_path: Path) -> None:
+    """AR pair 候选仅单向翻(T,F)→单元仍错，W 不被单题污染。"""
+    workspace = _setup_workspace(tmp_path)
+    log = _make_log(workspace)
+    cache = BaselineCache(workspace / "baseline_cache.json")
+    ladder = [*_pair("p1"), _single("s0")]
+
+    baseline = {"p1_o": False, "p1_m": False, "s0": False}
+    # pair 只翻一半（p1_o 对、p1_m 错）→ 单元 AND 仍错；s0 翻对
+    candidate = {"p1_o": True, "p1_m": False, "s0": True}
+    mock_fn, _ = _make_mock_run_inference(log, baseline, candidate)
+
+    try:
+        outcome = await validate_skill_local(
+            workspace_dir=workspace,
+            base_skills_version="v1",
+            task_type="temporal",
+            target_file="temporal.md",
+            candidate_content="improved skill",
+            base_skill_content="baseline skill content",
+            ladder_items=ladder,
+            gate_params=_DEFAULT_GATE_PARAMS,
+            gate_block=10,
+            gate_n_max=20,
+            gate_guard_err=0.5,
+            baseline_cache=cache,
+            prompts_version="p1",
+            run_inference=mock_fn,
+            log=log,
+            gate_run_prefix="step1_gate_test",
+        )
+        # 只有 s0 单元翻转，pair 单元不计 W（保真 #5：不被 P/Q 单题污染）
+        assert outcome.w == 1
+        assert outcome.l == 0
+        assert outcome.n_used == 2
+        # candidate_acc 分母按 unit（2 单元，1 对）→ 0.5
+        assert outcome.candidate_acc == 0.5
+    finally:
+        log.close()
+
+
+@pytest.mark.asyncio
+async def test_gate_baseline_cache_hit_by_unit(tmp_path: Path) -> None:
+    """基线缓存按 unit_id 预填充 → 基线侧全命中不发起推理。"""
+    workspace = _setup_workspace(tmp_path)
+    log = _make_log(workspace)
+    cache = BaselineCache(workspace / "baseline_cache.json")
+    ladder = [*_pair("p1"), _single("s0")]
+
+    s_hash = skill_hash("baseline skill content")
+    # 按 unit_id 预填充（pair→pair_id，single→question_id），全错
+    cache.put("temporal", s_hash, "p1", "p1", False)
+    cache.put("temporal", s_hash, "p1", "s0", False)
+
+    baseline = {"p1_o": False, "p1_m": False, "s0": False}
+    candidate = {"p1_o": True, "p1_m": True, "s0": True}
+    mock_fn, call_log = _make_mock_run_inference(log, baseline, candidate)
+
+    try:
+        outcome = await validate_skill_local(
+            workspace_dir=workspace,
+            base_skills_version="v1",
+            task_type="temporal",
+            target_file="temporal.md",
+            candidate_content="improved skill",
+            base_skill_content="baseline skill content",
+            ladder_items=ladder,
+            gate_params=_DEFAULT_GATE_PARAMS,
+            gate_block=10,
+            gate_n_max=20,
+            gate_guard_err=0.5,
+            baseline_cache=cache,
+            prompts_version="p1",
+            run_inference=mock_fn,
+            log=log,
+            gate_run_prefix="step1_gate_test",
+        )
+        base_calls = [c for c in call_log if c["run_id"].endswith("_base")]
+        assert base_calls == [], "unit 键全命中不应发起基线推理"
+        assert outcome.n_used == 2
+    finally:
+        log.close()