feat(batching): unit 粒度切分——pair 整锁 + 单元级分桶 + 非 AR 独立 rng

build_batches 改以 QuestionUnit 为原子调度单元：孪生对 2 题整锁进同一 batch、
按单元级正确性（双向 AND）落 correct/error 桶，不再因 P 对 Q 错被劈或被 FFD 拆箱。

- 非 AR（single）用 random.Random(seed) 复现旧逐题算法确切 draw 序列，AR（pair）
  用 _rng_ns(seed,"AR") SHA-256 派生独立流；二者 draw 流互不干扰，故 AR 折叠不改变
  非 AR 抽样/洗牌序列——纯非 AR 输入 build_batches 结果与引入 QuestionUnit 前逐字节一致。
- FFD 容量按 unit.size（pair 占 2），round-robin 遇碎片新开 bin 兜底而非报错。
- _select_mixed_by_task_type 分流各跑一次后合并，大类洗牌按 kind 拆分各用对应 rng。

新增黄金测试 test_batching_pair_lock.py 覆盖三条铁律（同 batch / 单元分桶 /
非 AR byte-identical + draw 流独立）；既有 batching 测试全绿。

app/harness/batching.py

@@ -1,13 +1,37 @@
-"""混合 mini-batch 切分：大类打散、小类整锁，供 runner 每 step 处理一个 batch。"""
+"""混合 mini-batch 切分：以 QuestionUnit 为最小调度粒度，大类打散、小类整锁。
+
+供 runner 每 step 处理一个 batch。孪生对（AR pair）作为 2 题单元整锁不拆、按单元级
+正确性分桶；非 AR single 单元的抽样/洗牌 draw 流与"引入 QuestionUnit 前"的旧逐题算法
+逐字节一致（AR 折叠不干扰非 AR draw 流）。
+"""
 
 from __future__ import annotations
 
+import hashlib
 import math
 import random
 from typing import TYPE_CHECKING
 
+from app.harness.question_units import build_units, flatten_units, unit_correctness
+
 if TYPE_CHECKING:
-    from core.types import GeneratedQuestion
+    from core.types import GeneratedQuestion, QuestionUnit
+
+
+def _rng_ns(seed: int, ns: str) -> random.Random:
+    """由 (seed, 命名空间) 稳定派生独立随机数发生器。
+
+    用 SHA-256 派生而非 Python 内置 ``hash()``——后者受 hash randomization 影响，
+    跨进程不可复现。不同命名空间的 draw 流互不干扰，使 AR 单元折叠不扰动非 AR 抽样。
+
+    参数:
+        seed: 实验随机种子。
+        ns: 命名空间标签（如 "AR"）。
+    返回:
+        以 SHA-256(f"{ns}:{seed}") 前 8 字节为种子的 ``random.Random``。
+    """
+    digest = hashlib.sha256(f"{ns}:{seed}".encode()).digest()
+    return random.Random(int.from_bytes(digest[:8], "big"))
 
 
 def build_batches(
@@ -18,55 +42,59 @@ def build_batches(
     seed: int,
     correct_ratio: float = 0.0,
 ) -> tuple[list[list[GeneratedQuestion]], int]:
-    """把诊断池里的题目切成多个混合 mini-batch。
+    """把诊断池里的题目切成多个混合 mini-batch（以 QuestionUnit 为原子调度单元）。
 
-    当 ``correct_ratio > 0`` 时，按题型为每组错题配比一定数量的正确题，使 batch
+    single 题为 1 题单元，AR pair 孪生对为 2 题单元；同一 pair 的两题整锁进同一 batch，
-    包含正误混合样本（"动量"机制）；``correct_ratio <= 0`` 时退化为纯错题模式。
+    按单元级正确性（双向 AND）分桶。当 ``correct_ratio > 0`` 时，按题型为每组错误单元配比
+    一定数量的正确单元（"动量"机制）；``correct_ratio <= 0`` 时退化为纯错误单元模式。
 
     参数:
-        items: 候选题目全集。
+        items: 候选题目全集（可混含 single 与孪生对成员）。
         correctness: question_id -> 基线是否答对。
-        batch_size: 单个 batch 的样本数上限（> 0）。
+        batch_size: 单个 batch 的题目数上限（> 0，pair 占 2）。
-        min_class_per_batch: 小类判定阈值——题目数 ≤ 此值的题型整组锁进单一
+        min_class_per_batch: 小类判定阈值——单元题目总数 ≤ 此值的题型整组锁进单一
             batch（> 0）。
         seed: 随机种子，保证相同输入产出完全一致的切分。
-        correct_ratio: 正确题占比（0.0 ~ 1.0）。0.0 = 纯错题；0.5 = 错题:正确题 = 1:1。
+        correct_ratio: 正确题占比（0.0 ~ 1.0）。0.0 = 纯错误单元；0.5 = 错:正 = 1:1。
     返回:
-        (非空 mini-batch 列表, selected_count)；无错题时返回 ([], 0)。
+        (非空 mini-batch 列表, selected_count)；无错误单元时返回 ([], 0)。
-        selected_count 是所有 batch 中题目总数。
+        selected_count 是所有 batch 中题目（展开后）总数。
     异常:
         ValueError: batch_size 或 min_class_per_batch < 1, 或
             min_class_per_batch >= batch_size（破坏小类整组装箱不超容的前提）。
     关键实现细节:
-        装箱顺序为「先小类后大类」。小类整组用 first-fit-decreasing 装箱：按组大小
+        非 AR（single）与 AR（pair）各用独立稳定派生的 rng：非 AR 用 ``random.Random(seed)``
-        降序处理（同大小再按 task_type 排序保证确定性），每组放进第一个剩余容量足够
+        （复现旧逐题算法的确切 draw 序列，保证纯非 AR 输入逐字节一致），AR 用
-        的 batch；若现有 batch 都装不下就新开一个空 batch——因小类组大小
+        ``_rng_ns(seed, "AR")``；二者 draw 流互不干扰，故加入/移除 pair 不改变非 AR 的
-        ≤ min_class_per_batch < batch_size，新空 batch 必能容纳，故小类装箱永不抛
+        抽样/洗牌序列。抽样在合并前按流分别进行（``_select_mixed_by_task_type`` 各跑一次），
-        ValueError，且保证整组不拆。再把大类样本（seed 确定性 shuffle 后）round-robin
+        大类洗牌按单元 kind 拆分后各用对应流。装箱顺序「先小类后大类」：小类整组
-        分发到所有现存 batch 填充剩余容量。这样小类聚集于单 batch、大类散布多 batch
+        first-fit-decreasing（容量按单元 ``size`` 计，pair 占 2）装入首个容得下的 batch，
-        且与小类共箱，自然产生多类混合 batch（纯类切片会被 multiclass 断言拒绝）。
+        装不下新开 bin；大类洗牌后 round-robin 分发，遇碎片（size-2 单元放不进任一现存
-        nb = ceil(总题数/batch_size) 是初始 batch 数下界估计而非硬上限：小类装箱可能
+        batch 的剩余容量）新开 bin 兜底而非报错。最终每个 batch 展开回题目列表。
-        新开 bin 使实际 batch 数超过 nb。每次新开 bin 都意味着总容量随之增加，故总容量
+        题型按名称排序处理以保证跨运行确定性。
-        恒 ≥ 总题数，大类 round-robin 跳过满箱后仍能放下全部样本，不会违反 batch_size
-        上限。题型按名称排序处理以保证跨运行确定性，不依赖 dict 遍历顺序。
     """
     _validate_params(batch_size, min_class_per_batch)
 
-    rng = random.Random(seed)
+    # 非 AR 复现旧版 random.Random(seed) 的确切序列以满足黄金 byte-identity；
-    grouped = _select_mixed_by_task_type(items, correctness, correct_ratio, rng)
+    # AR 走独立命名空间派生流，二者互不干扰。
-    total = sum(len(g) for g in grouped.values())
+    rng_nonar = random.Random(seed)
+    rng_ar = _rng_ns(seed, "AR")
+
+    grouped = _group_units_by_task_type(items, correctness, correct_ratio, rng_nonar, rng_ar)
+
+    total = sum(_group_load(g) for g in grouped.values())
     if total == 0:
         return [], 0
 
     nb = max(1, math.ceil(total / batch_size))
-    batches: list[list[GeneratedQuestion]] = [[] for _ in range(nb)]
+    batches: list[list[QuestionUnit]] = [[] for _ in range(nb)]
 
     small, large = _split_by_size(grouped, min_class_per_batch)
     for group in _small_groups_decreasing(small):
         _pack_small_class(batches, group, batch_size)
-    _distribute_large_classes(batches, large, batch_size, rng)
+    _distribute_large_classes(batches, large, batch_size, rng_nonar, rng_ar)
 
-    result = [b for b in batches if b]
+    result = [flatten_units(b) for b in batches if b]
     selected_count = sum(len(b) for b in result)
     return result, selected_count
 
@@ -74,7 +102,7 @@ def build_batches(
 def _validate_params(batch_size: int, min_class_per_batch: int) -> None:
     """校验切分参数，非法值直接报错而非用默认值掩盖。
 
-    除各自 >= 1 外，强制 min_class_per_batch < batch_size：小类组大小 ≤
+    除各自 >= 1 外，强制 min_class_per_batch < batch_size：小类组题目总数 ≤
     min_class_per_batch，唯有此前提成立才能保证小类整组放入单一 batch 而不超容；否则
     _pack_small_class 新开的 bin 会装入超 batch_size 的整组，静默违反容量合约。此约束
     与 config._validate_minibatch 一致，是 build_batches 对自身前提的防御性自校验（P5）。
@@ -91,52 +119,129 @@ def _validate_params(batch_size: int, min_class_per_batch: int) -> None:
         )
 
 
-def _split_by_size(
+def _group_units_by_task_type(
-    grouped: dict[str, list[GeneratedQuestion]],
-    min_class_per_batch: int,
-) -> tuple[dict[str, list[GeneratedQuestion]], dict[str, list[GeneratedQuestion]]]:
-    """按错题数把题型分为小类（≤ 阈值）与大类（> 阈值）两组。"""
-    small = {t: g for t, g in grouped.items() if len(g) <= min_class_per_batch}
-    large = {t: g for t, g in grouped.items() if len(g) > min_class_per_batch}
-    return small, large
-
-
-def _select_mixed_by_task_type(
     items: list[GeneratedQuestion],
     correctness: dict[str, bool],
     correct_ratio: float,
-    rng: random.Random,
+    rng_nonar: random.Random,
-) -> dict[str, list[GeneratedQuestion]]:
+    rng_ar: random.Random,
-    """按题型分组，为每组错题按比例采样正确题混入。
+) -> dict[str, list[QuestionUnit]]:
-
+    """把题目聚合为单元并按题型分组：非 AR 与 AR 各走独立 draw 流后合并。
-    只对有错题的题型做混合——无错题的题型不进 batch，即使有正确题。
-    ``correct_ratio <= 0`` 时退化为纯错题模式（向后兼容）。
 
     参数:
         items: 候选题目全集。
         correctness: question_id -> 基线是否答对。
-        correct_ratio: 正确题占比（0.0 ~ 1.0）。
+        correct_ratio: 正确题占比。
-        rng: 随机数发生器，用于采样正确题。
+        rng_nonar: 非 AR（single 单元）抽样用 rng。
+        rng_ar: AR（pair 单元）抽样用 rng。
     返回:
-        task_type -> 该题型的混合题目列表（错题全部 + 按比例采样的正确题）。
+        task_type -> 混合后的单元列表（single 单元在前、pair 单元在后）。
     """
-    errors_by_type: dict[str, list[GeneratedQuestion]] = {}
+    units = build_units(items)
-    correct_by_type: dict[str, list[GeneratedQuestion]] = {}
+    singles = [u for u in units if u.kind == "single"]
-    for q in items:
+    pairs = [u for u in units if u.kind == "pair"]
-        qid = q.question_id
+    grouped_nonar = _select_mixed_by_task_type(singles, correctness, correct_ratio, rng_nonar)
-        if correctness.get(qid) is False:
+    grouped_ar = _select_mixed_by_task_type(pairs, correctness, correct_ratio, rng_ar)
-            errors_by_type.setdefault(q.task_type, []).append(q)
+    return _merge_grouped(grouped_nonar, grouped_ar)
-        elif correctness.get(qid, False):
+
-            correct_by_type.setdefault(q.task_type, []).append(q)
+
+def _group_load(group: list[QuestionUnit]) -> int:
+    """一组单元展开后的题目总数（single 计 1，pair 计 2），即占用的 batch 容量。"""
+    return sum(u.size for u in group)
+
+
+def _batch_load(batch: list[QuestionUnit]) -> int:
+    """一个 batch 内单元展开后的题目总数，用于容量判断。"""
+    return sum(u.size for u in batch)
+
+
+def _merge_grouped(
+    grouped_nonar: dict[str, list[QuestionUnit]],
+    grouped_ar: dict[str, list[QuestionUnit]],
+) -> dict[str, list[QuestionUnit]]:
+    """按 task_type 合并非 AR 与 AR 两条流的分组（single 在前、pair 在后）。
+
+    参数:
+        grouped_nonar: 非 AR（single 单元）分组。
+        grouped_ar: AR（pair 单元）分组。
+    返回:
+        task_type -> 合并后的单元列表；每类 single 单元在前、pair 单元在后，顺序稳定。
+    """
+    merged: dict[str, list[QuestionUnit]] = {}
+    for task_type in sorted({*grouped_nonar, *grouped_ar}):
+        merged[task_type] = grouped_nonar.get(task_type, []) + grouped_ar.get(task_type, [])
+    return merged
+
+
+def _split_by_size(
+    grouped: dict[str, list[QuestionUnit]],
+    min_class_per_batch: int,
+) -> tuple[dict[str, list[QuestionUnit]], dict[str, list[QuestionUnit]]]:
+    """按题目总数（单元展开）把题型分为小类（≤ 阈值）与大类（> 阈值）两组。"""
+    small = {t: g for t, g in grouped.items() if _group_load(g) <= min_class_per_batch}
+    large = {t: g for t, g in grouped.items() if _group_load(g) > min_class_per_batch}
+    return small, large
+
+
+def _classify_unit(unit: QuestionUnit, correctness: dict[str, bool]) -> str | None:
+    """判定单元落入哪个桶：error / correct / None（未知，跳过）。
+
+    参数:
+        unit: 目标单元。
+        correctness: question_id -> 是否答对（缺键视为未知）。
+    返回:
+        "error"（单元级正确性为 False）、"correct"（双向 AND 为 True）；单元内任一题
+        未知（correctness 缺该键）返回 None，与旧逐题算法把未知题排除在错/对两桶之外
+        的语义一致。
+    关键实现:
+        先探测是否有未知题（get 返回 None ⟺ 键缺失，因 correctness 值恒为 bool），
+        全部已知后交由 unit_correctness 计双向 AND（此时 KeyError 不可达）。
+    """
+    if any(correctness.get(q.question_id) is None for q in unit.questions):
+        return None
+    return "correct" if unit_correctness(unit, correctness) else "error"
+
+
+def _select_mixed_by_task_type(
+    units: list[QuestionUnit],
+    correctness: dict[str, bool],
+    correct_ratio: float,
+    rng: random.Random,
+) -> dict[str, list[QuestionUnit]]:
+    """按题型分组，为每组错误单元按比例采样正确单元混入（单元粒度）。
+
+    只对有错误单元的题型做混合——无错误单元的题型不进 batch，即使有正确单元。
+    ``correct_ratio <= 0`` 时退化为纯错误单元模式。本函数只处理单一 draw 流（全 single
+    或全 pair），使非 AR 与 AR 的抽样互不干扰。
+
+    参数:
+        units: 同一流的候选单元（全 single 或全 pair）。
+        correctness: question_id -> 基线是否答对。
+        correct_ratio: 正确题占比（0.0 ~ 1.0）。
+        rng: 本流专用随机数发生器，用于采样正确单元。
+    返回:
+        task_type -> 该题型的混合单元列表（错误单元全部 + 按比例采样的正确单元）。
+    关键实现:
+        n_correct 按错误单元「题目总数」而非单元数计，与旧逐题语义对齐（纯 single 时
+        单元数 == 题目数，采样序列逐字节一致）。
+    """
+    errors_by_type: dict[str, list[QuestionUnit]] = {}
+    correct_by_type: dict[str, list[QuestionUnit]] = {}
+    for unit in units:
+        bucket = _classify_unit(unit, correctness)
+        if bucket == "error":
+            errors_by_type.setdefault(unit.task_type, []).append(unit)
+        elif bucket == "correct":
+            correct_by_type.setdefault(unit.task_type, []).append(unit)
 
     if correct_ratio <= 0:
         return errors_by_type
 
-    # 为每个有错题的 task_type 混入正确题
+    grouped: dict[str, list[QuestionUnit]] = {}
-    grouped: dict[str, list[GeneratedQuestion]] = {}
     for task_type in sorted(errors_by_type):
         errs = errors_by_type[task_type]
-        n_correct = round(len(errs) * correct_ratio / (1 - correct_ratio))
+        n_err = _group_load(errs)
+        n_correct = round(n_err * correct_ratio / (1 - correct_ratio))
         available = correct_by_type.get(task_type, [])
         sampled = (
             list(available) if len(available) <= n_correct else rng.sample(available, n_correct)
@@ -147,94 +252,100 @@ def _select_mixed_by_task_type(
 
 
 def _small_groups_decreasing(
-    small: dict[str, list[GeneratedQuestion]],
+    small: dict[str, list[QuestionUnit]],
-) -> list[list[GeneratedQuestion]]:
+) -> list[list[QuestionUnit]]:
-    """按组大小降序、同大小按 task_type 升序排出小类组（first-fit-decreasing 顺序）。
+    """按组题目总数降序、同大小按 task_type 升序排出小类组（first-fit-decreasing 顺序）。
 
     参数:
-        small: task_type -> 小类错题列表。
+        small: task_type -> 小类单元列表。
     返回:
         排好序的小类组列表；降序处理可降低碎片，确定性 tie-break 保证跨运行一致。
     """
-    return [small[t] for t in sorted(small, key=lambda t: (-len(small[t]), t))]
+    return [small[t] for t in sorted(small, key=lambda t: (-_group_load(small[t]), t))]
 
 
 def _pack_small_class(
-    batches: list[list[GeneratedQuestion]],
+    batches: list[list[QuestionUnit]],
-    group: list[GeneratedQuestion],
+    group: list[QuestionUnit],
     batch_size: int,
 ) -> None:
     """用 first-fit 把一个小类整组放入首个容得下的 batch，装不下则新开 bin（就地修改）。
 
-    因小类组大小 ≤ min_class_per_batch < batch_size，新开的空 batch 必能容纳整组，
+    因小类组题目总数 ≤ min_class_per_batch < batch_size，新开的空 batch 必能容纳整组，
-    故此函数永不抛 ValueError，且整组不拆。
+    故此函数永不抛 ValueError，且整组（含内部 pair 单元）不拆。
 
     参数:
         batches: 当前各 batch（就地追加，必要时 append 新空 batch）。
-        group: 待锁定的小类错题（整组不拆）。
+        group: 待锁定的小类单元组（整组不拆）。
-        batch_size: 单 batch 容量上限。
+        batch_size: 单 batch 题目容量上限。
     """
+    load = _group_load(group)
     for b in batches:
-        if len(b) + len(group) <= batch_size:
+        if _batch_load(b) + load <= batch_size:
             b.extend(group)
             return
     batches.append(list(group))
 
 
 def _distribute_large_classes(
-    batches: list[list[GeneratedQuestion]],
+    batches: list[list[QuestionUnit]],
-    large: dict[str, list[GeneratedQuestion]],
+    large: dict[str, list[QuestionUnit]],
     batch_size: int,
-    rng: random.Random,
+    rng_nonar: random.Random,
+    rng_ar: random.Random,
 ) -> None:
-    """将各大类样本 shuffle 后 round-robin 分发到所有现存 batch（就地修改）。
+    """将各大类单元洗牌后 round-robin 分发到所有现存 batch（就地修改）。
 
     参数:
         batches: 当前各 batch（含小类装箱可能新开的 bin，就地追加）。
-        large: task_type -> 大类错题列表。
+        large: task_type -> 大类单元列表。
-        batch_size: 单 batch 容量上限。
+        batch_size: 单 batch 题目容量上限。
-        rng: 复用的随机数发生器，保证 shuffle 确定性。
+        rng_nonar: 非 AR（single 单元）洗牌用 rng。
-    异常:
+        rng_ar: AR（pair 单元）洗牌用 rng。
-        ValueError: 所有 batch 均满仍有样本未放置（总容量估算异常，合法输入不可达）。
     关键实现细节:
-        轮转范围是「所有现存 batch」而非固定 nb 个——小类装箱新开的 bin 也参与分发。
+        每组按单元 kind 拆成 single 子列与 pair 子列，分别用 rng_nonar / rng_ar 洗牌后
-        总容量 = 现存 batch 数 × batch_size，每次新开 bin 都同步抬高总容量，故总容量恒
+        拼接（single 在前），使非 AR 洗牌 draw 流不受 pair 存在与否影响（纯 single 时
-        ≥ 总错题数，防御性 ValueError 在合法输入下不可达。全局指针在所有大类样本间持续
+        single 子列即整组，复现旧版单一 rng.shuffle 的序列）。全局指针在所有大类单元间
-        轮转（不为每类重置），满箱即跳过，使大类充分散布并与已锁定的小类共箱。题型按名称
+        持续轮转，遇满箱跳过、遇碎片新开 bin。题型按名称排序以保证分发顺序确定。
-        排序以保证分发顺序确定。
     """
-    nb = len(batches)
     pointer = 0
     for task_type in sorted(large):
-        group = list(large[task_type])
+        group = large[task_type]
-        rng.shuffle(group)
+        singles = [u for u in group if u.kind == "single"]
-        for q in group:
+        pairs = [u for u in group if u.kind == "pair"]
-            pointer = _place_round_robin(batches, q, pointer, batch_size, nb)
+        rng_nonar.shuffle(singles)
+        rng_ar.shuffle(pairs)
+        for unit in singles + pairs:
+            pointer = _place_round_robin(batches, unit, pointer, batch_size)
 
 
 def _place_round_robin(
-    batches: list[list[GeneratedQuestion]],
+    batches: list[list[QuestionUnit]],
-    q: GeneratedQuestion,
+    unit: QuestionUnit,
     pointer: int,
     batch_size: int,
-    nb: int,
 ) -> int:
-    """从 pointer 起找第一个未满 batch 放入 q，返回下一次起始指针。
+    """从 pointer 起找第一个容量够放 unit 的 batch 放入，返回下一次起始指针。
 
     参数:
         batches: 当前各 batch（就地追加）。
-        q: 待放置的样本。
+        unit: 待放置的单元（占用 unit.size 个容量）。
         pointer: 本次轮转起始 batch 下标。
-        batch_size: 单 batch 容量上限。
+        batch_size: 单 batch 题目容量上限。
-        nb: batch 总数。
     返回:
         下一次轮转的起始指针（已前移一位）。
-    异常:
+    关键实现:
-        ValueError: 扫描一轮所有 batch 均满（总容量估算异常）。
+        单个单元容量 ≤ batch_size 是前提（pair 占 2，而 batch_size > min_class ≥ 1 ⇒
+        batch_size ≥ 2），故此处断言防御。扫描一轮所有现存 batch 都放不下（size-2 单元
+        遇满地碎片）时新开 bin 兜底而非报错——聚合容量足够但单箱剩余不足是合法碎片场景。
+        纯 single（size 1）永不触发新开分支，故与旧逐题 round-robin 逐字节一致。
     """
+    assert unit.size <= batch_size, f"单元 size={unit.size} 超过 batch_size={batch_size}"
+    nb = len(batches)
     for offset in range(nb):
         idx = (pointer + offset) % nb
-        if len(batches[idx]) < batch_size:
+        if _batch_load(batches[idx]) + unit.size <= batch_size:
-            batches[idx].append(q)
+            batches[idx].append(unit)
             return (idx + 1) % nb
-    raise ValueError("所有 batch 均满仍有样本待放置, 总容量估算异常")
+    batches.append([unit])
+    return len(batches) % len(batches)

tests/unit/test_batching_pair_lock.py

@@ -0,0 +1,304 @@
+"""app/harness/batching.py 的 unit 粒度切分测试（Task 5）。
+
+覆盖 pair 契约在 mini-batch 构建中的三条铁律：
+- (a) 同 pair_id 两题整锁进同一 batch（像小类整组不拆）；
+- (b) pair 按 unit correctness（双向 AND）落 correct/error 桶，不因 P 对 Q 错被劈；
+- (c) 非 AR byte-identical：AR unit 折叠不改变非 AR 的 rng.sample/shuffle 抽样序列——
+      纯非 AR 输入下 build_batches 结果与"引入 QuestionUnit 前"的旧算法逐字节一致。
+"""
+
+from __future__ import annotations
+
+import math
+import random
+
+from app.harness.batching import build_batches
+from core.types import GeneratedQuestion
+
+# ---------------------------------------------------------------------------
+# 辅助构造
+# ---------------------------------------------------------------------------
+
+
+def _make_single(
+    qid: str,
+    task_type: str = "default",
+    video_id: str = "v1",
+) -> GeneratedQuestion:
+    """构造最小 single GeneratedQuestion。"""
+    return GeneratedQuestion(
+        question_id=qid,
+        video_id=video_id,
+        task_type=task_type,
+        question=f"question_{qid}",
+        options=("A. a", "B. b", "C. c", "D. d"),
+        answer="A",
+        source_nodes=("n1",),
+        difficulty="medium",
+        question_role="single",
+    )
+
+
+def _make_pair(
+    pair_id: str,
+    task_type: str = "AR",
+    video_id: str = "v1",
+) -> tuple[GeneratedQuestion, GeneratedQuestion]:
+    """构造一个孪生对（original + mirror），共享 pair_id / flip_axis。"""
+    common = {
+        "video_id": video_id,
+        "task_type": task_type,
+        "question": "?",
+        "options": ("A. a", "B. b", "C. c", "D. d"),
+        "answer": "A",
+        "source_nodes": ("n1",),
+        "difficulty": "medium",
+        "pair_id": pair_id,
+        "flip_axis": "before_after",
+    }
+    original = 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 original, mirror
+
+
+def _batch_of(batches: list[list[GeneratedQuestion]], qid: str) -> int:
+    """返回 qid 所在 batch 的下标；不存在返回 -1。"""
+    for i, b in enumerate(batches):
+        if any(q.question_id == qid for q in b):
+            return i
+    return -1
+
+
+# ---------------------------------------------------------------------------
+# (a) 同 pair_id 两题落同一 batch（整锁不拆）
+# ---------------------------------------------------------------------------
+
+
+class TestPairStaysInSameBatch:
+    """孪生对两题必须整锁进同一 batch，无论散布在多少 single 之间。"""
+
+    def test_single_pair_same_batch(self) -> None:
+        po, pm = _make_pair("pairA", task_type="AR")
+        singles = [_make_single(f"s{i}", task_type="RETRIEVAL") for i in range(10)]
+        items = [*singles, po, pm]
+        correctness = {q.question_id: False for q in items}
+        batches, count = build_batches(
+            items, correctness, batch_size=4, min_class_per_batch=2, seed=3
+        )
+        assert count == 12
+        idx_o = _batch_of(batches, "pairA_o")
+        idx_m = _batch_of(batches, "pairA_m")
+        assert idx_o != -1 and idx_o == idx_m
+
+    def test_many_pairs_each_intact(self) -> None:
+        items: list[GeneratedQuestion] = []
+        for k in range(5):
+            po, pm = _make_pair(f"p{k}", task_type="AR")
+            items.extend([po, pm])
+        items.extend(_make_single(f"s{i}", task_type="SPATIAL") for i in range(6))
+        correctness = {q.question_id: False for q in items}
+        batches, _ = build_batches(items, correctness, batch_size=6, min_class_per_batch=2, seed=11)
+        for k in range(5):
+            idx_o = _batch_of(batches, f"p{k}_o")
+            idx_m = _batch_of(batches, f"p{k}_m")
+            assert idx_o != -1 and idx_o == idx_m, f"pair p{k} 被拆到不同 batch"
+        # 每个 batch 容量不超限（pair 占 2）
+        for b in batches:
+            assert len(b) <= 6
+
+
+# ---------------------------------------------------------------------------
+# (b) pair 按 unit correctness（双向 AND）分桶
+# ---------------------------------------------------------------------------
+
+
+class TestPairBucketedByUnitCorrectness:
+    """P 对 Q 错的 pair 是 error 单元，不因单题分歧被劈到两个桶。"""
+
+    def test_p_correct_q_wrong_pair_is_error_unit(self) -> None:
+        po, pm = _make_pair("pairErr", task_type="AR")
+        # 混一个 single 错题避免空池边界
+        s0 = _make_single("s0", task_type="AR")
+        items = [po, pm, s0]
+        # original 对、mirror 错 → 单元级 AND = 错 → 应作为 error 单元整体进 batch
+        correctness = {"pairErr_o": True, "pairErr_m": False, "s0": False}
+        batches, count = build_batches(
+            items, correctness, batch_size=6, min_class_per_batch=2, seed=0, correct_ratio=0.0
+        )
+        idx_o = _batch_of(batches, "pairErr_o")
+        idx_m = _batch_of(batches, "pairErr_m")
+        # 两题都在（未被"P 对"劈掉）且同 batch
+        assert idx_o != -1 and idx_o == idx_m
+        # 纯错题模式下 pair 单元整体被选入
+        assert count == 3
+
+    def test_both_correct_pair_excluded_in_pure_error_mode(self) -> None:
+        po, pm = _make_pair("pairOk", task_type="AR")
+        s_err = _make_single("s_err", task_type="AR")
+        items = [po, pm, s_err]
+        correctness = {"pairOk_o": True, "pairOk_m": True, "s_err": False}
+        batches, count = build_batches(
+            items, correctness, batch_size=6, min_class_per_batch=2, seed=0, correct_ratio=0.0
+        )
+        # both-correct pair 是 correct 单元，纯错题模式下不入池
+        assert count == 1
+        assert _batch_of(batches, "pairOk_o") == -1
+        assert _batch_of(batches, "pairOk_m") == -1
+        assert _batch_of(batches, "s_err") != -1
+
+    def test_both_correct_pair_stays_intact_when_mixed(self) -> None:
+        """correct_ratio>0 时 both-correct pair 可作为整体 correct 单元混入。"""
+        errs = [_make_single(f"e{i}", task_type="AR") for i in range(4)]
+        po, pm = _make_pair("pairOk", task_type="AR")
+        items = [*errs, po, pm]
+        correctness = {f"e{i}": False for i in range(4)}
+        correctness["pairOk_o"] = True
+        correctness["pairOk_m"] = True
+        batches, _ = build_batches(
+            items, correctness, batch_size=10, min_class_per_batch=2, seed=1, correct_ratio=0.5
+        )
+        idx_o = _batch_of(batches, "pairOk_o")
+        idx_m = _batch_of(batches, "pairOk_m")
+        # 若被混入，两题必须整体同 batch；若未被采样，两题都不在
+        if idx_o == -1:
+            assert idx_m == -1
+        else:
+            assert idx_o == idx_m
+
+
+# ---------------------------------------------------------------------------
+# (c) 非 AR byte-identical：与"引入 QuestionUnit 前"旧算法逐字节一致
+# ---------------------------------------------------------------------------
+
+
+def _reference_select_mixed(
+    items: list[GeneratedQuestion],
+    correctness: dict[str, bool],
+    correct_ratio: float,
+    rng: random.Random,
+) -> dict[str, list[GeneratedQuestion]]:
+    """旧版 _select_mixed_by_task_type 的忠实副本（逐题、单一 rng）。"""
+    errors_by_type: dict[str, list[GeneratedQuestion]] = {}
+    correct_by_type: dict[str, list[GeneratedQuestion]] = {}
+    for q in items:
+        qid = q.question_id
+        if correctness.get(qid) is False:
+            errors_by_type.setdefault(q.task_type, []).append(q)
+        elif correctness.get(qid, False):
+            correct_by_type.setdefault(q.task_type, []).append(q)
+    if correct_ratio <= 0:
+        return errors_by_type
+    grouped: dict[str, list[GeneratedQuestion]] = {}
+    for task_type in sorted(errors_by_type):
+        errs = errors_by_type[task_type]
+        n_correct = round(len(errs) * correct_ratio / (1 - correct_ratio))
+        available = correct_by_type.get(task_type, [])
+        sampled = (
+            list(available) if len(available) <= n_correct else rng.sample(available, n_correct)
+        )
+        grouped[task_type] = errs + sampled
+    return grouped
+
+
+def _reference_build_batches(
+    items: list[GeneratedQuestion],
+    correctness: dict[str, bool],
+    batch_size: int,
+    min_class_per_batch: int,
+    seed: int,
+    correct_ratio: float = 0.0,
+) -> tuple[list[list[GeneratedQuestion]], int]:
+    """引入 QuestionUnit 前的旧版 build_batches 的忠实副本（逐题、单一 rng）。"""
+    rng = random.Random(seed)
+    grouped = _reference_select_mixed(items, correctness, correct_ratio, rng)
+    total = sum(len(g) for g in grouped.values())
+    if total == 0:
+        return [], 0
+    nb = max(1, math.ceil(total / batch_size))
+    batches: list[list[GeneratedQuestion]] = [[] for _ in range(nb)]
+    small = {t: g for t, g in grouped.items() if len(g) <= min_class_per_batch}
+    large = {t: g for t, g in grouped.items() if len(g) > min_class_per_batch}
+    for t in sorted(small, key=lambda t: (-len(small[t]), t)):
+        group = small[t]
+        placed = False
+        for b in batches:
+            if len(b) + len(group) <= batch_size:
+                b.extend(group)
+                placed = True
+                break
+        if not placed:
+            batches.append(list(group))
+    nb_live = len(batches)
+    pointer = 0
+    for task_type in sorted(large):
+        group = list(large[task_type])
+        rng.shuffle(group)
+        for q in group:
+            for offset in range(nb_live):
+                idx = (pointer + offset) % nb_live
+                if len(batches[idx]) < batch_size:
+                    batches[idx].append(q)
+                    pointer = (idx + 1) % nb_live
+                    break
+    result = [b for b in batches if b]
+    return result, sum(len(b) for b in result)
+
+
+class TestNonARByteIdentical:
+    """纯 single 输入下 build_batches 与旧逐题算法逐字节一致（黄金对照）。"""
+
+    def _assert_identical(
+        self,
+        items: list[GeneratedQuestion],
+        correctness: dict[str, bool],
+        batch_size: int,
+        min_class_per_batch: int,
+        seed: int,
+        correct_ratio: float,
+    ) -> None:
+        got_batches, got_count = build_batches(
+            items, correctness, batch_size, min_class_per_batch, seed, correct_ratio
+        )
+        ref_batches, ref_count = _reference_build_batches(
+            items, correctness, batch_size, min_class_per_batch, seed, correct_ratio
+        )
+        got_ids = [[q.question_id for q in b] for b in got_batches]
+        ref_ids = [[q.question_id for q in b] for b in ref_batches]
+        assert got_ids == ref_ids
+        assert got_count == ref_count
+
+    def test_pure_errors_identical(self) -> None:
+        items = [_make_single(f"q{i}", task_type=f"type_{i % 3}") for i in range(20)]
+        correctness = {f"q{i}": False for i in range(20)}
+        self._assert_identical(items, correctness, 5, 2, 42, 0.0)
+
+    def test_mixed_ratio_identical(self) -> None:
+        items = [_make_single(f"q{i}", task_type=f"type_{i % 4}") for i in range(40)]
+        correctness = {f"q{i}": (i % 3 == 0) for i in range(40)}
+        self._assert_identical(items, correctness, 8, 3, 7, 0.5)
+
+    def test_large_round_robin_identical(self) -> None:
+        items = [_make_single(f"q{i}", task_type="big") for i in range(30)]
+        correctness = {f"q{i}": False for i in range(30)}
+        self._assert_identical(items, correctness, 6, 2, 99, 0.0)
+
+    def test_ar_folding_does_not_shift_nonar_draws(self) -> None:
+        """加入 AR pair 不改变非 AR single 的抽样序列（draw 流独立）。"""
+        singles = [_make_single(f"q{i}", task_type=f"type_{i % 3}") for i in range(24)]
+        s_correctness = {f"q{i}": (i % 4 == 0) for i in range(24)}
+        base, _ = build_batches(singles, s_correctness, 6, 2, 5, 0.5)
+        base_ids = {q.question_id for b in base for q in b}
+
+        # 追加若干 AR pair（不同 task_type），非 AR single 的入选集合应不变
+        items = list(singles)
+        correctness = dict(s_correctness)
+        for k in range(3):
+            po, pm = _make_pair(f"p{k}", task_type="AR")
+            items.extend([po, pm])
+            correctness[po.question_id] = False
+            correctness[pm.question_id] = False
+        with_ar, _ = build_batches(items, correctness, 6, 2, 5, 0.5)
+        with_ar_single_ids = {q.question_id for b in with_ar for q in b if q.pair_id is None}
+        assert with_ar_single_ids == base_ids

tests/unit/test_harness_batching.py

@@ -7,22 +7,23 @@ correctness False vs None 精确匹配。
 
 from __future__ import annotations
 
-import pytest
-
-from core.types import GeneratedQuestion
-from app.harness.batching import (
-    build_batches,
-    _validate_params,
-    _select_mixed_by_task_type,
-)
-
 import random
 
+import pytest
+
+from app.harness.batching import (
+    _select_mixed_by_task_type,
+    _validate_params,
+    build_batches,
+)
+from app.harness.question_units import build_units
+from core.types import GeneratedQuestion
 
 # ---------------------------------------------------------------------------
 # 辅助构造
 # ---------------------------------------------------------------------------
 
+
 def _make_q(
     qid: str,
     task_type: str = "default",
@@ -45,6 +46,7 @@ def _make_q(
 # test_build_batches_deterministic
 # ---------------------------------------------------------------------------
 
+
 class TestBuildBatchesDeterministic:
     """相同输入 + 相同 seed 产出完全一致的切分。"""
 
@@ -73,6 +75,7 @@ class TestBuildBatchesDeterministic:
 # test_small_class_not_split
 # ---------------------------------------------------------------------------
 
+
 class TestSmallClassNotSplit:
     """小类（≤ min_class_per_batch）整组不拆，锁在同一 batch。"""
 
@@ -90,10 +93,7 @@ class TestSmallClassNotSplit:
         )
         assert count == 10
         # 找到包含 small_type 的 batch
-        small_batch = [
+        small_batch = [b for b in batches if any(q.task_type == "small_type" for q in b)]
-            b for b in batches
-            if any(q.task_type == "small_type" for q in b)
-        ]
         assert len(small_batch) == 1  # 整组在同一个 batch
         small_ids = {q.question_id for q in small_batch[0] if q.task_type == "small_type"}
         assert small_ids == {"s1", "s2"}
@@ -103,6 +103,7 @@ class TestSmallClassNotSplit:
 # test_large_class_round_robin
 # ---------------------------------------------------------------------------
 
+
 class TestLargeClassRoundRobin:
     """大类样本 round-robin 散布到多个 batch，不集中于单一 batch。"""
 
@@ -124,6 +125,7 @@ class TestLargeClassRoundRobin:
 # test_correct_ratio_mixing
 # ---------------------------------------------------------------------------
 
+
 class TestCorrectRatioMixing:
     """correct_ratio > 0 时混入正确题。"""
 
@@ -137,7 +139,11 @@ class TestCorrectRatioMixing:
         ]
         correctness = {"e1": False, "e2": False, "c1": True, "c2": True, "c3": True}
         batches, count = build_batches(
-            items, correctness, batch_size=10, min_class_per_batch=2, seed=0,
+            items,
+            correctness,
+            batch_size=10,
+            min_class_per_batch=2,
+            seed=0,
             correct_ratio=0.5,
         )
         # correct_ratio=0.5 → 错:正 = 1:1 → 2 错 + 2 正 = 4 题
@@ -155,7 +161,11 @@ class TestCorrectRatioMixing:
         ]
         correctness = {"e1": False, "c1": True}
         batches, count = build_batches(
-            items, correctness, batch_size=10, min_class_per_batch=2, seed=0,
+            items,
+            correctness,
+            batch_size=10,
+            min_class_per_batch=2,
+            seed=0,
             correct_ratio=0.0,
         )
         assert count == 1
@@ -166,6 +176,7 @@ class TestCorrectRatioMixing:
 # test_no_wrong_answers_empty
 # ---------------------------------------------------------------------------
 
+
 class TestNoWrongAnswersEmpty:
     """无错题时返回空列表。"""
 
@@ -173,14 +184,22 @@ class TestNoWrongAnswersEmpty:
         items = [_make_q(f"q{i}") for i in range(5)]
         correctness = {f"q{i}": True for i in range(5)}
         batches, count = build_batches(
-            items, correctness, batch_size=3, min_class_per_batch=1, seed=0,
+            items,
+            correctness,
+            batch_size=3,
+            min_class_per_batch=1,
+            seed=0,
         )
         assert batches == []
         assert count == 0
 
     def test_empty_items_returns_empty(self) -> None:
         batches, count = build_batches(
-            [], {}, batch_size=3, min_class_per_batch=1, seed=0,
+            [],
+            {},
+            batch_size=3,
+            min_class_per_batch=1,
+            seed=0,
         )
         assert batches == []
         assert count == 0
@@ -190,6 +209,7 @@ class TestNoWrongAnswersEmpty:
 # test_validate_params_strict
 # ---------------------------------------------------------------------------
 
+
 class TestValidateParamsStrict:
     """参数校验：batch_size < 1、min_class < 1、min_class >= batch_size 都报错。"""
 
@@ -221,6 +241,7 @@ class TestValidateParamsStrict:
 # test_correctness_false_vs_none
 # ---------------------------------------------------------------------------
 
+
 class TestCorrectnessFalseVsNone:
     """correctness.get(qid) is False 精确匹配：None（未知题）不算错题。"""
 
@@ -233,7 +254,11 @@ class TestCorrectnessFalseVsNone:
         # wrong=False（错题），right=True（正确题），unknown 不在 correctness（None）
         correctness: dict[str, bool] = {"wrong": False, "right": True}
         batches, count = build_batches(
-            items, correctness, batch_size=10, min_class_per_batch=2, seed=0,
+            items,
+            correctness,
+            batch_size=10,
+            min_class_per_batch=2,
+            seed=0,
             correct_ratio=0.0,
         )
         # 仅 wrong 进入 batch，unknown 不算错题
@@ -241,7 +266,7 @@ class TestCorrectnessFalseVsNone:
         assert batches[0][0].question_id == "wrong"
 
     def test_explicit_false_only(self) -> None:
-        """直接测试 _select_mixed_by_task_type 内部逻辑。"""
+        """直接测试 _select_mixed_by_task_type 内部逻辑（single 单元粒度）。"""
         items = [
             _make_q("f1", task_type="t1"),
             _make_q("n1", task_type="t1"),  # None（未知）
@@ -249,19 +274,19 @@ class TestCorrectnessFalseVsNone:
         ]
         correctness: dict[str, bool] = {"f1": False, "t1": True}
         rng = random.Random(0)
-        result = _select_mixed_by_task_type(items, correctness, 0.0, rng)
+        result = _select_mixed_by_task_type(build_units(items), correctness, 0.0, rng)
         assert "t1" in result
         assert len(result["t1"]) == 1
-        assert result["t1"][0].question_id == "f1"
+        assert result["t1"][0].unit_id == "f1"
 
     def test_none_not_treated_as_correct(self) -> None:
-        """None（未知）不进正确组，不被 correct_ratio 采样。"""
+        """None（未知）不进正确组，不被 correct_ratio 采样（single 单元粒度）。"""
         items = [
             _make_q("err", task_type="t1"),
             _make_q("unk", task_type="t1"),
         ]
         correctness: dict[str, bool] = {"err": False}
         rng = random.Random(0)
-        result = _select_mixed_by_task_type(items, correctness, 0.5, rng)
+        result = _select_mixed_by_task_type(build_units(items), correctness, 0.5, rng)
         # 只有 err 一题错题，unk 不在 correctness 中 → get 返回 None → 不进 correct 组
         assert len(result["t1"]) == 1  # 只有错题，无正确题可混入

tests/unit/test_harness_runner.py

@@ -53,7 +53,7 @@ class _FakeInferenceResult:
 
 @dataclass(frozen=True)
 class _FakeQuestion:
-    """GeneratedQuestion 替身。"""
+    """GeneratedQuestion 替身（含 pair 契约字段，供 build_units 聚合）。"""
 
     question_id: str
     video_id: str = "v1"
@@ -63,6 +63,15 @@ class _FakeQuestion:
     answer: str = "A"
     source_nodes: tuple = ()
     difficulty: str = "medium"
+    pair_id: str | None = None
+    question_role: str = "single"
+    unit_id: str = ""
+    flip_axis: str | None = None
+
+    def __post_init__(self) -> None:
+        """缺省 unit_id 回填为 pair_id 或 question_id，对齐真实 GeneratedQuestion。"""
+        if not self.unit_id:
+            object.__setattr__(self, "unit_id", self.pair_id or self.question_id)
 
 
 @dataclass