From 1ecb6ba25c4971e1ef6f6291f1f0853ed65fabba Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 15 Jul 2026 09:31:42 -0400 Subject: [PATCH] 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 +++++++++++++------------- tests/unit/test_pair_atomic_write.py | 48 +++++++++++++++++- 2 files changed, 81 insertions(+), 36 deletions(-) diff --git a/app/question_gen/pair_atomic_writer.py b/app/question_gen/pair_atomic_writer.py index e5458b5..8660da8 100644 --- a/app/question_gen/pair_atomic_writer.py +++ b/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)) diff --git a/tests/unit/test_pair_atomic_write.py b/tests/unit/test_pair_atomic_write.py index b6f3b8c..71c9c6c 100644 --- a/tests/unit/test_pair_atomic_write.py +++ b/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