fix: fail-loud on disk-corrupt pairs and preserve buffer on reject

read_accepted now distinguishes size==1 dangling orphans (warn+drop) from
structural corruption / role duplication (raise), and runs explicit binding
checks (video_id/task_type/flip_axis) that survive python -O. add() validates
before evicting the buffered partner so pending_orphans can recover it.

app/question_gen/pair_atomic_writer.py

@@ -69,9 +69,10 @@ class PairPendingBuffer:
         成员到齐后返回 kind="pair" 单元。同 pair_id 两成员绑定不一致或角色非法（如
         两个 original）即 raise ValueError。
 
-        关键实现:配齐后先 ``_check_pair_binding`` 显式绑定校验（把不一致落成清晰
-        ValueError，不依赖会被 ``-O`` 剥除的 assert），再 ``QuestionUnit.from_pair``
-        组装。
+        关键实现:配齐后**先校验、成功组装出 unit 才从 pending 删除**——若
+        ``_check_pair_binding`` / ``_order_pair`` raise，首成员仍留在 pending，调用方
+        ``pending_orphans`` 可取回被拒的悬挂成员。绑定校验用显式 ValueError（不依赖
+        会被 ``-O`` 剥除的 assert）。
         """
         if not q.pair_id:
             return QuestionUnit.from_single(q)
@@ -81,10 +82,11 @@ class PairPendingBuffer:
             self._pending[q.pair_id] = q
             return None
 
-        del self._pending[q.pair_id]
         _check_pair_binding(partner, q)
         original, mirror = _order_pair(partner, q)
-        return QuestionUnit.from_pair(original, mirror)
+        unit = QuestionUnit.from_pair(original, mirror)
+        del self._pending[q.pair_id]
+        return unit
 
     def pending_orphans(self) -> list[GeneratedQuestion]:
         """返回当前仍未配齐的孤儿成员（供批次末尾检测悬挂项）。
@@ -113,37 +115,34 @@ def _order_pair(
     return originals[0], mirrors[0]
 
 
-def _is_intact_pair(pair_id: str, group: list[GeneratedQuestion]) -> bool:
-    """判定同 pair_id 分组 group 是否为完整合法孪生对（恰好 1 original + 1 mirror）。
+def _validate_disk_pair(pair_id: str, group: list[GeneratedQuestion]) -> list[GeneratedQuestion]:
+    """三态判定磁盘上同 pair_id 分组，区分悬挂孤儿（drop）与结构损坏（raise）。
 
-    完整为 True；否则 warn（含成员构成）并返回 False，由调用方剔除该悬挂 unit。
+    - 恰好 1 original + 1 mirror：合法孪生对，额外做 ``_check_pair_binding`` 显式绑定
+      校验（video_id/task_type/flip_axis，-O 下仍生效），返回其两题。
+    - size==1（只落 P 未落 Q）：业务上合法的悬挂孤儿，warn + drop，返回 []。
+    - 其余（size>2 超员、或 size==2 角色重复/缺角色）：数据损坏/外部篡改，按 P5
+      fail-loud，raise ValueError（含 pair_id 与成员构成），绝不静默吞。
     """
     originals = sum(1 for q in group if q.question_role == "pair_original")
     mirrors = sum(1 for q in group if q.question_role == "pair_mirror")
     if len(group) == 2 and originals == 1 and mirrors == 1:
-        return True
-    logger.warning(
-        "磁盘悬挂孤儿 pair {}：成员数={}（original={} mirror={}），剔除该 unit",
-        pair_id,
-        len(group),
-        originals,
-        mirrors,
+        _check_pair_binding(group[0], group[1])
+        return group
+    if len(group) == 1:
+        logger.warning("磁盘悬挂孤儿 pair {}：仅 1 成员（缺伙伴），warn+drop 该 unit", pair_id)
+        return []
+    raise ValueError(
+        f"磁盘 pair {pair_id} 结构损坏：成员数={len(group)}"
+        f"（original={originals} mirror={mirrors}），需恰好 1 original + 1 mirror"
     )
-    return False
 
 
-def _drop_dangling_pairs(questions: list[GeneratedQuestion]) -> list[GeneratedQuestion]:
-    """剔除磁盘上收不齐 2 条 / 角色非法的悬挂孤儿 pair，告警不静默。
+def _sift_disk_pairs(questions: list[GeneratedQuestion]) -> list[GeneratedQuestion]:
+    """筛选磁盘读回的题目：single 全保留、pair 按 ``_validate_disk_pair`` 三态处理。
 
-    与 T6 inference 的 ``_drop_orphan_pairs`` 语义一致：磁盘上"只落 P 未落 Q"是
-    业务上合法的悬挂项（非结构损坏），warn+drop 而非 raise，使后续 build_units 只
-    面对合法孪生对。single 全保留。
-
-    参数:
-        questions: 从磁盘读回的题目列表（可混含 single 与孪生对成员）。
-
-    返回:
-        可安全交给 build_units 的题目列表（single 全保留，pair 仅保留合法成对者）。
+    悬挂孤儿 warn+drop、结构损坏/绑定不一致 raise ValueError、合法成对保留后交给
+    build_units（single 全保留）。
     """
     by_pair: dict[str, list[GeneratedQuestion]] = defaultdict(list)
     singles: list[GeneratedQuestion] = []
@@ -153,7 +152,7 @@ def _drop_dangling_pairs(questions: list[GeneratedQuestion]) -> list[GeneratedQu
         else:
             singles.append(q)
 
-    kept_pairs = [q for pid, grp in by_pair.items() if _is_intact_pair(pid, grp) for q in grp]
+    kept_pairs = [q for pid, grp in by_pair.items() for q in _validate_disk_pair(pid, grp)]
     return singles + kept_pairs
 
 
@@ -181,19 +180,21 @@ def write_accepted(path: Path, units: list[QuestionUnit]) -> None:
 
 
 def read_accepted(path: Path) -> list[QuestionUnit]:
-    """读回 path 的 accepted JSON → 聚合为单元列表，剔除磁盘悬挂孤儿。
+    """读回 path 的 accepted JSON → 聚合为单元列表。
 
-    返回校验通过的单元列表（single + 合法 pair）；磁盘上"只落 P 未落 Q"的悬挂孤儿
-    被 warn+drop，不进结果、不 raise。
+    磁盘是外部输入，按 P5 全量校验后再用：``_sift_disk_pairs`` 对 pair 分组三态处理
+    ——"只落 P 未落 Q"的悬挂孤儿 warn+drop（不进结果、不 raise）；结构损坏（超员/
+    角色重复）或绑定不一致（video_id/task_type/flip_axis）显式 raise ValueError（不
+    依赖 build_units 内会被 ``-O`` 剥除的 assert）。single 全保留。
 
     关键实现:反序列化复用 T9 pools.py 的 ``_dict_to_q``（pair 四字段 .get 兼容），
-    函数内 import 规避循环依赖。先 ``_drop_dangling_pairs`` 剔除悬挂孤儿，再
-    ``build_units`` 聚合、``validate_units`` 二次防御闸门。
+    函数内 import 规避循环依赖。sift 后 ``build_units`` 聚合、``validate_units`` 二次
+    防御闸门。
     """
     from app.harness.pools import _dict_to_q
     from app.harness.question_units import build_units, validate_units
 
     raw = json.loads(path.read_text(encoding="utf-8"))
     questions = [_dict_to_q(d) for d in raw]
-    kept = _drop_dangling_pairs(questions)
+    kept = _sift_disk_pairs(questions)
     return validate_units(build_units(kept))

tests/unit/test_pair_atomic_write.py

@@ -120,8 +120,8 @@ def test_dangling_original_stays_in_pending() -> None:
     assert orphans[0].question_id == original.question_id
 
 
-def test_inconsistent_pair_binding_rejected() -> None:
-    """同 pair_id 两成员 flip_axis 不一致 → 拒绝（unit_hash 语义代偿）。"""
+def test_inconsistent_pair_binding_rejected_and_partner_recoverable() -> None:
+    """两成员 flip_axis 不一致 → 拒绝；且校验失败后首成员仍留 pending 可取回。"""
     buf = PairPendingBuffer()
     original, _ = _make_pair(flip_axis="before_after")
     _, bad_mirror = _make_pair(flip_axis="left_right")
@@ -130,6 +130,11 @@ def test_inconsistent_pair_binding_rejected() -> None:
     with pytest.raises(ValueError, match="绑定不一致"):
         buf.add(bad_mirror)
 
+    # Important: 先校验后删除——raise 后首成员未被驱逐，调用方仍能检测到悬挂项
+    orphans = buf.pending_orphans()
+    assert len(orphans) == 1
+    assert orphans[0].question_id == original.question_id
+
 
 def test_same_role_pair_rejected() -> None:
     """同 pair_id 两成员角色相同（两个 original）→ fail-fast 拒绝。"""
@@ -190,6 +195,45 @@ def test_read_drops_disk_dangling_orphan(tmp_path, caplog) -> None:
     assert loaded[0].questions[0].question_id == single.question_id
 
 
+def test_read_raises_on_disk_corrupt_duplicate_role(tmp_path) -> None:
+    """磁盘上同 pair_id 出现两个 pair_original（角色重复）→ 数据损坏，fail-fast raise。"""
+    o1, _ = _make_pair()
+    o2, _ = _make_pair()
+    # 同 pair_id、两条都是 pair_original（size==2 但角色非法）→ 结构损坏
+    dup = GeneratedQuestion(
+        question_id="p1_o2",
+        video_id=o2.video_id,
+        task_type=o2.task_type,
+        question="重复 original",
+        options=o2.options,
+        answer=o2.answer,
+        source_nodes=o2.source_nodes,
+        difficulty=o2.difficulty,
+        pair_id=o1.pair_id,
+        question_role="pair_original",
+        flip_axis=o1.flip_axis,
+    )
+    records = [_q_to_dict(o1), _q_to_dict(dup)]
+    out = tmp_path / "corrupt.json"
+    out.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="结构损坏"):
+        read_accepted(out)
+
+
+def test_read_raises_on_disk_binding_mismatch(tmp_path) -> None:
+    """磁盘上合法成对但 flip_axis 绑定不一致 → 显式 raise（不依赖会被 -O 剥除的 assert）。"""
+    original, _ = _make_pair(pair_id="p1", flip_axis="before_after")
+    _, mirror = _make_pair(pair_id="p1", flip_axis="left_right")
+    # 直接写盘绕过 write_accepted 的 from_pair 组装，模拟外部篡改的不一致成对
+    records = [_q_to_dict(original), _q_to_dict(mirror)]
+    out = tmp_path / "mismatch.json"
+    out.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="绑定不一致"):
+        read_accepted(out)
+
+
 def test_write_accepted_rejects_malformed_unit(tmp_path) -> None:
     """validate_units 闸门：结构非法的 pair 单元（size≠2）落盘前 fail-fast。"""
     from core.types import QuestionUnit