fix: count units not questions in trainability pre-flight; fail-fast when all types filtered

app/harness/runner.py

@@ -298,37 +298,48 @@ def _filter_untrainable_types(
     eval_min_per_class: int,
     trainable_min_units: int,
 ) -> tuple[Pools, list[str] | None]:
-    """剔除不可训练题型（val<eval_min_per_class 或 非test单元<trainable_min_units）。
+    """剔除不可训练题型（val 单元<eval_min_per_class 或 diag+val 单元<trainable_min_units）。
 
-    非test单元数 = 该题型 diag+val 题数（single 题 unit==题；等于 gate 阶梯该类候选数）。
+    计数以**单元（unit）**为原子：AR pair 孪生对折叠计 1 个单元（等于 gate 阶梯该类
-    test 池不过滤（继续报告全题型准确率）。在 gate 建立前调用，避免样本不足的
+    候选数），非按题目计数——否则 pair 题型会以 2 倍题目数误通过阈值。test 池不过滤
-    微型题型进入信息量阶梯导致门控崩溃。
+    （继续报告全题型准确率）。在 gate 建立前调用，避免样本不足的微型题型进入信息量
+    阶梯导致门控崩溃。
 
     参数:
         pools: 冻结三池。
         task_types: 显式题型子集（None 表示全部），过滤后按 keep 收窄。
-        eval_min_per_class: 验证池每类保底题数下限。
+        eval_min_per_class: 验证池每类保底单元数下限。
         trainable_min_units: 每类可训练所需最小 diag+val 单元数。
 
     返回:
         过滤后的 (pools, task_types)：pools.diagnosis/validation 仅保留 keep 题型，
         test 原样；task_types 收窄为 keep（原 None 时返回 sorted(keep)）。
+
+    异常:
+        RuntimeError: 过滤后无任何可训练题型（切分/阈值需调整，fail-fast 不空转训练）。
     """
-    diag_by_type = Counter(q.task_type for q in pools.diagnosis)
+    diag_by_type = Counter(u.task_type for u in build_units(pools.diagnosis))
-    val_by_type = Counter(q.task_type for q in pools.validation)
+    val_by_type = Counter(u.task_type for u in build_units(pools.validation))
     keep: set[str] = set()
     dropped: list[tuple[str, str]] = []
     for tt in set(diag_by_type) | set(val_by_type):
         n_val = val_by_type.get(tt, 0)
         n_units = diag_by_type.get(tt, 0) + n_val
         if n_val < eval_min_per_class:
-            dropped.append((tt, f"val={n_val}<{eval_min_per_class}"))
+            dropped.append((tt, f"val_units={n_val}<{eval_min_per_class}"))
         elif n_units < trainable_min_units:
             dropped.append((tt, f"units={n_units}<{trainable_min_units}"))
         else:
             keep.add(tt)
     for tt, why in sorted(dropped):
         logger.warning("可训练性预检剔除题型 {}（{}）", tt, why)
+    if not keep:
+        detail = "；".join(f"{tt}（{why}）" for tt, why in sorted(dropped))
+        raise RuntimeError(
+            "可训练性预检剔除了全部题型，无题型满足 "
+            f"val_units>={eval_min_per_class} 且 units>={trainable_min_units}："
+            f"{detail}。请调整池切分或降低阈值。"
+        )
     new_pools = replace(
         pools,
         diagnosis=[q for q in pools.diagnosis if q.task_type in keep],

tests/unit/test_harness_runner.py

@@ -75,6 +75,26 @@ class _FakeQuestion:
             object.__setattr__(self, "unit_id", self.pair_id or self.question_id)
 
 
+def _fake_pair(pair_id: str, task_type: str, video_id: str = "v1") -> list[_FakeQuestion]:
+    """构造合法孪生对（original + mirror），共享 pair_id/video_id/task_type/flip_axis。"""
+    return [
+        _FakeQuestion(
+            question_id=f"{pair_id}-o",
+            video_id=video_id,
+            task_type=task_type,
+            pair_id=pair_id,
+            question_role="pair_original",
+        ),
+        _FakeQuestion(
+            question_id=f"{pair_id}-m",
+            video_id=video_id,
+            task_type=task_type,
+            pair_id=pair_id,
+            question_role="pair_mirror",
+        ),
+    ]
+
+
 @dataclass
 class _FakePools:
     """Pools 替身。"""
@@ -426,19 +446,59 @@ class TestFilterUntrainableTypes:
         assert new_types == ["A"]
 
     def test_units_below_threshold_filtered(self) -> None:
-        """val 达标但 diag+val 单元数 < trainable_min_units 的题型被剔除。"""
+        """val 达标但 diag+val 单元数 < trainable_min_units 的题型被剔除（保留另一可训题型）。"""
-        # 题型 C：val=2（>=2）但 units=2+0=... 补 diag 使总数不足
+        # 可训题型 A：diag=8 + val=2 → units=10、val=2，保留
-        val = [_FakeQuestion(question_id=f"C-v{i}", task_type="C") for i in range(2)]
+        keep_diag = [_FakeQuestion(question_id=f"A-d{i}", task_type="A") for i in range(8)]
-        diag = [_FakeQuestion(question_id="C-d0", task_type="C")]  # units=3 < 8
+        keep_val = [_FakeQuestion(question_id=f"A-v{i}", task_type="A") for i in range(2)]
+        # 题型 C：val=2（>=2）但 units=2+1=3 < 8 → 剔除
+        val = keep_val + [_FakeQuestion(question_id=f"C-v{i}", task_type="C") for i in range(2)]
+        diag = keep_diag + [_FakeQuestion(question_id="C-d0", task_type="C")]
         pools = _FakePools(diagnosis=diag, validation=val, test=[])
 
         new_pools, new_types = _filter_untrainable_types(
             pools, task_types=None, eval_min_per_class=2, trainable_min_units=8
         )
 
-        assert new_pools.diagnosis == []
+        assert {q.task_type for q in new_pools.diagnosis} == {"A"}
-        assert new_pools.validation == []
+        assert {q.task_type for q in new_pools.validation} == {"A"}
-        assert new_types == []
+        assert new_types == ["A"]
+
+    def test_ar_pair_counted_as_units_not_questions(self) -> None:
+        """AR pair 按单元折叠计数：题目数达标但单元数不足的题型仍被剔除。"""
+        # 可训题型 A：diag=8 + val=2 single → units=10，保留
+        keep_diag = [_FakeQuestion(question_id=f"A-d{i}", task_type="A") for i in range(8)]
+        keep_val = [_FakeQuestion(question_id=f"A-v{i}", task_type="A") for i in range(2)]
+        # 题型 P：diag 3 对（6 题=3 单元）+ val 2 对（4 题=2 单元）→ 单元数=5<8，
+        # 但题目数=10>=8。按单元计数须剔除（按题目计数会误通过）。
+        pair_diag: list[_FakeQuestion] = []
+        for i in range(3):
+            pair_diag.extend(_fake_pair(f"P-d{i}", "P"))
+        pair_val: list[_FakeQuestion] = []
+        for i in range(2):
+            pair_val.extend(_fake_pair(f"P-v{i}", "P"))
+        pools = _FakePools(
+            diagnosis=keep_diag + pair_diag,
+            validation=keep_val + pair_val,
+            test=[],
+        )
+
+        new_pools, new_types = _filter_untrainable_types(
+            pools, task_types=None, eval_min_per_class=2, trainable_min_units=8
+        )
+
+        assert "P" not in {q.task_type for q in new_pools.diagnosis}
+        assert "P" not in new_types
+        assert "A" in new_types
+
+    def test_all_filtered_raises(self) -> None:
+        """所有题型都被剔除时 fail-fast：raise RuntimeError 并列出剔除原因。"""
+        diag = [_FakeQuestion(question_id="B-d0", task_type="B")]  # val=0 < 2
+        pools = _FakePools(diagnosis=diag, validation=[], test=[])
+
+        with pytest.raises(RuntimeError, match="可训练"):
+            _filter_untrainable_types(
+                pools, task_types=None, eval_min_per_class=2, trainable_min_units=8
+            )
 
     def test_test_pool_untouched(self) -> None:
         """test 池不参与过滤（继续报告全题型准确率）。"""