feat(harness): 迁移 gate 信息量阶梯到 unit 粒度 + schema_version

核心算法保真#5（信息阶梯）：gate_ladder.py 从逐题迁移到 unit 粒度，
只换键 question_id→unit_id，冷启动 2:1 错优先交错、gamma-EMA 公式、
Beta(1,1) 先验、反泄漏 _gate_ 过滤的公式/比例/顺序语义一字不改。

- LadderEntry 按 unit_id 键；AR pair 折叠为一个阶梯单元
- build_cold_entries 收单元列表，unit 错 = 任一成员错（双向 AND）折叠，
  2:1 交错 + probe 探针按 unit 抽，Beta 先验 p0 不变
- ladder_for 返回 unit_id 序、exclude 迁到 unit 口径（防半 pair 灌入
  触发下游 _ladder_units fail-fast）
- update_probs 先把逐题观测折叠成单元观测再按 unit_id 匹配更新，
  半观测单元跳过（防按 qid 匹配 pair 失效致 gamma-EMA 停摆）
- GatePools.save/load 加 schema_version=2；存量无版本/旧版本 json
  加载直接报错，拒绝静默混用 qid/unit 键
- BaselineCache 第四维键改名 unit_id（与 T7 validate 路径对齐）
- build_or_load_gate_pools 先折叠单元再排除 test（抽 helper 控复杂度 B）
- runner：_init_gate_pools 建 unit 索引；gate 验证 exclude/展开、
  _refresh_gate_ladder 折叠观测走 units_by_id

反泄漏 run_id 含 _gate_ 过滤不受影响（未改）。
测试：新增 test_gate_ladder_unit_migration.py（15 例覆盖 a-e），
既有 test_harness_gate_ladder.py 迁移到 unit API。全量 1363 passed。

app/harness/gate_ladder.py

@@ -1,13 +1,18 @@
-"""CE-Gate 信息量阶梯与基线缓存。
+"""CE-Gate 信息量阶梯与基线缓存（unit 粒度，核心算法保真 #5）。
 
-阶梯（每题型一条）：gate 的出题顺序表。冷启动（FRESH）用种子基线对错
+阶梯（每题型一条）：gate 的出题顺序表，键为 **unit_id**（single 题 unit_id
-两档粗排（错题高优先 2:1 交错 + 全错题 probe_quota 探针插尾）；
+等于 question_id，AR pair 折叠为一个单元、unit_id 等于共享 pair_id）。冷启动
-epoch >=1 用非 gate run 观测做 gamma-EMA 更新 p_hat，按信息量 p_hat(1-p_hat) 降序、
+（FRESH）用种子基线的**单元级**对错两档粗排（错 unit 高优先 2:1 交错 + 全错
-剔 p_hat 不在 [p_low, p_high]。防泄露铁律：gate 内 rollout 永不回流 p_hat
+unit 的 probe_quota 探针插尾）；epoch >=1 用非 gate run 观测**折叠成单元观测**后做
-（调用方以 run_id 含 "_gate_" 过滤观测源）。
+gamma-EMA 更新 p_hat，按信息量 p_hat(1-p_hat) 降序、剔 p_hat 不在 [p_low, p_high]。
+单元错 = 该单元任一成员错（AR pair 双向 AND）。防泄露铁律：gate 内 rollout 永不
+回流 p_hat（调用方以 run_id 含 "_gate_" 过滤观测源），本迁移不改此过滤。
 
-BaselineCache：基线侧逐题对错缓存，键 = (task_type, skill_hash,
+持久化门控：gate_pools.json 带 schema_version（当前 = 2，unit 键）。旧版无
-prompts_version, qid) 内容寻址、无显式失效。JSON 持久化到 workspace，
+schema_version（v1、qid 键）加载时**直接报错**，拒绝静默混用 qid/unit 键。
+
+BaselineCache：基线侧单元级对错缓存，键 = (task_type, skill_hash,
+prompts_version, unit_id) 内容寻址、无显式失效。JSON 持久化到 workspace，
 供 resume 后合法复用已冻结阶梯上的新鲜 draw。
 """
 
@@ -22,10 +27,16 @@ from typing import TYPE_CHECKING
 
 from loguru import logger
 
+from app.harness.question_units import build_units
+
 if TYPE_CHECKING:
     from pathlib import Path
 
-    from core.types import GeneratedQuestion
+    from core.types import GeneratedQuestion, QuestionUnit
+
+# gate_pools.json 结构版本。v1（隐式、无此字段）为逐题 qid 键的存量格式；
+# v2 起改为 unit_id 键。load 时严格校验，不匹配即报错（不静默迁移/混用）。
+SCHEMA_VERSION = 2
 
 
 def skill_hash(content: str) -> str:
@@ -42,50 +53,66 @@ def skill_hash(content: str) -> str:
 
 @dataclass
 class LadderEntry:
-    """阶梯单元：题目与其估计答对率。
+    """阶梯单元：题目单元与其估计答对率。
 
     字段:
-        question_id: 题目唯一标识。
+        unit_id: 单元唯一标识（single 等于 question_id，AR pair 等于共享 pair_id）。
         p_hat: 估计答对率。冷启动为 Beta(1,1) 平滑的单次观测后验均值
             （错=1/3、对=2/3），此后经 gamma-EMA 更新。
     """
 
-    question_id: str
+    unit_id: str
     p_hat: float
 
 
+def _unit_correct(unit: QuestionUnit, correctness: dict[str, bool]) -> bool:
+    """单元级正确性：AR pair 双向 AND，single 即单题；单元错 = 任一成员错。
+
+    参数:
+        unit: 目标单元。
+        correctness: question_id -> 是否答对（缺项按未答对处理，与迁移前
+            correctness.get(qid, False) 的默认语义一致，不改判定）。
+
+    返回:
+        单元内所有成员均答对时 True，否则 False。
+    """
+    return all(correctness.get(q.question_id, False) for q in unit.questions)
+
+
 def build_cold_entries(
-    questions: list[GeneratedQuestion],
+    units: list[QuestionUnit],
     correctness: dict[str, bool],
     probe_quota: float,
     seed: int,
 ) -> list[LadderEntry]:
-    """冷启动排序：错题高优先 2:1 交错 + 全错题 probe_quota 探针插尾。
+    """冷启动排序（unit 粒度）：错 unit 高优先 2:1 交错 + 全错 unit 探针插尾。
 
     参数:
-        questions: 该题型的全部候选题（已排除 test 池）。
+        units: 该题型的全部候选单元（已排除 test 池；AR pair 已折叠成单元）。
-        correctness: question_id -> 种子基线是否答对（900 题全量对错）。
+        correctness: question_id -> 种子基线是否答对（900 题全量逐题对错）。
-        probe_quota: 从错题中随机抽出插到梯尾的探针比例（防"解锁新能力"盲区）。
+            单元级对错由 _unit_correct 折叠（任一成员错 → 单元错）。
+        probe_quota: 从错 unit 中随机抽出插到梯尾的探针比例（防"解锁新能力"盲区）。
         seed: 洗牌种子，保证确定性重建。
 
     返回:
-        排序后的 LadderEntry 列表（p_hat 用 Beta(1,1) 平滑：错=1/3、对=2/3，
+        排序后的 LadderEntry 列表（键=unit_id；p_hat 用 Beta(1,1) 平滑：错=1/3、
-        与 warm 阶段 gamma-EMA / 信息量排序自然衔接）。
+        对=2/3，与 warm 阶段 gamma-EMA / 信息量排序自然衔接）。
 
     关键实现细节:
-        错题、对题各自固定种子洗牌 -> 抽探针 -> 剩余按 错错对 2:1 交错
+        与逐题版**同公式、同比例、同顺序**，仅把调度粒度从题换成单元：错 unit、
-        （一方耗尽后顺排另一方）-> 探针追加尾部。
+        对 unit 各自固定种子洗牌 -> 按 probe_quota 从错 unit 抽探针 -> 剩余按
+        错错对 2:1 交错（一方耗尽后顺排另一方）-> 探针追加尾部。
     """
     rng = random.Random(seed)
-    wrong = [q for q in questions if not correctness.get(q.question_id, False)]
+    wrong = [u for u in units if not _unit_correct(u, correctness)]
-    right = [q for q in questions if correctness.get(q.question_id, False)]
+    right = [u for u in units if _unit_correct(u, correctness)]
     rng.shuffle(wrong)
     rng.shuffle(right)
 
     n_probe = int(len(wrong) * probe_quota)
     probes, wrong_main = wrong[:n_probe], wrong[n_probe:]
 
-    interleaved: list[GeneratedQuestion] = []
+    interleaved: list[QuestionUnit] = []
     wi, ri = 0, 0
     while wi < len(wrong_main) or ri < len(right):
         for _ in range(2):
@@ -97,10 +124,10 @@ def build_cold_entries(
             ri += 1
     interleaved.extend(probes)
 
-    def _p0(q: GeneratedQuestion) -> float:
+    def _p0(u: QuestionUnit) -> float:
-        return 2 / 3 if correctness.get(q.question_id, False) else 1 / 3
+        return 2 / 3 if _unit_correct(u, correctness) else 1 / 3
 
-    return [LadderEntry(q.question_id, _p0(q)) for q in interleaved]
+    return [LadderEntry(u.unit_id, _p0(u)) for u in interleaved]
 
 
 def order_ladder(entries: list[LadderEntry], p_low: float, p_high: float) -> list[LadderEntry]:
@@ -135,23 +162,24 @@ class GatePools:
     def ladder_for(
         self,
         task_type: str,
-        exclude_qids: set[str],
+        exclude_units: set[str],
         p_low: float,
         p_high: float,
         cold: bool,
     ) -> list[str]:
-        """取该题型的 gate 出题序（qid 列表），排除本 step 进化案例包题。
+        """取该题型的 gate 出题序（unit_id 列表），排除本 step 进化案例包所在单元。
 
         参数:
             task_type: 目标题型。
-            exclude_qids: 本 step 案例包（failure/success cases）的题目 id，
+            exclude_units: 本 step 案例包（failure/success cases）所在单元的
-                防止在"刚学的那道题"上自测。
+                unit_id，防止在"刚学的那道题"上自测。按 **unit** 排除：命中单元
+                整体剔除，避免只排 AR pair 半个成员而向 gate 池灌入半个 pair。
             p_low / p_high: warm 阶段的 p_hat 保留区间。
             cold: True 表示尚无 epoch 级观测（epoch 1），用冷启动存储序；
                 False 走 order_ladder 信息量排序。
 
         返回:
-            排除后的有序 question_id 列表。
+            排除后的有序 unit_id 列表。
 
         异常:
             ValueError: 该题型无阶梯（冷启动构建缺失），或该题型阶梯为空。
@@ -162,33 +190,54 @@ class GatePools:
         if not pool:
             raise ValueError(f"task_type={task_type} 阶梯为空，无可出题目")
         ordered = pool if cold else order_ladder(pool, p_low, p_high)
-        return [e.question_id for e in ordered if e.question_id not in exclude_qids]
+        return [e.unit_id for e in ordered if e.unit_id not in exclude_units]
 
-    def update_probs(self, observations: dict[str, bool], gamma: float) -> None:
+    def update_probs(
-        """gamma-EMA 更新 p_hat：p_hat <- gamma * p_hat + (1-gamma) * obs。只更新有新观测的题。
+        self,
+        per_q_observations: dict[str, bool],
+        units_by_id: dict[str, QuestionUnit],
+        gamma: float,
+    ) -> None:
+        """gamma-EMA 更新 p_hat：先把逐题观测折叠成单元观测，再按 unit_id 匹配更新。
+
+        p_hat <- gamma * p_hat + (1-gamma) * unit_obs。只更新"整个单元都被观测到"
+        的单元；单元观测 = 成员逐题对错的 AND（任一成员错 → 单元错）。折叠是必需的：
+        AR pair 的 unit_id 是 pair_id，若直接按 unit_id 去逐题观测里匹配将永不命中、
+        导致 gamma-EMA 停摆（核心算法保真 #5）。
 
         参数:
-            observations: question_id -> 本 epoch 非 gate run 的最新对错。
+            per_q_observations: question_id -> 本 epoch 非 gate run 的最新逐题对错。
                 调用方必须已按 run_id 过滤掉 gate 内 rollout（防泄露铁律）。
+            units_by_id: unit_id -> QuestionUnit，用于把逐题观测折叠成单元观测。
             gamma: EMA 衰减系数。
+
+        关键实现细节:
+            单元只有在其**全部**成员都出现在 per_q_observations 时才更新；半观测
+            （AR pair 只见一半）跳过，避免用不完整证据污染 p_hat。
         """
         for entries in self.entries.values():
             for e in entries:
-                if e.question_id in observations:
+                unit = units_by_id.get(e.unit_id)
-                    obs = 1.0 if observations[e.question_id] else 0.0
+                if unit is None:
+                    continue
+                if not all(q.question_id in per_q_observations for q in unit.questions):
+                    continue
+                unit_correct = all(per_q_observations[q.question_id] for q in unit.questions)
+                obs = 1.0 if unit_correct else 0.0
                 e.p_hat = gamma * e.p_hat + (1 - gamma) * obs
 
     def save(self, path: Path) -> None:
-        """原子写 gate_pools.json（.tmp 再 replace）。
+        """原子写 gate_pools.json（.tmp 再 replace），落 schema_version + unit_id 键。
 
         参数:
             path: 目标 JSON 路径。
         """
         payload = {
+            "schema_version": SCHEMA_VERSION,
             "seed": self.seed,
             "fingerprint": self.fingerprint,
             "entries": {
-                t: [{"question_id": e.question_id, "p_hat": e.p_hat} for e in es]
+                t: [{"unit_id": e.unit_id, "p_hat": e.p_hat} for e in es]
                 for t, es in self.entries.items()
             },
         }
@@ -198,18 +247,28 @@ class GatePools:
 
     @classmethod
     def load(cls, path: Path) -> GatePools:
-        """从 gate_pools.json 恢复。
+        """从 gate_pools.json 恢复；schema_version 不匹配直接报错（不静默混用）。
 
         参数:
             path: gate_pools.json 路径。
 
         返回:
             复活的 GatePools。
+
+        异常:
+            RuntimeError: 缺 schema_version（存量 v1、qid 键）或版本不等于
+                SCHEMA_VERSION——拒绝把 qid 键当 unit 键静默复用，须 FRESH 重建。
         """
         d = json.loads(path.read_text(encoding="utf-8"))
+        version = d.get("schema_version")
+        if version != SCHEMA_VERSION:
+            raise RuntimeError(
+                f"gate_pools.json schema_version={version!r} 与当前 {SCHEMA_VERSION} 不符"
+                f"（存量 qid 键池不可当 unit 键复用），请删除后 FRESH 重建: {path}"
+            )
         return cls(
             entries={
-                t: [LadderEntry(x["question_id"], x["p_hat"]) for x in es]
+                t: [LadderEntry(x["unit_id"], x["p_hat"]) for x in es]
                 for t, es in d["entries"].items()
             },
             seed=d["seed"],
@@ -264,22 +323,46 @@ def build_or_load_gate_pools(
 
     entries: dict[str, list[LadderEntry]] = {}
     for t in task_types:
-        pool = [q for q in questions if q.task_type == t and q.question_id not in test_qids]
+        units = _task_units_excluding_test(questions, t, test_qids)
-        if not pool:
+        if not units:
-            raise ValueError(f"task_type={t} 无非 test 题，无法建阶梯")
+            raise ValueError(f"task_type={t} 无非 test 单元，无法建阶梯")
-        entries[t] = build_cold_entries(pool, baseline_correctness, probe_quota, seed)
+        entries[t] = build_cold_entries(units, baseline_correctness, probe_quota, seed)
-        logger.info("gate 阶梯[{}]: {} 题（冷启动）", t, len(entries[t]))
+        logger.info("gate 阶梯[{}]: {} 单元（冷启动）", t, len(entries[t]))
     pools = GatePools(entries=entries, seed=seed, fingerprint=fingerprint)
     pools.save(path)
     return pools
 
 
-class BaselineCache:
+def _task_units_excluding_test(
-    """基线侧逐题对错缓存（内容寻址，JSON 持久化）。
+    questions: list[GeneratedQuestion], task_type: str, test_qids: set[str]
+) -> list[QuestionUnit]:
+    """取某题型的非 test 候选单元：先按 unit 折叠，再整体排除含 test 成员的单元。
 
-    键 = (task_type, skill_hash, prompts_version, qid)：任何影响该题型
+    先折叠后排除保证 AR pair 不被拆半（否则半个 pair 交给下游会触发 build_units 的
+    孤儿 fail-fast）；single 单元等价于逐题排除（核心算法保真 #5）。
+
+    参数:
+        questions: benchmark 全量题。
+        task_type: 目标题型。
+        test_qids: held-out test 池题目 id。
+
+    返回:
+        该题型下不含任何 test 成员的候选单元列表。
+    """
+    pool = [q for q in questions if q.task_type == task_type]
+    return [
+        u for u in build_units(pool) if all(q.question_id not in test_qids for q in u.questions)
+    ]
+
+
+class BaselineCache:
+    """基线侧单元级对错缓存（内容寻址，JSON 持久化）。
+
+    键 = (task_type, skill_hash, prompts_version, unit_id)：任何影响该题型
     有效 skill 的变化（含共享 default-strategy.md 被他类 accept 改写）
-    都使 skill_hash 变化、缓存自然 miss；prompts 版本变化同理。
+    都使 skill_hash 变化、缓存自然 miss；prompts 版本变化同理。unit_id 维度
+    使 single 题以自身 question_id、AR pair 以共享 pair_id 寻址，缓存单元级
+    对错（pair 双向 AND 折叠后一个布尔）。
     """
 
     def __init__(self, path: Path) -> None:
@@ -294,32 +377,32 @@ class BaselineCache:
             self._store = json.loads(path.read_text(encoding="utf-8"))
 
     @staticmethod
-    def _key(task_type: str, s_hash: str, prompts_version: str, qid: str) -> str:
+    def _key(task_type: str, s_hash: str, prompts_version: str, unit_id: str) -> str:
-        """拼缓存键（四维内容寻址）。"""
+        """拼缓存键（四维内容寻址，第四维为 unit_id）。"""
-        return f"{task_type}|{s_hash}|{prompts_version}|{qid}"
+        return f"{task_type}|{s_hash}|{prompts_version}|{unit_id}"
 
-    def get(self, task_type: str, s_hash: str, prompts_version: str, qid: str) -> bool | None:
+    def get(self, task_type: str, s_hash: str, prompts_version: str, unit_id: str) -> bool | None:
         """读缓存；未命中返回 None。
 
         参数:
             task_type: 题型。
             s_hash: 基线侧生效 skill 文件的内容哈希。
             prompts_version: 当前 prompts 版本。
-            qid: 题目 id。
+            unit_id: 单元 id（single=question_id，AR pair=pair_id）。
 
         返回:
-            缓存的对错；未命中 None。
+            缓存的单元级对错；未命中 None。
         """
-        return self._store.get(self._key(task_type, s_hash, prompts_version, qid))
+        return self._store.get(self._key(task_type, s_hash, prompts_version, unit_id))
 
     def put(
-        self, task_type: str, s_hash: str, prompts_version: str, qid: str, correct: bool
+        self, task_type: str, s_hash: str, prompts_version: str, unit_id: str, correct: bool
     ) -> None:
         """写缓存并落盘（原子写，gate 频度低、全量重写成本可忽略）。
 
         参数:
-            task_type / s_hash / prompts_version / qid: 缓存键四维。
+            task_type / s_hash / prompts_version / unit_id: 缓存键四维。
-            correct: 基线侧该题对错。
+            correct: 基线侧该单元对错（AR pair 双向 AND 折叠后一个布尔）。
 
         关键实现细节:
             先盘后存：新条目先原子落盘（tmp 写 + os.replace）成功后才更新
@@ -327,7 +410,7 @@ class BaselineCache:
         """
         updated = {
             **self._store,
-            self._key(task_type, s_hash, prompts_version, qid): correct,
+            self._key(task_type, s_hash, prompts_version, unit_id): correct,
         }
         tmp = self._path.with_suffix(".json.tmp")
         tmp.write_text(json.dumps(updated, ensure_ascii=False), encoding="utf-8")

app/harness/runner.py

@@ -854,6 +854,11 @@ class Runner:
         self._gate_questions_by_id: dict[str, GeneratedQuestion] = {
             q.question_id: q for q in questions
         }
+        # unit 索引：ladder_for 返回 unit_id、update_probs 折叠逐题观测均需按 unit_id
+        # 反查成员题（核心算法保真 #5：gate 阶梯 unit 化）。不进 checkpoint、每次启动重建。
+        self._gate_units_by_id: dict[str, QuestionUnit] = {
+            u.unit_id: u for u in build_units(questions)
+        }
         with HarnessLog(str(self._paths.db_path), pools.baseline_run_id) as log:
             rows = log.query(
                 "SELECT question_id, prediction, answer FROM predictions WHERE run_id=?",
@@ -1108,21 +1113,29 @@ class Runner:
         from app.harness.log import HarnessLog
         from app.harness.validate import validate_skill_local
 
-        exclude_qids = {c.question_id for c in pack.failure_cases + pack.success_cases}
+        # 案例包按 unit 排除：把每个 case 的 question_id 映射到其所属 unit_id，
-        ladder_qids = state.gate_pools.ladder_for(
+        # 命中单元整体排除，防止只排 AR pair 半个成员而给 gate 池灌半个 pair
+        # （下游 _ladder_units 会 fail-fast）。核心算法保真 #5。
+        exclude_units = {
+            self._gate_questions_by_id[c.question_id].unit_id
+            for c in pack.failure_cases + pack.success_cases
+            if c.question_id in self._gate_questions_by_id
+        }
+        ladder_unit_ids = state.gate_pools.ladder_for(
             task_type,
-            exclude_qids,
+            exclude_units,
             p_low=self._config.gate_p_low,
             p_high=self._config.gate_p_high,
             cold=not state.gate_epoch_observed,
         )
-        missing = [qid for qid in ladder_qids if qid not in self._gate_questions_by_id]
+        missing = [uid for uid in ladder_unit_ids if uid not in self._gate_units_by_id]
         if missing:
             raise ValueError(
-                f"gate 阶梯[{task_type}] 含 benchmark 中不存在的题: "
+                f"gate 阶梯[{task_type}] 含 benchmark 中不存在的单元: "
                 f"{missing[:5]}（gate_pools.json 与题库失配）"
             )
-        ladder_items = [self._gate_questions_by_id[qid] for qid in ladder_qids]
+        # 单元展开为逐题（unit 内成员顺序保持），下游 validate 再按阶梯序聚合回单元。
+        ladder_items = [q for uid in ladder_unit_ids for q in self._gate_units_by_id[uid].questions]
         base_skill_content = (self._paths.skills_dir / record.target_file).read_text(
             encoding="utf-8"
         )
@@ -1799,7 +1812,10 @@ class Runner:
         obs.update({r["question_id"]: r["prediction"] == r["answer"] for r in slow_rows})
         for extra_rows in extra_rows_lists:
             obs.update({r["question_id"]: r["prediction"] == r["answer"] for r in extra_rows})
-        state.gate_pools.update_probs(obs, gamma=self._config.gate_gamma_decay)
+        # 逐题观测折叠成单元观测后按 unit_id 匹配更新（防按 qid 匹配 pair 失效致 EMA 停摆）。
+        state.gate_pools.update_probs(
+            obs, self._gate_units_by_id, gamma=self._config.gate_gamma_decay
+        )
         state.gate_pools.save(self._config.workspace_dir / "gate_pools.json")
         state.gate_epoch_observed = True
 

tests/unit/test_gate_ladder_unit_migration.py

@@ -0,0 +1,295 @@
+"""gate_ladder.py 单元化迁移测试（核心算法保真 #5）。
+
+验证 gate 信息量阶梯从"逐题"迁移到"unit"粒度后，冷启动 2:1 错优先交错、
+gamma-EMA、Beta 先验、schema_version 门控等语义"只换键、不改公式/比例/顺序"：
+
+(a) LadderEntry 按 unit_id 键（AR pair 折叠为一个阶梯单元）；
+(b) 冷启动"错优先 2:1"以 unit 为单位，unit 错 = P 或 Q 任一错；
+(c) update_probs 观测先折叠成 unit 再匹配（防按 qid 匹配失效致 EMA 停摆）；
+(d) GatePools.save/load 带 schema_version，存量无版本 json 明确报错（不静默混用）；
+(e) BaselineCache 键含 unit_id（pair 的 unit_id 与成员 qid 不同）。
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+import pytest
+
+from app.harness.gate_ladder import (
+    SCHEMA_VERSION,
+    BaselineCache,
+    GatePools,
+    LadderEntry,
+    build_cold_entries,
+)
+from app.harness.question_units import build_units
+from core.types import GeneratedQuestion
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+# ── 构造工具 ──────────────────────────────────────────────────────────
+
+
+def _single(qid: str, task_type: str = "AR") -> GeneratedQuestion:
+    """构造 single 题（unit_id 回填为 question_id）。"""
+    return GeneratedQuestion(
+        question_id=qid,
+        video_id="v1",
+        task_type=task_type,
+        question="dummy",
+        options=("A", "B", "C", "D"),
+        answer="A",
+        source_nodes=("n1",),
+        difficulty="easy",
+    )
+
+
+def _pair(pair_id: str, task_type: str = "AR") -> list[GeneratedQuestion]:
+    """构造一条 AR 孪生对（original + mirror），共享 pair_id 即 unit_id。"""
+    base = {
+        "video_id": "v1",
+        "task_type": task_type,
+        "question": "dummy",
+        "options": ("A", "B", "C", "D"),
+        "answer": "A",
+        "source_nodes": ("n1",),
+        "difficulty": "easy",
+        "pair_id": pair_id,
+        "flip_axis": "before_after",
+    }
+    return [
+        GeneratedQuestion(question_id=f"{pair_id}_o", question_role="pair_original", **base),
+        GeneratedQuestion(question_id=f"{pair_id}_m", question_role="pair_mirror", **base),
+    ]
+
+
+# ── (a) LadderEntry 按 unit_id ─────────────────────────────────────────
+
+
+class TestLadderEntryKeyedByUnit:
+    """LadderEntry 以 unit_id 为键；AR pair 折叠为一个阶梯单元。"""
+
+    def test_entry_has_unit_id(self) -> None:
+        """LadderEntry 暴露 unit_id 字段。"""
+        e = LadderEntry("u1", 0.5)
+        assert e.unit_id == "u1"
+
+    def test_pair_collapses_to_single_entry(self) -> None:
+        """一条孪生对（2 题）在阶梯中只产生 1 个 unit 条目（键=pair_id）。"""
+        units = build_units(_pair("pr1"))
+        correctness = {"pr1_o": True, "pr1_m": True}
+        entries = build_cold_entries(units, correctness, probe_quota=0.0, seed=1)
+        assert len(entries) == 1
+        assert entries[0].unit_id == "pr1"
+
+
+# ── (b) 冷启动 2:1 错优先 + unit 错 = P 或 Q 任一错 ───────────────────────
+
+
+class TestColdStartUnit:
+    """冷启动排序在 unit 粒度保持 2:1 交错与 Beta 先验；unit 错 = 任一成员错。"""
+
+    def test_unit_wrong_if_any_member_wrong(self) -> None:
+        """pair 中任一成员错 → 该 unit 判错（p_hat=1/3）；全对才判对（2/3）。"""
+        units = build_units(_pair("wrong") + _pair("right"))
+        # wrong: original 对、mirror 错 → unit 错；right: 两题均对 → unit 对
+        correctness = {
+            "wrong_o": True,
+            "wrong_m": False,
+            "right_o": True,
+            "right_m": True,
+        }
+        entries = build_cold_entries(units, correctness, probe_quota=0.0, seed=3)
+        p_by_unit = {e.unit_id: e.p_hat for e in entries}
+        assert p_by_unit["wrong"] == pytest.approx(1 / 3)
+        assert p_by_unit["right"] == pytest.approx(2 / 3)
+
+    def test_two_to_one_interleave_over_units(self) -> None:
+        """6 错 unit + 3 对 unit（含 pair），probe_quota=0 → 交错序 W W R W W R W W R。
+
+        交错是 unit 粒度：pair 折叠成一个 unit 参与交错，序列长度为 9（unit 数），
+        而非 18（题数），证明 2:1 比例语义只换键不改。
+        """
+        wrong_units_q: list[GeneratedQuestion] = []
+        for i in range(6):
+            wrong_units_q += _pair(f"w{i}")  # 6 个 pair unit
+        right_units_q = [_single(f"r{i}") for i in range(3)]  # 3 个 single unit
+        units = build_units(wrong_units_q + right_units_q)
+
+        correctness: dict[str, bool] = {}
+        for i in range(6):
+            correctness[f"w{i}_o"] = True
+            correctness[f"w{i}_m"] = False  # 任一错 → unit 错
+        for i in range(3):
+            correctness[f"r{i}"] = True
+
+        entries = build_cold_entries(units, correctness, probe_quota=0.0, seed=42)
+        assert len(entries) == 9
+        # unit 错 = p_hat≈1/3；unit 对 = p_hat≈2/3
+        pattern = ["W" if e.p_hat < 0.5 else "R" for e in entries]
+        assert pattern == ["W", "W", "R", "W", "W", "R", "W", "W", "R"]
+
+    def test_probe_at_tail_unit(self) -> None:
+        """probe_quota>0 时从错 unit 抽探针追加梯尾（按 unit 抽，非按题）。"""
+        wrong_q: list[GeneratedQuestion] = []
+        for i in range(10):
+            wrong_q += _pair(f"w{i}")
+        right_q = [_single(f"r{i}") for i in range(2)]
+        units = build_units(wrong_q + right_q)
+        correctness = {}
+        for i in range(10):
+            correctness[f"w{i}_o"] = False
+            correctness[f"w{i}_m"] = False
+        for i in range(2):
+            correctness[f"r{i}"] = True
+
+        entries = build_cold_entries(units, correctness, probe_quota=0.3, seed=7)
+        # 10 错 unit * 0.3 = 3 个探针 unit 在尾部，均为错 unit
+        tail = entries[-3:]
+        for e in tail:
+            assert e.p_hat < 0.5
+
+
+# ── (c) update_probs 折叠成 unit 再匹配 ──────────────────────────────────
+
+
+class TestUpdateProbsFold:
+    """gamma-EMA 更新前先把逐题观测折叠成 unit 观测（AR pair 双向 AND）。"""
+
+    def _pools_and_units(self) -> tuple[GatePools, dict]:
+        """构造含一个 pair unit + 一个 single unit 的池与 unit 索引。"""
+        units = build_units(_pair("pr1") + [_single("s1")])
+        units_by_id = {u.unit_id: u for u in units}
+        entries = {"AR": [LadderEntry("pr1", 0.5), LadderEntry("s1", 0.5)]}
+        return GatePools(entries=entries, seed=0, fingerprint="x"), units_by_id
+
+    def test_pair_updates_after_fold(self) -> None:
+        """pair 两成员均观测 → 折叠成 unit 观测 → EMA 更新（不停摆）。"""
+        pools, units_by_id = self._pools_and_units()
+        # pair 两题均对 → unit 对(1.0)；single 错(0.0)
+        per_q = {"pr1_o": True, "pr1_m": True, "s1": False}
+        pools.update_probs(per_q, units_by_id, gamma=0.8)
+        p = {e.unit_id: e.p_hat for e in pools.entries["AR"]}
+        assert p["pr1"] == pytest.approx(0.8 * 0.5 + 0.2 * 1.0)  # 0.6
+        assert p["s1"] == pytest.approx(0.8 * 0.5 + 0.2 * 0.0)  # 0.4
+
+    def test_pair_wrong_if_any_member_wrong(self) -> None:
+        """pair 任一成员错 → unit 观测为错(0.0)，EMA 向下。"""
+        pools, units_by_id = self._pools_and_units()
+        per_q = {"pr1_o": True, "pr1_m": False, "s1": True}
+        pools.update_probs(per_q, units_by_id, gamma=0.8)
+        p = {e.unit_id: e.p_hat for e in pools.entries["AR"]}
+        assert p["pr1"] == pytest.approx(0.8 * 0.5 + 0.2 * 0.0)  # 0.4
+
+    def test_partial_pair_observation_skips_update(self) -> None:
+        """pair 只观测到半个成员 → 无法折叠 → 该 unit p_hat 不变（不半 pair 污染）。"""
+        pools, units_by_id = self._pools_and_units()
+        per_q = {"pr1_o": True}  # 缺 mirror
+        pools.update_probs(per_q, units_by_id, gamma=0.8)
+        p = {e.unit_id: e.p_hat for e in pools.entries["AR"]}
+        assert p["pr1"] == pytest.approx(0.5)  # 未更新
+
+    def test_qid_keyed_observation_does_not_match_pair(self) -> None:
+        """若观测里错用 pair_id 之外的裸 qid 键、且未提供 unit 折叠，pair 不应被误更新。
+
+        证明"必须折叠"：单 single unit 用其自身 qid 可更新，pair 需 unit 折叠。
+        """
+        pools, units_by_id = self._pools_and_units()
+        # 只给 single 的观测，pair 两成员均无观测
+        per_q = {"s1": True}
+        pools.update_probs(per_q, units_by_id, gamma=0.8)
+        p = {e.unit_id: e.p_hat for e in pools.entries["AR"]}
+        assert p["pr1"] == pytest.approx(0.5)  # pair 未观测 → 不变
+        assert p["s1"] == pytest.approx(0.8 * 0.5 + 0.2 * 1.0)  # single 更新
+
+
+# ── (d) schema_version 门控 ─────────────────────────────────────────────
+
+
+class TestSchemaVersion:
+    """GatePools.save/load 带 schema_version；存量无版本 json 明确报错。"""
+
+    def test_save_writes_schema_version_and_unit_id(self, tmp_path: Path) -> None:
+        """save 落盘含 schema_version 且 entries 用 unit_id 键。"""
+        entries = {"AR": [LadderEntry("pr1", 0.33), LadderEntry("s1", 0.67)]}
+        pools = GatePools(entries=entries, seed=1, fingerprint="fp")
+        path = tmp_path / "gate_pools.json"
+        pools.save(path)
+
+        raw = json.loads(path.read_text(encoding="utf-8"))
+        assert raw["schema_version"] == SCHEMA_VERSION
+        assert raw["entries"]["AR"][0]["unit_id"] == "pr1"
+        assert "question_id" not in raw["entries"]["AR"][0]
+
+    def test_save_load_roundtrip(self, tmp_path: Path) -> None:
+        """save → load 往返保真（unit_id + p_hat + seed + fingerprint）。"""
+        entries = {"AR": [LadderEntry("pr1", 0.4)]}
+        pools = GatePools(entries=entries, seed=9, fingerprint="fp2")
+        path = tmp_path / "gate_pools.json"
+        pools.save(path)
+        loaded = GatePools.load(path)
+        assert loaded.seed == 9
+        assert loaded.fingerprint == "fp2"
+        assert loaded.entries["AR"][0].unit_id == "pr1"
+        assert loaded.entries["AR"][0].p_hat == pytest.approx(0.4)
+
+    def test_load_legacy_without_schema_version_raises(self, tmp_path: Path) -> None:
+        """存量无 schema_version（旧 qid 键）→ 明确报错，不静默混用。"""
+        legacy = {
+            "seed": 1,
+            "fingerprint": "fp",
+            "entries": {"AR": [{"question_id": "q1", "p_hat": 0.5}]},
+        }
+        path = tmp_path / "gate_pools.json"
+        path.write_text(json.dumps(legacy), encoding="utf-8")
+        with pytest.raises(RuntimeError, match="schema_version"):
+            GatePools.load(path)
+
+    def test_load_wrong_schema_version_raises(self, tmp_path: Path) -> None:
+        """schema_version 不匹配 → 明确报错。"""
+        bad = {
+            "schema_version": SCHEMA_VERSION + 99,
+            "seed": 1,
+            "fingerprint": "fp",
+            "entries": {"AR": [{"unit_id": "pr1", "p_hat": 0.5}]},
+        }
+        path = tmp_path / "gate_pools.json"
+        path.write_text(json.dumps(bad), encoding="utf-8")
+        with pytest.raises(RuntimeError, match="schema_version"):
+            GatePools.load(path)
+
+
+# ── (e) BaselineCache 键含 unit_id ─────────────────────────────────────
+
+
+class TestBaselineCacheUnitKey:
+    """BaselineCache 以 unit_id 为第四维；pair 的 unit_id 与成员 qid 区分。"""
+
+    def test_unit_id_key_distinct_from_member_qid(self, tmp_path: Path) -> None:
+        """pair unit_id（=pair_id）与其成员 qid 是不同缓存键。"""
+        cache = BaselineCache(tmp_path / "baseline_cache.json")
+        cache.put("AR", "h1", "v1", "pr1", True)  # unit_id=pair_id
+        assert cache.get("AR", "h1", "v1", "pr1") is True
+        # 成员 qid 不是同一键 → miss
+        assert cache.get("AR", "h1", "v1", "pr1_o") is None
+        assert cache.get("AR", "h1", "v1", "pr1_m") is None
+
+
+# ── ladder_for 排除按 unit ─────────────────────────────────────────────
+
+
+class TestLadderForExcludeUnit:
+    """ladder_for 返回 unit_id 序，按 unit 排除（防半 pair 灌入）。"""
+
+    def test_exclude_units_filters_whole_unit(self) -> None:
+        """exclude_units 命中的 unit 被整体排除，返回 unit_id 列表。"""
+        entries = {"AR": [LadderEntry("pr1", 0.5), LadderEntry("s1", 0.4), LadderEntry("s2", 0.6)]}
+        pools = GatePools(entries=entries, seed=0, fingerprint="x")
+        result = pools.ladder_for("AR", exclude_units={"pr1"}, p_low=0.0, p_high=1.0, cold=True)
+        assert "pr1" not in result
+        assert "s1" in result
+        assert "s2" in result

tests/unit/test_harness_gate_ladder.py

@@ -20,6 +20,7 @@ from app.harness.gate_ladder import (
     order_ladder,
     skill_hash,
 )
+from app.harness.question_units import build_units
 from core.types import GeneratedQuestion
 
 if TYPE_CHECKING:
@@ -43,6 +44,11 @@ def _make_q(qid: str, task_type: str = "AR") -> GeneratedQuestion:
     )
 
 
+def _units(questions: list[GeneratedQuestion]) -> list:
+    """把题目列表折叠为单元列表（single 题 unit_id 等于 question_id）。"""
+    return build_units(questions)
+
+
 # ── 冷启动 ────────────────────────────────────────────────────────────
 
 
@@ -50,52 +56,52 @@ class TestColdStart:
     """冷启动排序：2:1 交错 + 探针插尾 + Beta(1,1) 平滑。"""
 
     def test_cold_start_interleaving(self) -> None:
-        """错题:对题 = 2:1 交错顺序。
+        """错 unit:对 unit = 2:1 交错顺序。
 
         6 错 3 对（probe_quota=0 无探针）→ 交错序应为 W W R W W R W W R。
         """
         wrong_ids = [f"w{i}" for i in range(6)]
         right_ids = [f"r{i}" for i in range(3)]
-        questions = [_make_q(qid) for qid in wrong_ids + right_ids]
+        units = _units([_make_q(qid) for qid in wrong_ids + right_ids])
         correctness = dict.fromkeys(wrong_ids, False)
         correctness.update(dict.fromkeys(right_ids, True))
 
-        entries = build_cold_entries(questions, correctness, probe_quota=0.0, seed=42)
+        entries = build_cold_entries(units, correctness, probe_quota=0.0, seed=42)
 
         assert len(entries) == 9
         # 验证 2:1 交错模式（seed 固定后 shuffle 结果确定）
-        pattern = ["W" if not correctness[e.question_id] else "R" for e in entries]
+        pattern = ["W" if not correctness[e.unit_id] else "R" for e in entries]
         # 前 9 个交错应为 W W R W W R W W R
         assert pattern == ["W", "W", "R", "W", "W", "R", "W", "W", "R"]
 
     def test_cold_start_p_hat_beta(self) -> None:
         """p_hat 遵循 Beta(1,1) 平滑：错=1/3，对=2/3。"""
-        questions = [_make_q("q1"), _make_q("q2")]
+        units = _units([_make_q("q1"), _make_q("q2")])
         correctness = {"q1": False, "q2": True}
 
-        entries = build_cold_entries(questions, correctness, probe_quota=0.0, seed=0)
+        entries = build_cold_entries(units, correctness, probe_quota=0.0, seed=0)
 
-        p_map = {e.question_id: e.p_hat for e in entries}
+        p_map = {e.unit_id: e.p_hat for e in entries}
         assert p_map["q1"] == pytest.approx(1 / 3)
         assert p_map["q2"] == pytest.approx(2 / 3)
 
     def test_cold_start_probe_at_tail(self) -> None:
-        """probe_quota > 0 时探针题追加在尾部。"""
+        """probe_quota > 0 时探针 unit 追加在尾部。"""
         wrong_ids = [f"w{i}" for i in range(10)]
         right_ids = [f"r{i}" for i in range(2)]
-        questions = [_make_q(qid) for qid in wrong_ids + right_ids]
+        units = _units([_make_q(qid) for qid in wrong_ids + right_ids])
         correctness = dict.fromkeys(wrong_ids, False)
         correctness.update(dict.fromkeys(right_ids, True))
 
-        entries = build_cold_entries(questions, correctness, probe_quota=0.3, seed=7)
+        entries = build_cold_entries(units, correctness, probe_quota=0.3, seed=7)
 
         # 10 错 * 0.3 = 3 个探针在尾部
         n_probe = int(10 * 0.3)
         assert n_probe == 3
-        # 尾部 3 个都应为错题
+        # 尾部 3 个都应为错 unit
         tail = entries[-n_probe:]
         for e in tail:
-            assert not correctness[e.question_id]
+            assert not correctness[e.unit_id]
 
 
 # ── warm 排序 ──────────────────────────────────────────────────────────
@@ -113,9 +119,9 @@ class TestWarmOrdering:
             LadderEntry("d", 0.3),
         ]
         ordered = order_ladder(entries, p_low=0.0, p_high=1.0)
-        assert ordered[0].question_id == "b"  # 0.5*(1-0.5)=0.25 最高
+        assert ordered[0].unit_id == "b"  # 0.5*(1-0.5)=0.25 最高
         # d: 0.3*0.7=0.21, a: 0.1*0.9=0.09, c: 0.9*0.1=0.09
-        assert ordered[1].question_id == "d"
+        assert ordered[1].unit_id == "d"
 
     def test_warm_filter_bounds(self) -> None:
         """p_hat 不在 [p_low, p_high] 区间的题被剔除。"""
@@ -125,7 +131,7 @@ class TestWarmOrdering:
             LadderEntry("high", 0.95),
         ]
         ordered = order_ladder(entries, p_low=0.1, p_high=0.9)
-        ids = [e.question_id for e in ordered]
+        ids = [e.unit_id for e in ordered]
         assert "mid" in ids
         assert "low" not in ids
         assert "high" not in ids
@@ -155,9 +161,9 @@ class TestGatePoolsPersistence:
         assert loaded.seed == 42
         assert loaded.fingerprint == "abc123"
         assert len(loaded.entries["AR"]) == 2
-        assert loaded.entries["AR"][0].question_id == "q1"
+        assert loaded.entries["AR"][0].unit_id == "q1"
         assert loaded.entries["AR"][0].p_hat == pytest.approx(0.33)
-        assert loaded.entries["CR"][0].question_id == "q3"
+        assert loaded.entries["CR"][0].unit_id == "q3"
 
     def test_gate_pools_fingerprint_mismatch(self, tmp_path: Path) -> None:
         """指纹不一致 -> RuntimeError（不静默重建）。"""
@@ -196,8 +202,8 @@ class TestGatePoolsPersistence:
 class TestLadderFor:
     """ladder_for 取题序与排除逻辑。"""
 
-    def test_ladder_for_excludes_qids(self) -> None:
+    def test_ladder_for_excludes_units(self) -> None:
-        """exclude_qids 中的题被排除。"""
+        """exclude_units 中的单元被排除。"""
         entries = {
             "AR": [
                 LadderEntry("q1", 0.5),
@@ -206,7 +212,7 @@ class TestLadderFor:
             ],
         }
         pools = GatePools(entries=entries, seed=0, fingerprint="x")
-        result = pools.ladder_for("AR", exclude_qids={"q2"}, p_low=0.0, p_high=1.0, cold=True)
+        result = pools.ladder_for("AR", exclude_units={"q2"}, p_low=0.0, p_high=1.0, cold=True)
         assert "q2" not in result
         assert "q1" in result
         assert "q3" in result
@@ -236,28 +242,30 @@ class TestLadderFor:
 
 
 class TestGammaEMA:
-    """gamma-EMA 更新 p_hat。"""
+    """gamma-EMA 更新 p_hat（single 单元：unit_id 等于 question_id）。"""
 
     def test_gamma_ema_update(self) -> None:
         """p_hat <- gamma * p_hat + (1-gamma) * obs。"""
         entries = {"AR": [LadderEntry("q1", 0.5)]}
         pools = GatePools(entries=entries, seed=0, fingerprint="x")
+        units_by_id = {u.unit_id: u for u in _units([_make_q("q1")])}
 
         # 观测为正确(1.0), gamma=0.8
-        pools.update_probs({"q1": True}, gamma=0.8)
+        pools.update_probs({"q1": True}, units_by_id, gamma=0.8)
         expected = 0.8 * 0.5 + 0.2 * 1.0  # 0.6
         assert pools.entries["AR"][0].p_hat == pytest.approx(expected)
 
         # 再次观测为错误(0.0), gamma=0.8
-        pools.update_probs({"q1": False}, gamma=0.8)
+        pools.update_probs({"q1": False}, units_by_id, gamma=0.8)
         expected2 = 0.8 * expected + 0.2 * 0.0  # 0.48
         assert pools.entries["AR"][0].p_hat == pytest.approx(expected2)
 
     def test_update_probs_no_observation_unchanged(self) -> None:
-        """无观测的题 p_hat 不变。"""
+        """无观测的单元 p_hat 不变。"""
         entries = {"AR": [LadderEntry("q1", 0.5), LadderEntry("q2", 0.3)]}
         pools = GatePools(entries=entries, seed=0, fingerprint="x")
-        pools.update_probs({"q1": True}, gamma=0.9)
+        units_by_id = {u.unit_id: u for u in _units([_make_q("q1"), _make_q("q2")])}
+        pools.update_probs({"q1": True}, units_by_id, gamma=0.9)
         assert pools.entries["AR"][1].p_hat == pytest.approx(0.3)
 
 
@@ -290,7 +298,8 @@ class TestLeakPrevention:
 
         # 只有普通 run 的观测进入 update_probs
         assert filtered == {"q1": True}
-        pools.update_probs(filtered, gamma=0.8)
+        units_by_id = {u.unit_id: u for u in _units([_make_q("q1")])}
+        pools.update_probs(filtered, units_by_id, gamma=0.8)
         expected = 0.8 * 0.5 + 0.2 * 1.0
         assert pools.entries["AR"][0].p_hat == pytest.approx(expected)