From bdcc93d7de58a1ae466b0cdf17da5739ac94017c Mon Sep 17 00:00:00 2001
From: iomgaa <iomgaaycz@gmail.com>
Date: Wed, 15 Jul 2026 09:22:41 -0400
Subject: [PATCH] feat: add pair atomic accepted-bank writer helper

PairPendingBuffer collects pair members by pair_id and emits units;
write_accepted does tmp+os.replace atomic write; read_accepted rebuilds
units and drops disk-dangling orphans. Pure helper; on_accept wiring is
Phase 2 (real write point: adversarial_filter.write_final_bank).
---
 app/question_gen/pair_atomic_writer.py | 199 ++++++++++++++++++++++++
 tests/unit/test_pair_atomic_write.py   | 200 +++++++++++++++++++++++++
 2 files changed, 399 insertions(+)
 create mode 100644 app/question_gen/pair_atomic_writer.py
 create mode 100644 tests/unit/test_pair_atomic_write.py

diff --git a/app/question_gen/pair_atomic_writer.py b/app/question_gen/pair_atomic_writer.py
new file mode 100644
index 0000000..e5458b5
--- /dev/null
+++ b/app/question_gen/pair_atomic_writer.py
@@ -0,0 +1,199 @@
+"""accepted 题库的 pair 原子成对落盘 helper（纯件，不依赖 pipeline）。
+
+三件可复用纯件供 Phase 2 新 pipeline 的 on_accept 回调复用：``PairPendingBuffer``
+（按 pair_id 收齐才 emit 单元）、``write_accepted``（tmp + os.replace 原子写）、
+``read_accepted``（聚合成单元并剔除磁盘悬挂孤儿）。
+
+wiring 归属：本模块**只是纯件**，不接任何生成侧回调；真正的 on_accept wiring
+**见 Phase 2**。当前项目真实 accepted 写入点是 ``adversarial_filter.write_final_bank``
+（Phase 2 待重建的旧代码），本模块沿用其 tmp + os.replace 模式但不 import/不改动它。
+
+unit_hash 校验：``QuestionUnit.unit_hash`` 目前恒为 ""（填充是 Phase 2 的事），且
+``GeneratedQuestion`` 不携带 unit_hash 字段。故把"unit_hash 不一致→拒"落地为**同
+pair_id 两成员的绑定一致性校验**（video_id / task_type / flip_axis），不一致即
+fail-fast raise，语义等价——只有 payload 绑定一致的孪生对才允许聚合。
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from collections import defaultdict
+from typing import TYPE_CHECKING
+
+from loguru import logger
+
+from core.types import GeneratedQuestion, QuestionUnit
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _check_pair_binding(first: GeneratedQuestion, second: GeneratedQuestion) -> None:
+    """校验同 pair_id 两成员的绑定一致性，不一致 fail-fast（unit_hash 语义代偿）。
+
+    参数 first/second 为先后到达的孪生对成员。video_id / task_type / flip_axis 任一
+    不一致即 raise ValueError——绑定不一致的两条题目不构成同一 payload 的孪生对，
+    拒绝聚合而非静默兜底。
+    """
+    mismatches: list[str] = []
+    if first.video_id != second.video_id:
+        mismatches.append(f"video_id: {first.video_id} != {second.video_id}")
+    if first.task_type != second.task_type:
+        mismatches.append(f"task_type: {first.task_type} != {second.task_type}")
+    if first.flip_axis != second.flip_axis:
+        mismatches.append(f"flip_axis: {first.flip_axis} != {second.flip_axis}")
+    if mismatches:
+        raise ValueError(
+            f"pair {first.pair_id} 两成员绑定不一致（" + "；".join(mismatches) + "），拒绝聚合"
+        )
+
+
+class PairPendingBuffer:
+    """按 pair_id 收齐孪生对才 emit 单元的有状态缓冲器。
+
+    喂题接口 ``add`` 逐条消费题目：single 立即 emit ``QuestionUnit.from_single``；
+    pair 成员先缓存，等同一 pair_id 的 original+mirror 都到齐才 emit 一个 pair 单元
+    （复用 ``QuestionUnit.from_pair`` 走 fail-fast 校验）。批次末尾用
+    ``pending_orphans`` 检测"只落 P 未落 Q"的悬挂项。
+    """
+
+    def __init__(self) -> None:
+        """初始化空缓冲器。pending 以 pair_id 索引首个到达的孪生对成员。"""
+        self._pending: dict[str, GeneratedQuestion] = {}
+
+    def add(self, q: GeneratedQuestion) -> QuestionUnit | None:
+        """喂入一条题目 q，返回本次凑齐的单元或 None（pair 尚未配齐）。
+
+        single 立即返回 kind="single" 单元；pair 首个成员缓存并返回 None，第二个
+        成员到齐后返回 kind="pair" 单元。同 pair_id 两成员绑定不一致或角色非法（如
+        两个 original）即 raise ValueError。
+
+        关键实现:配齐后先 ``_check_pair_binding`` 显式绑定校验（把不一致落成清晰
+        ValueError，不依赖会被 ``-O`` 剥除的 assert），再 ``QuestionUnit.from_pair``
+        组装。
+        """
+        if not q.pair_id:
+            return QuestionUnit.from_single(q)
+
+        partner = self._pending.get(q.pair_id)
+        if partner is None:
+            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)
+
+    def pending_orphans(self) -> list[GeneratedQuestion]:
+        """返回当前仍未配齐的孤儿成员（供批次末尾检测悬挂项）。
+
+        返回:
+            尚在缓冲、未等到伙伴的孪生对成员列表（按 pair_id 插入顺序）。
+        """
+        return list(self._pending.values())
+
+
+def _order_pair(
+    a: GeneratedQuestion, b: GeneratedQuestion
+) -> tuple[GeneratedQuestion, GeneratedQuestion]:
+    """按 question_role 把两成员 a/b 定序为 (original, mirror)。
+
+    两成员不构成恰好 1 original + 1 mirror（角色缺失或重复）即 raise ValueError，
+    防 next(...) 静默 StopIteration。
+    """
+    originals = [q for q in (a, b) if q.question_role == "pair_original"]
+    mirrors = [q for q in (a, b) if q.question_role == "pair_mirror"]
+    if len(originals) != 1 or len(mirrors) != 1:
+        raise ValueError(
+            f"pair {a.pair_id} 角色非法：original={len(originals)} mirror={len(mirrors)}，"
+            "需各恰好 1 条"
+        )
+    return originals[0], mirrors[0]
+
+
+def _is_intact_pair(pair_id: str, group: list[GeneratedQuestion]) -> bool:
+    """判定同 pair_id 分组 group 是否为完整合法孪生对（恰好 1 original + 1 mirror）。
+
+    完整为 True；否则 warn（含成员构成）并返回 False，由调用方剔除该悬挂 unit。
+    """
+    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,
+    )
+    return False
+
+
+def _drop_dangling_pairs(questions: list[GeneratedQuestion]) -> list[GeneratedQuestion]:
+    """剔除磁盘上收不齐 2 条 / 角色非法的悬挂孤儿 pair，告警不静默。
+
+    与 T6 inference 的 ``_drop_orphan_pairs`` 语义一致：磁盘上"只落 P 未落 Q"是
+    业务上合法的悬挂项（非结构损坏），warn+drop 而非 raise，使后续 build_units 只
+    面对合法孪生对。single 全保留。
+
+    参数:
+        questions: 从磁盘读回的题目列表（可混含 single 与孪生对成员）。
+
+    返回:
+        可安全交给 build_units 的题目列表（single 全保留，pair 仅保留合法成对者）。
+    """
+    by_pair: dict[str, list[GeneratedQuestion]] = defaultdict(list)
+    singles: list[GeneratedQuestion] = []
+    for q in questions:
+        if q.pair_id:
+            by_pair[q.pair_id].append(q)
+        else:
+            singles.append(q)
+
+    kept_pairs = [q for pid, grp in by_pair.items() if _is_intact_pair(pid, grp) for q in grp]
+    return singles + kept_pairs
+
+
+def write_accepted(path: Path, units: list[QuestionUnit]) -> None:
+    """把 units 全量原子写到 path（tmp + os.replace），孪生对两题相邻落盘。
+
+    parent 不存在则自动创建；任一 pair 单元结构非法（size≠2）落盘前 fail-fast raise。
+
+    关键实现:沿用 ``write_final_bank`` 的原子写模式（先写同目录 ``.tmp`` 再
+    ``os.replace`` 覆盖，保证读到的 JSON 恒完整）。序列化复用 T9 pools.py 的
+    ``_q_to_dict``（唯一 GeneratedQuestion↔dict schema，含 pair 四字段），函数内
+    import 规避 app.question_gen↔app.harness 循环依赖。
+    """
+    from app.harness.pools import _q_to_dict
+    from app.harness.question_units import flatten_units, validate_units
+
+    validate_units(units)
+    records = [_q_to_dict(q) for q in flatten_units(units)]
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(".tmp")
+    tmp.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
+    os.replace(str(tmp), str(path))
+    logger.info("accepted 题库全量原子写: {} 单元 / {} 题 → {}", len(units), len(records), path)
+
+
+def read_accepted(path: Path) -> list[QuestionUnit]:
+    """读回 path 的 accepted JSON → 聚合为单元列表，剔除磁盘悬挂孤儿。
+
+    返回校验通过的单元列表（single + 合法 pair）；磁盘上"只落 P 未落 Q"的悬挂孤儿
+    被 warn+drop，不进结果、不 raise。
+
+    关键实现:反序列化复用 T9 pools.py 的 ``_dict_to_q``（pair 四字段 .get 兼容），
+    函数内 import 规避循环依赖。先 ``_drop_dangling_pairs`` 剔除悬挂孤儿，再
+    ``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)
+    return validate_units(build_units(kept))
diff --git a/tests/unit/test_pair_atomic_write.py b/tests/unit/test_pair_atomic_write.py
new file mode 100644
index 0000000..b6f3b8c
--- /dev/null
+++ b/tests/unit/test_pair_atomic_write.py
@@ -0,0 +1,200 @@
+"""pair_atomic_writer 纯 helper 单元测试。
+
+覆盖三件核心行为：
+- PairPendingBuffer：single 直通成单元、pair 按 pair_id 收齐才 emit、
+  未配对孤儿可查询、绑定不一致（unit_hash 语义代偿）拒绝。
+- write_accepted：全量 tmp + os.replace 原子写，落盘无残留 .tmp。
+- read_accepted：build_units + validate_units 剔除磁盘悬挂孤儿（warn+drop）。
+
+测试用真实 GeneratedQuestion 构造（single + 合法 AR pair + 故意缺 mirror 的孤儿），
+跑完整 add → write → read 往返。
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from app.harness.pools import _q_to_dict
+from app.question_gen.pair_atomic_writer import (
+    PairPendingBuffer,
+    read_accepted,
+    write_accepted,
+)
+from core.types import GeneratedQuestion
+
+
+def _make_single(qid: str = "s1") -> GeneratedQuestion:
+    """构造一条真实 single 题（pair_id=None，question_role='single'）。"""
+    return GeneratedQuestion(
+        question_id=qid,
+        video_id="vid_A",
+        task_type="Action Reasoning",
+        question=f"{qid} 问题文本？",
+        options=("A. 甲", "B. 乙", "C. 丙", "D. 丁"),
+        answer="B",
+        source_nodes=("L2_0", "L3_1"),
+        difficulty="medium",
+    )
+
+
+def _make_pair(
+    pair_id: str = "p1",
+    *,
+    video_id: str = "vid_B",
+    flip_axis: str = "before_after",
+) -> tuple[GeneratedQuestion, GeneratedQuestion]:
+    """构造一对合法 AR 孪生对（original + mirror，共享 pair_id/flip_axis）。"""
+    original = GeneratedQuestion(
+        question_id=f"{pair_id}_o",
+        video_id=video_id,
+        task_type="Action Reasoning",
+        question="事件 X 发生在事件 Y 之前吗？",
+        options=("A. 是", "B. 否", "C. 无关", "D. 无法判断"),
+        answer="A",
+        source_nodes=("L2_2",),
+        difficulty="hard",
+        pair_id=pair_id,
+        question_role="pair_original",
+        flip_axis=flip_axis,
+    )
+    mirror = GeneratedQuestion(
+        question_id=f"{pair_id}_m",
+        video_id=video_id,
+        task_type="Action Reasoning",
+        question="事件 Y 发生在事件 X 之前吗？",
+        options=("A. 是", "B. 否", "C. 无关", "D. 无法判断"),
+        answer="B",
+        source_nodes=("L2_2",),
+        difficulty="hard",
+        pair_id=pair_id,
+        question_role="pair_mirror",
+        flip_axis=flip_axis,
+    )
+    return original, mirror
+
+
+def test_single_add_emits_unit_immediately() -> None:
+    """single 题 add 后立即返回 kind='single' 单元，不进 pending。"""
+    buf = PairPendingBuffer()
+    unit = buf.add(_make_single())
+    assert unit is not None
+    assert unit.kind == "single"
+    assert unit.size == 1
+    assert buf.pending_orphans() == []
+
+
+def test_pair_emits_only_when_both_arrived() -> None:
+    """pair 首个成员进 pending 返回 None，第二个到齐才 emit 一个 pair 单元。"""
+    buf = PairPendingBuffer()
+    original, mirror = _make_pair()
+
+    assert buf.add(original) is None
+    assert len(buf.pending_orphans()) == 1
+
+    unit = buf.add(mirror)
+    assert unit is not None
+    assert unit.kind == "pair"
+    assert unit.size == 2
+    assert buf.pending_orphans() == []
+
+
+def test_pair_completes_regardless_of_arrival_order() -> None:
+    """mirror 先到、original 后到也能正确成对。"""
+    buf = PairPendingBuffer()
+    original, mirror = _make_pair()
+
+    assert buf.add(mirror) is None
+    unit = buf.add(original)
+    assert unit is not None and unit.kind == "pair" and unit.size == 2
+
+
+def test_dangling_original_stays_in_pending() -> None:
+    """只落 original 未落 mirror → 悬挂在 pending，供调用方批次末尾检测。"""
+    buf = PairPendingBuffer()
+    original, _ = _make_pair()
+    assert buf.add(original) is None
+    orphans = buf.pending_orphans()
+    assert len(orphans) == 1
+    assert orphans[0].question_id == original.question_id
+
+
+def test_inconsistent_pair_binding_rejected() -> None:
+    """同 pair_id 两成员 flip_axis 不一致 → 拒绝（unit_hash 语义代偿）。"""
+    buf = PairPendingBuffer()
+    original, _ = _make_pair(flip_axis="before_after")
+    _, bad_mirror = _make_pair(flip_axis="left_right")
+
+    buf.add(original)
+    with pytest.raises(ValueError, match="绑定不一致"):
+        buf.add(bad_mirror)
+
+
+def test_same_role_pair_rejected() -> None:
+    """同 pair_id 两成员角色相同（两个 original）→ fail-fast 拒绝。"""
+    buf = PairPendingBuffer()
+    o1, _ = _make_pair()
+    o2, _ = _make_pair()
+    buf.add(o1)
+    with pytest.raises(ValueError):
+        buf.add(o2)
+
+
+def test_write_read_roundtrip_keeps_pair_and_single(tmp_path) -> None:
+    """add → write_accepted → read_accepted 往返：single 直通、pair 不拆、内容等价。"""
+    buf = PairPendingBuffer()
+    single = _make_single()
+    original, mirror = _make_pair()
+
+    units = []
+    for q in (single, original, mirror):
+        emitted = buf.add(q)
+        if emitted is not None:
+            units.append(emitted)
+    assert buf.pending_orphans() == []
+    assert len(units) == 2  # 1 single + 1 pair
+
+    out = tmp_path / "accepted" / "bank.json"
+    write_accepted(out, units)
+
+    assert out.exists()
+    assert not out.with_suffix(".tmp").exists()  # 原子写无残留
+
+    loaded = read_accepted(out)
+    kinds = sorted(u.kind for u in loaded)
+    assert kinds == ["pair", "single"]
+
+    loaded_pair = next(u for u in loaded if u.kind == "pair")
+    assert loaded_pair.size == 2
+    assert {q.question_id for q in loaded_pair.questions} == {
+        original.question_id,
+        mirror.question_id,
+    }
+    assert {q.flip_axis for q in loaded_pair.questions} == {"before_after"}
+
+
+def test_read_drops_disk_dangling_orphan(tmp_path, caplog) -> None:
+    """磁盘上某 pair 只有 original（缺 mirror）→ read_accepted warn+drop，不 raise。"""
+    single = _make_single()
+    original, _ = _make_pair()
+    # 手工写入"只落 P 未落 Q"的记录：single + 孤零 original
+    records = [_q_to_dict(single), _q_to_dict(original)]
+    out = tmp_path / "bank.json"
+    out.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    loaded = read_accepted(out)
+
+    assert len(loaded) == 1
+    assert loaded[0].kind == "single"
+    assert loaded[0].questions[0].question_id == single.question_id
+
+
+def test_write_accepted_rejects_malformed_unit(tmp_path) -> None:
+    """validate_units 闸门：结构非法的 pair 单元（size≠2）落盘前 fail-fast。"""
+    from core.types import QuestionUnit
+
+    original, _ = _make_pair()
+    bad_unit = QuestionUnit("pair", "p1", "Action Reasoning", (original,))
+    with pytest.raises(ValueError):
+        write_accepted(tmp_path / "bad.json", [bad_unit])