refactor: fail-loud unit grading and stricter orphan pair drop

Address review: replace bare KeyError with a contextual ValueError invariant
check in _grade_unit (fail-loud, no catch/skip/fallback); tighten
_drop_orphan_pairs to require exactly one original + one mirror with no extra
illegal-role records (total==2); clarify InferenceResult docstring to unit-grained
semantics. Add tests for missing-prediction descriptive error and extra-illegal-role
pair drop.

app/harness/inference.py

@@ -34,17 +34,17 @@ if TYPE_CHECKING:
 
 @dataclass(frozen=True)
 class InferenceResult:
-    """推理聚合结果。
+    """推理聚合结果（正确率按 unit 粒度）。
 
     属性:
         run_id: 运行标识。
-        accuracy: 总正确率。
+        accuracy: unit 级正确率（correct / total）。
-        total: 总题数。
+        total: unit 总数（single 数 + pair 数，孤儿 pair 已剔除不计入）。
-        correct: 正确题数。
+        correct: 正确 unit 数（single 单题正确；pair 走 original/mirror 双向 AND）。
-        per_task_type: 按题型分组的指标 {task_type: {accuracy, total, correct}}。
+        per_task_type: 按题型分组的 unit 级指标 {task_type: {accuracy, total, correct}}。
-        steps_mean: 平均步数。
+        steps_mean: 平均步数（record 粒度，逐题溯源）。
-        token_usage: token 总用量 {prompt_tokens, completion_tokens}。
+        token_usage: token 总用量 {prompt_tokens, completion_tokens}（record 粒度）。
-        stop_reason_counts: 终止原因计数 {reason: count}。
+        stop_reason_counts: 终止原因计数 {reason: count}（record 粒度）。
     """
 
     run_id: str
@@ -210,19 +210,38 @@ def _group_by_task_type(graded: list[tuple[QuestionUnit, bool]]) -> dict[str, di
     return per_task_type
 
 
+def _is_valid_pair(group: list[GeneratedQuestion]) -> bool:
+    """判定同一 pair_id 分组是否为合法孪生对（恰好 1 original + 1 mirror，无多余）。
+
+    要求分组总数恰为 2 且角色齐备唯一；有额外非法 role 记录（total>2）或角色
+    缺失/重复均视为非法，交由调用方剔除，防非法记录混入 build_units。
+
+    参数:
+        group: 归属同一 pair_id 的题目列表。
+
+    返回:
+        合法孪生对为 True，否则 False。
+    """
+    if len(group) != 2:
+        return False
+    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")
+    return originals == 1 and mirrors == 1
+
+
 def _drop_orphan_pairs(questions: list[GeneratedQuestion]) -> list[GeneratedQuestion]:
     """剔除收不齐 2 条 / 角色非法的孤儿 pair，告警不静默。
 
     每条题目均会各答一次并逐题落库；能否合成 pair 单元仅取决于 questions
-    是否同时含该 pair_id 的 original + mirror。收不齐者告警并整对剔除，使
+    是否同时含该 pair_id 的 original + mirror（且无多余非法记录）。非法者告警并
-    后续 build_units 只面对合法孪生对（不触发 fail-fast），孤儿 unit 不计入
+    整对剔除，使后续 build_units 只面对合法孪生对（不触发 fail-fast），孤儿 unit
-    total（对齐设计 §8 聚合入口的"告警 + 剔除"）。
+    不计入 total（对齐设计 §8 聚合入口的"告警 + 剔除"）。
 
     参数:
         questions: 待聚合的题目列表（可混含 single 与孪生对成员）。
 
     返回:
-        可安全交给 build_units 的题目列表（single 全保留，pair 仅保留成对者）。
+        可安全交给 build_units 的题目列表（single 全保留，pair 仅保留合法成对者）。
     """
     by_pair: dict[str, list[GeneratedQuestion]] = defaultdict(list)
     singles: list[GeneratedQuestion] = []
@@ -234,16 +253,13 @@ def _drop_orphan_pairs(questions: list[GeneratedQuestion]) -> list[GeneratedQues
 
     kept_pairs: list[GeneratedQuestion] = []
     for pair_id, group in by_pair.items():
-        originals = [q for q in group if q.question_role == "pair_original"]
+        if _is_valid_pair(group):
-        mirrors = [q for q in group if q.question_role == "pair_mirror"]
-        if len(originals) == 1 and len(mirrors) == 1:
             kept_pairs.extend(group)
         else:
             logger.warning(
-                "孤儿 pair {}：收不齐 2 条（original={} mirror={}），剔除该 unit 不计入 total",
+                "孤儿 pair {}：非法配对（total={}），剔除该 unit 不计入 total",
                 pair_id,
-                len(originals),
+                len(group),
-                len(mirrors),
             )
     return singles + kept_pairs
 
@@ -262,6 +278,33 @@ def _per_question_correctness(records: list[dict[str, Any]]) -> dict[str, bool]:
     return {r["question_id"]: r["prediction"] == r["answer"] for r in records}
 
 
+def _grade_unit(unit: QuestionUnit, per_q: dict[str, bool]) -> bool:
+    """判定单元整体正确性，缺 prediction 时 fail-loud（带上下文）。
+
+    _drop_orphan_pairs 已剔除孤儿/非法配对，正常情况下 unit 内每题都应有对应
+    record；若仍缺失说明聚合不变量被破坏（如 records 与 questions 不同源）。此处
+    显式抛带上下文的 ValueError（fail-loud，不 catch/不跳过/不兜底），而非放任
+    unit_correctness 抛裸 KeyError 丢失定位信息。
+
+    参数:
+        unit: 待判定单元。
+        per_q: question_id → 该题是否作答正确 的映射。
+
+    返回:
+        单元整体是否正确（single 即单题正确；pair 走双向 AND）。
+
+    异常:
+        ValueError: unit 内某 question_id 不在 per_q 中（聚合不变量被破坏）。
+    """
+    missing = [q.question_id for q in unit.questions if q.question_id not in per_q]
+    if missing:
+        raise ValueError(
+            f"unit {unit.unit_id} 的 question {missing} 缺 prediction"
+            "（_drop_orphan_pairs 后不应发生，聚合不变量被破坏）"
+        )
+    return unit_correctness(unit, per_q)
+
+
 def _aggregate_results(
     records: list[dict[str, Any]],
     questions: list[GeneratedQuestion],
@@ -287,8 +330,7 @@ def _aggregate_results(
 
     per_q = _per_question_correctness(records)
     units = build_units(_drop_orphan_pairs(questions))
-    # unit 内任一题缺 record → unit_correctness 抛 KeyError（防静默兜底/读回校验）。
+    graded = [(unit, _grade_unit(unit, per_q)) for unit in units]
-    graded = [(unit, unit_correctness(unit, per_q)) for unit in units]
 
     total = len(graded)
     correct = sum(1 for _, is_correct in graded if is_correct)

tests/unit/test_inference_pair_aggregate.py

@@ -238,6 +238,45 @@ class TestUnitLevelAggregation:
         assert result.correct == 1
         assert any("orphan" in msg for msg in captured), "孤儿 pair 未告警（静默）"
 
+    def test_pair_with_extra_illegal_role_dropped(self) -> None:
+        """pair_id 下混入额外非法 role 记录（total>2）→ 整对剔除、不计入 total。"""
+        questions = [
+            _make_question("s1", answer="B"),
+            _make_question("po", answer="B", pair_id="p", question_role="pair_original"),
+            _make_question("pm", answer="A", pair_id="p", question_role="pair_mirror"),
+            # 共享 pair_id 的额外非法记录（重复 original 角色）
+            _make_question("px", answer="C", pair_id="p", question_role="pair_original"),
+        ]
+        records = [
+            _make_record("s1", prediction="B", answer="B"),
+            _make_record("po", prediction="B", answer="B"),
+            _make_record("pm", prediction="A", answer="A"),
+            _make_record("px", prediction="C", answer="C"),
+        ]
+        captured: list[str] = []
+        sink_id = logger.add(captured.append, level="WARNING", format="{message}")
+        try:
+            result = _aggregate_results(records, questions, "run-illegal-role")
+        finally:
+            logger.remove(sink_id)
+
+        assert result.total == 1  # 仅 single 存活，非法配对整对剔除
+        assert result.correct == 1
+        assert any("total=3" in msg for msg in captured), "非法配对未告警（静默）"
+
+    def test_unit_missing_prediction_raises_descriptive_error(self) -> None:
+        """unit 缺 prediction → 描述性 ValueError（fail-loud，非静默、非裸 KeyError）。
+
+        故意破坏聚合不变量（questions 含 s2 但 records 无 s2），验证带上下文报错。
+        """
+        questions = [
+            _make_question("s1", answer="B"),
+            _make_question("s2", answer="A"),
+        ]
+        records = [_make_record("s1", prediction="B", answer="B")]  # 缺 s2 的 record
+        with pytest.raises(ValueError, match=r"缺 prediction.*聚合不变量被破坏"):
+            _aggregate_results(records, questions, "run-broken-invariant")
+
     def test_empty_records_returns_zero(self) -> None:
         """空 records/questions → 零值结果。"""
         result = _aggregate_results([], [], "run-empty")