feat(question_gen): loader 按 unit 分层采样 + load_benchmark 读回 pair 字段

stratified_sample 先 build_units 聚合，以 QuestionUnit 为采样原子做
分层/去重/补足/rng.sample，返回前 flatten_units 展开为逐题列表；
size/correct_ratio/min_per_class 均按 unit 计数，单元正确性走成员 AND，
孪生对两题永不被劈开。纯 single 输入下 build_units 1:1 折叠、顺序不变，
rng 消耗与旧逐题实现字节级一致（新增回归测试守护）。

_backfill_per_class candidates 改按 unit 枚举去重；build_units/flatten_units
函数内延迟导入以规避 question_gen<->harness 循环依赖（沿用 adversarial_filter）。

load_benchmark 反序列化补 pair_id/question_role/flip_axis/unit_id 四字段，
用 .get 兼容旧 JSON（缺失退化为 single，unit_id 由 __post_init__ 回填）。

pools._sample_excluding 随之改为透传 flatten_units(candidates) 给已单元化的
stratified_sample（不再用 lone pair-original 代表），行为对 single-only 保持等价。

app/harness/pools.py

@@ -214,16 +214,16 @@ def _sample_excluding(
 ) -> list[GeneratedQuestion]:
     """排除已选 unit 后，以 unit 为原子按 cfg 分层采样，返回展开后的逐题列表。
 
-    每个单元以其首题作为分层采样的代表参与 stratified_sample，correct_ratio /
-    size 因此按 unit 计数（pair 计 1 个 unit）；命中的单元整体展开，孪生对两题
-    永远同进同出。single-only 输入下 unit 与 question 一一对应、顺序不变，采样
-    结果与逐题采样完全一致。
+    候选单元展开为逐题列表后透传给 stratified_sample，后者内部重新 build_units
+    做单元原子采样：correct_ratio / size 按 unit 计数（pair 计 1 个 unit），单元级
+    正确性由 stratified_sample 内部对成员取 AND，命中的孪生对两题永远同进同出。
+    single-only 输入下 unit 与 question 一一对应、顺序不变，采样结果与逐题采样一致。
 
     参数:
         units: 单元全集（single 单封、pair 成对聚合）。
         exclude_unit_ids: 已被其他池选走的 unit_id，从候选中剔除以保证三池互斥。
-        correctness: question_id -> 基线是否答对；单元级正确性取成员的 AND
-            （缺失按 False，与 stratified_sample 的宽松口径一致）。
+        correctness: question_id -> 基线是否答对；单元级正确性由 stratified_sample
+            对成员取 AND（缺失按 False，宽松口径）。
         cfg: 透传给 stratified_sample 的采样配置
             （size/correct_ratio/task_types[/seed/min_per_class]）。
 
@@ -231,12 +231,7 @@ def _sample_excluding(
         采样命中单元展开后的题目列表。
     """
     candidates = [u for u in units if u.unit_id not in exclude_unit_ids]
-    rep_to_unit = {u.questions[0].question_id: u for u in candidates}
-    reps = [u.questions[0] for u in candidates]
-    unit_correct = {u.questions[0].question_id: _unit_correct(u, correctness) for u in candidates}
-    sampled_reps = stratified_sample(reps, unit_correct, **cfg)
-    sampled_units = [rep_to_unit[rep.question_id] for rep in sampled_reps]
-    return flatten_units(sampled_units)
+    return stratified_sample(flatten_units(candidates), correctness, **cfg)
 
 
 def _q_to_dict(q: GeneratedQuestion) -> dict:

app/question_gen/loader.py

@@ -15,6 +15,8 @@ from core.types import GeneratedQuestion
 if TYPE_CHECKING:
     from pathlib import Path
 
+    from core.types import QuestionUnit
+
 _LEGACY_DEFAULT_DIFFICULTY = "medium"
 
 
@@ -26,6 +28,11 @@ def load_benchmark(questions_dir: Path) -> list[GeneratedQuestion]:
     video_id），v2 生成题把多视频题目合并在单个 JSON 中（每条记录自带
     ``video_id``），两种格式均兼容。
 
+    pair 契约字段（``pair_id`` / ``question_role`` / ``flip_axis`` / ``unit_id``）
+    用 ``.get`` 读取：旧 benchmark 无这些键时退化为 single（``question_role``
+    默认 "single"，``unit_id`` 留空由 __post_init__ 回填为 question_id），
+    保证历史题库可无缝加载。
+
     参数:
         questions_dir: 包含 *.json 文件的目录路径。
 
@@ -52,6 +59,12 @@ def load_benchmark(questions_dir: Path) -> list[GeneratedQuestion]:
                     skill_target=qa.get("skill_target"),
                     difficulty_steps=qa.get("difficulty_steps"),
                     sub_pattern=qa.get("sub_pattern"),
+                    # pair 契约字段：旧 benchmark 无这些键时按 single 默认兜底，
+                    # unit_id 留空交由 GeneratedQuestion.__post_init__ 回填。
+                    pair_id=qa.get("pair_id"),
+                    question_role=qa.get("question_role", "single"),
+                    flip_axis=qa.get("flip_axis"),
+                    unit_id=qa.get("unit_id", ""),
                 )
             )
     return results
@@ -66,62 +79,88 @@ def stratified_sample(
     seed: int,
     min_per_class: int | None,
 ) -> list[GeneratedQuestion]:
-    """按题型过滤后采样 size 道题，可选按对错比例分层并按题型保底。
+    """按题型过滤后采样 size 个单元，可选按对错比例分层并按题型保底。
 
     参数:
-        questions: 候选题目全集。
-        correctness: question_id -> 基线是否答对。
-        size: 采样总量。
-        correct_ratio: 采样中"基线答对"题的占比；None 表示自然分布。
+        questions: 候选题目全集（single 与孪生对成员可混含）。
+        correctness: question_id -> 基线是否答对（单元级正确性取成员 AND）。
+        size: 采样单元总量（single 计 1、pair 计 1）。
+        correct_ratio: 采样中"基线答对"单元的占比；None 表示自然分布。
         task_types: 限定题型；None 表示不限。
         seed: 随机种子，保证可复现。
-        min_per_class: 每个题型补足到的下限；None 表示不补足。
+        min_per_class: 每个题型补足到的单元下限；None 表示不补足。
 
     返回:
-        采样后的题目列表。
+        采样后的题目列表（pair 单元展开为原始的两道题）。
 
     异常:
-        ValueError: 自然分布时池不足 size，或分层时某层题目不足。
+        ValueError: 自然分布时单元池不足 size，或分层时某层单元不足。
+
+    关键实现:
+        以 **QuestionUnit 为采样原子**（single 计 1、pair 计 1），size /
+        correct_ratio / min_per_class 均按 unit 计数，孪生对两题永不被劈开。
+        采样完成后 flatten_units 展开回逐题列表。纯 single 输入时 build_units
+        与题目一一对应、顺序不变，rng 消耗与旧逐题实现完全一致（字节级回归）。
+
+        build_units / flatten_units 采用函数内延迟导入：loader 属 question_gen，
+        question_units 属 harness，而 harness 包初始化会反向 import question_gen，
+        模块级导入将触发循环依赖（沿用 adversarial_filter 的既有做法）。
     """
+    from app.harness.question_units import build_units, flatten_units
+
     rng = random.Random(seed)
-    pool = [q for q in questions if task_types is None or q.task_type in task_types]
+    units = build_units(questions)
+    pool = [u for u in units if task_types is None or u.task_type in task_types]
 
     if correct_ratio is None:
         if len(pool) < size:
-            raise ValueError(f"自然分布采样不足: 需 {size} 道, 实有 {len(pool)} 道")
+            raise ValueError(f"自然分布采样不足: 需 {size} 个单元, 实有 {len(pool)} 个")
         sampled = rng.sample(pool, size)
     else:
         sampled = _ratio_stratified_sample(pool, correctness, size, correct_ratio, rng)
 
     if min_per_class is not None:
         sampled = _backfill_per_class(sampled, pool, min_per_class, rng)
-    return sampled
+    return flatten_units(sampled)
+
+
+def _unit_correct(unit: QuestionUnit, correctness: dict[str, bool]) -> bool:
+    """单元级正确性：成员全部答对才算对（缺失按 False，宽松口径）。
+
+    参数:
+        unit: 目标单元（single 1 题，pair 2 题）。
+        correctness: question_id -> 基线是否答对。
+
+    返回:
+        pair 走双向 AND、single 即单题正确性；任一成员缺失或答错即 False。
+    """
+    return all(correctness.get(q.question_id, False) for q in unit.questions)
 
 
 def _ratio_stratified_sample(
-    pool: list[GeneratedQuestion],
+    pool: list[QuestionUnit],
     correctness: dict[str, bool],
     size: int,
     correct_ratio: float,
     rng: random.Random,
-) -> list[GeneratedQuestion]:
-    """按对错比例分层采样：对题占 correct_ratio，其余为错题。
+) -> list[QuestionUnit]:
+    """按对错比例分层采样：对单元占 correct_ratio，其余为错单元。
 
     参数:
-        pool: 题型过滤后的候选题。
+        pool: 题型过滤后的候选单元。
         correctness: question_id -> 基线是否答对。
-        size: 采样总量。
-        correct_ratio: 对题占比。
+        size: 采样单元总量。
+        correct_ratio: 对单元占比。
         rng: 随机数发生器。
 
     返回:
-        采样后的题目列表（对题在前、错题在后）。
+        采样后的单元列表（对单元在前、错单元在后）。
 
     异常:
-        ValueError: 对题或错题层不足。
+        ValueError: 对单元或错单元层不足。
     """
-    correct = [q for q in pool if correctness.get(q.question_id, False)]
-    wrong = [q for q in pool if not correctness.get(q.question_id, False)]
+    correct = [u for u in pool if _unit_correct(u, correctness)]
+    wrong = [u for u in pool if not _unit_correct(u, correctness)]
     n_correct = round(size * correct_ratio)
     n_wrong = size - n_correct
     if len(correct) < n_correct or len(wrong) < n_wrong:
@@ -132,42 +171,40 @@ def _ratio_stratified_sample(
 
 
 def _backfill_per_class(
-    sampled: list[GeneratedQuestion],
-    pool: list[GeneratedQuestion],
+    sampled: list[QuestionUnit],
+    pool: list[QuestionUnit],
     min_per_class: int,
     rng: random.Random,
-) -> list[GeneratedQuestion]:
-    """对候选池中出现的每个题型，将采样结果补足到 min_per_class 道。
+) -> list[QuestionUnit]:
+    """对候选池中出现的每个题型，将采样单元补足到 min_per_class 个。
 
     遍历对象是候选池 pool 里出现的全部题型（非仅 sampled 命中的），
-    保证任意稀疏题型都能拿到足额样本。
+    保证任意稀疏题型都能拿到足额样本。补足以 unit 为原子，孪生对整进整出。
 
     参数:
-        sampled: 主采样结果（不修改，返回新列表）。
-        pool: 候选题全集（补足来源 + 题型枚举来源）。
-        min_per_class: 每个题型的下限。
+        sampled: 主采样结果单元（不修改，返回新列表）。
+        pool: 候选单元全集（补足来源 + 题型枚举来源）。
+        min_per_class: 每个题型的单元下限。
         rng: 随机数发生器。
 
     返回:
-        补足后的题目列表。
+        补足后的单元列表。
     """
-    selected_ids = {q.question_id for q in sampled}
+    selected_ids = {u.unit_id for u in sampled}
     result = list(sampled)
     counts: dict[str, int] = {}
-    for q in sampled:
-        counts[q.task_type] = counts.get(q.task_type, 0) + 1
+    for u in sampled:
+        counts[u.task_type] = counts.get(u.task_type, 0) + 1
     ordered_task_types: dict[str, None] = {}
-    for q in pool:
-        ordered_task_types.setdefault(q.task_type, None)
+    for u in pool:
+        ordered_task_types.setdefault(u.task_type, None)
     for task_type in ordered_task_types:
         deficit = min_per_class - counts.get(task_type, 0)
         if deficit <= 0:
             continue
-        candidates = [
-            q for q in pool if q.task_type == task_type and q.question_id not in selected_ids
-        ]
+        candidates = [u for u in pool if u.task_type == task_type and u.unit_id not in selected_ids]
         take = rng.sample(candidates, min(deficit, len(candidates)))
-        for q in take:
-            selected_ids.add(q.question_id)
-            result.append(q)
+        for u in take:
+            selected_ids.add(u.unit_id)
+            result.append(u)
     return result

tests/unit/test_loader_unit_sampling.py

@@ -0,0 +1,297 @@
+"""loader.stratified_sample 按 unit 采样 + load_benchmark 读回 pair 字段的单元测试。
+
+覆盖 question-gen v3 Phase 1 Task 4 的四项契约：
+    (a) pair 采样同进同出——同一 pair_id 两题要么都被选、要么都不被选；
+    (b) size / correct_ratio 按 **unit 计数**（pair 计 1 个 unit，unit 正确性走双向 AND）；
+    (c) min_per_class 补足路径不拆 pair；
+    (d) load_benchmark 反序列化读回 pair_id/question_role/flip_axis/unit_id，
+        且旧 JSON 缺这些字段时按 single 默认兜底、不崩。
+另加一条纯 single 输入的字节级回归守护：单题场景采样序列必须与旧逐题行为一致。
+"""
+
+from __future__ import annotations
+
+import json
+import random
+from pathlib import Path
+
+import pytest
+
+from app.harness.question_units import build_units
+from app.question_gen.loader import load_benchmark, stratified_sample
+from core.types import GeneratedQuestion
+
+
+def _single(qid: str, task_type: str = "Single") -> GeneratedQuestion:
+    """构造一道 single 题（无 pair 归属）。"""
+    return GeneratedQuestion(
+        question_id=qid,
+        video_id="v",
+        task_type=task_type,
+        question="?",
+        options=("A", "B", "C", "D"),
+        answer="A",
+        source_nodes=(),
+        difficulty="medium",
+    )
+
+
+def _pair(pid: str, task_type: str = "AR") -> list[GeneratedQuestion]:
+    """构造一个合法孪生对（original + mirror），共享 pair_id / unit_id / flip_axis。"""
+    base = dict(
+        video_id="v",
+        task_type=task_type,
+        question="?",
+        options=("A", "B", "C", "D"),
+        answer="A",
+        source_nodes=(),
+        difficulty="hard",
+        pair_id=pid,
+        unit_id=pid,
+        flip_axis="before_after",
+    )
+    return [
+        GeneratedQuestion(question_id=f"{pid}_o", question_role="pair_original", **base),
+        GeneratedQuestion(question_id=f"{pid}_m", question_role="pair_mirror", **base),
+    ]
+
+
+def _pair_roles_by_id(questions: list[GeneratedQuestion]) -> dict[str, set[str]]:
+    """将采样结果按 pair_id 聚合出现的角色集合（仅统计配对题）。"""
+    loc: dict[str, set[str]] = {}
+    for q in questions:
+        if q.pair_id:
+            loc.setdefault(q.pair_id, set()).add(q.question_role)
+    return loc
+
+
+class TestPairAtomicSampling:
+    def test_pair_never_split_in_natural_sample(self) -> None:
+        """自然分布采样：命中的 pair 必两题齐全，size 按 unit 计数。"""
+        qs = [q for i in range(8) for q in _pair(f"p{i}")]
+        result = stratified_sample(
+            questions=qs,
+            correctness={},
+            size=4,
+            correct_ratio=None,
+            task_types=None,
+            seed=7,
+            min_per_class=None,
+        )
+        loc = _pair_roles_by_id(result)
+        assert len(loc) == 4, "size=4 应命中 4 个 pair 单元"
+        for pid, roles in loc.items():
+            assert roles == {"pair_original", "pair_mirror"}, f"pair {pid} 被拆: {roles}"
+        assert len(result) == 8, "4 个 pair 单元展开应为 8 道题"
+
+    def test_size_counts_units_not_questions(self) -> None:
+        """混合 single + pair 时 size 仍按 unit 计数（pair 计 1）。"""
+        qs = [_single(f"s{i}") for i in range(3)]
+        qs += [q for i in range(3) for q in _pair(f"p{i}")]
+        result = stratified_sample(
+            questions=qs,
+            correctness={},
+            size=6,  # 全部 6 个单元（3 single + 3 pair）
+            correct_ratio=None,
+            task_types=None,
+            seed=1,
+            min_per_class=None,
+        )
+        singles = [q for q in result if not q.pair_id]
+        loc = _pair_roles_by_id(result)
+        assert len(singles) == 3
+        assert len(loc) == 3
+        for roles in loc.values():
+            assert roles == {"pair_original", "pair_mirror"}
+        assert len(result) == 3 + 3 * 2
+
+    def test_ratio_counts_units(self) -> None:
+        """correct_ratio 按 unit 计数：对/错单元按比例各取整数个。"""
+        correct_pairs = [_pair(f"c{i}") for i in range(4)]
+        wrong_pairs = [_pair(f"w{i}") for i in range(4)]
+        qs = [q for p in correct_pairs + wrong_pairs for q in p]
+        correctness: dict[str, bool] = {}
+        for p in correct_pairs:
+            for q in p:
+                correctness[q.question_id] = True
+        result = stratified_sample(
+            questions=qs,
+            correctness=correctness,
+            size=4,
+            correct_ratio=0.5,
+            task_types=None,
+            seed=3,
+            min_per_class=None,
+        )
+        loc = _pair_roles_by_id(result)
+        assert len(loc) == 4, "4 个 unit"
+        for roles in loc.values():
+            assert roles == {"pair_original", "pair_mirror"}
+        correct_units = sum(
+            1 for pid in loc if all(correctness.get(f"{pid}_{s}", False) for s in ("o", "m"))
+        )
+        assert correct_units == 2, "correct_ratio=0.5 * 4 unit = 2 个对单元"
+
+    def test_unit_correctness_uses_and_not_any(self) -> None:
+        """unit 正确性走双向 AND：混合对（P 对 Q 错）不得计入对单元层。
+
+        构造 2 个全对 pair + 1 个混合 pair（original 对、mirror 错）。请求 3 个对单元，
+        若按 AND，对池仅 2 个 → 分层不足报错；若错误地按 any-member，对池 3 个 → 不报错。
+        以是否抛错来判别语义，规避随机命中导致的假阳性。
+        """
+        good = [_pair("g0"), _pair("g1")]
+        mixed = _pair("mx")
+        qs = [q for p in good for q in p] + mixed
+        correctness: dict[str, bool] = {}
+        for p in good:
+            for q in p:
+                correctness[q.question_id] = True
+        correctness[mixed[0].question_id] = True
+        correctness[mixed[1].question_id] = False
+        with pytest.raises(ValueError, match="分层不足"):
+            stratified_sample(
+                questions=qs,
+                correctness=correctness,
+                size=3,
+                correct_ratio=1.0,
+                task_types=None,
+                seed=1,
+                min_per_class=None,
+            )
+
+
+class TestBackfillPairAtomic:
+    def test_backfill_keeps_pairs_atomic(self) -> None:
+        """min_per_class 补足稀疏 pair 题型时，补入的 pair 两题齐全不拆。"""
+        ar = [q for i in range(3) for q in _pair(f"a{i}", task_type="AR")]
+        main = [_single(f"m{i}", task_type="Main") for i in range(10)]
+        result = stratified_sample(
+            questions=ar + main,
+            correctness={},
+            size=2,
+            correct_ratio=None,
+            task_types=None,
+            seed=5,
+            min_per_class=2,
+        )
+        loc = _pair_roles_by_id(result)
+        assert len(loc) >= 2, "AR 至少被补足到 2 个 pair 单元"
+        for pid, roles in loc.items():
+            assert roles == {"pair_original", "pair_mirror"}, f"补足拆散了 pair {pid}: {roles}"
+
+    def test_backfill_no_duplicate_units(self) -> None:
+        """补足不得重复选入同一 pair 单元。"""
+        ar = [q for i in range(4) for q in _pair(f"a{i}", task_type="AR")]
+        result = stratified_sample(
+            questions=ar,
+            correctness={},
+            size=1,
+            correct_ratio=None,
+            task_types=None,
+            seed=9,
+            min_per_class=3,
+        )
+        ids = [q.question_id for q in result]
+        assert len(ids) == len(set(ids)), "存在重复题目"
+
+
+class TestLoadBenchmarkPairFields:
+    def test_reads_pair_fields(self, tmp_path: Path) -> None:
+        """新格式 JSON 的 pair_id/question_role/flip_axis/unit_id 被正确读回。"""
+        data = [
+            {
+                "question_id": "pp_o",
+                "video_id": "v",
+                "task_type": "AR",
+                "question": "?",
+                "options": ["A", "B", "C", "D"],
+                "answer": "A",
+                "pair_id": "pp",
+                "question_role": "pair_original",
+                "flip_axis": "before_after",
+                "unit_id": "pp",
+            },
+            {
+                "question_id": "pp_m",
+                "video_id": "v",
+                "task_type": "AR",
+                "question": "?",
+                "options": ["A", "B", "C", "D"],
+                "answer": "A",
+                "pair_id": "pp",
+                "question_role": "pair_mirror",
+                "flip_axis": "before_after",
+                "unit_id": "pp",
+            },
+        ]
+        (tmp_path / "vid.json").write_text(json.dumps(data), encoding="utf-8")
+        qs = load_benchmark(tmp_path)
+        q0 = next(q for q in qs if q.question_id == "pp_o")
+        assert q0.pair_id == "pp"
+        assert q0.question_role == "pair_original"
+        assert q0.flip_axis == "before_after"
+        assert q0.unit_id == "pp"
+        # 读回后应能被 build_units 重新聚合成 1 个 pair 单元
+        units = build_units(qs)
+        assert len(units) == 1
+        assert units[0].kind == "pair"
+
+    def test_legacy_json_defaults_to_single(self, tmp_path: Path) -> None:
+        """旧 JSON 缺 pair 字段时按 single 兜底、unit_id 回填为 question_id，不崩。"""
+        data = [
+            {
+                "question_id": "x1",
+                "video_id": "v",
+                "task_type": "T",
+                "question": "?",
+                "options": ["A", "B", "C", "D"],
+                "answer": "A",
+            }
+        ]
+        (tmp_path / "legacy.json").write_text(json.dumps(data), encoding="utf-8")
+        qs = load_benchmark(tmp_path)
+        assert qs[0].pair_id is None
+        assert qs[0].question_role == "single"
+        assert qs[0].flip_axis is None
+        assert qs[0].unit_id == "x1"
+        units = build_units(qs)
+        assert len(units) == 1
+        assert units[0].kind == "single"
+
+
+class TestPureSingleByteIdentical:
+    def test_natural_matches_legacy_rng(self) -> None:
+        """纯 single 输入的自然分布采样，与旧逐题 rng.sample 序列字节级一致。"""
+        qs = [_single(f"s{i}") for i in range(20)]
+        seed = 123
+        result = stratified_sample(
+            questions=qs,
+            correctness={},
+            size=8,
+            correct_ratio=None,
+            task_types=None,
+            seed=seed,
+            min_per_class=None,
+        )
+        reference = random.Random(seed).sample(qs, 8)
+        assert [q.question_id for q in result] == [q.question_id for q in reference]
+
+    def test_ratio_matches_legacy_rng(self) -> None:
+        """纯 single 输入的比例分层采样，与旧逐题实现的抽样序列一致。"""
+        qs = [_single(f"s{i}") for i in range(20)]
+        correctness = {f"s{i}": i < 10 for i in range(20)}
+        seed = 77
+        result = stratified_sample(
+            questions=qs,
+            correctness=correctness,
+            size=10,
+            correct_ratio=0.6,
+            task_types=None,
+            seed=seed,
+            min_per_class=None,
+        )
+        rng = random.Random(seed)
+        correct = [q for q in qs if correctness.get(q.question_id, False)]
+        wrong = [q for q in qs if not correctness.get(q.question_id, False)]
+        reference = rng.sample(correct, 6) + rng.sample(wrong, 4)
+        assert [q.question_id for q in result] == [q.question_id for q in reference]