fix: gate INFRA isolation edge cases (all-INFRA fail-loud, parse_error in guard)

app/harness/observation.py

@@ -135,7 +135,8 @@ def write_dual_metric(
         db_path: SQLite 路径。
         run_id: 训练 run ID。
         epoch: 轮次（1-based）。
-        version_kind: baseline / best_hard / best_mixed / final。
+        version_kind: baseline / best_hard / best_mixed / final / slow_candidate
+            （slow_candidate = 慢更新 R2 可能被 revert 的候选，不占 epoch 终值 final 口径）。
         skills_version / prompts_version: 评估的资源版本。
         pool: val / test。
         hard_acc: hard 准确率。

app/harness/validate.py

@@ -332,7 +332,9 @@ async def _resolve_baseline_block(
     if miss_units:
         miss_questions = flatten_units(miss_units)
         r_b = await run_inference(miss_questions, run_id=run_id, skills_dir=base_skills_dir)
-        errors_inc = r_b.stop_reason_counts.get("error", 0)
+        # 护栏错误计数与 INFRA 判定口径一致：error + parse_error 都计入，
+        # 使 parse_error 风暴同样能触发 gate_guard_err 熔断（不被绕过）。
+        errors_inc = sum(r_b.stop_reason_counts.get(reason, 0) for reason in _INFRA_STOP_REASONS)
         denom_inc = r_b.total
         infra_qids = _infra_question_ids_from_db(log, r_b.run_id, miss_questions)
         fresh_per_q = _candidate_correctness_from_db(log, r_b.run_id, miss_questions)
@@ -380,7 +382,9 @@ async def _run_candidate_block(
     questions = flatten_units(units)
     r_c = await run_inference(questions, run_id=run_id, skills_dir=cand_dir)
     c_per_q = _candidate_correctness_from_db(log, r_c.run_id, questions)
-    return c_per_q, r_c.stop_reason_counts.get("error", 0), r_c.total
+    # 护栏错误计数与 INFRA 判定口径一致：error + parse_error 都计入。
+    errors_inc = sum(r_c.stop_reason_counts.get(reason, 0) for reason in _INFRA_STOP_REASONS)
+    return c_per_q, errors_inc, r_c.total
 
 
 def _build_evidence_rows(
@@ -624,6 +628,15 @@ async def _run_local_validation(
             log=log,
             run_id=f"{gate_run_prefix}_b{block_idx}_base",
         )
+        # 本块全 INFRA：无有效单元可配对——候选无需空跑，仅把基线侧错误计入护栏后
+        # 累计剔除数进入下一块（护栏仍能在整轮 INFRA 错误率超阈值时熔断）。
+        n_excluded += len(unit_chunk) - len(valid_chunk)
+        if not valid_chunk:
+            errors += err_b
+            infra_denom += den_b
+            _check_infra_guard(errors, infra_denom, gate_guard_err)
+            continue
+
         # 候选侧只跑基线侧判定有效（非 INFRA）的单元，保证配对 unit_ids 两侧一致
         c_per_q, err_c, den_c = await _run_candidate_block(
             units=valid_chunk,
@@ -639,11 +652,6 @@ async def _run_local_validation(
         infra_denom += den_b + den_c
         _check_infra_guard(errors, infra_denom, gate_guard_err)
 
-        # 本块全 INFRA：无有效单元可配对，累计剔除数后跳过判定进入下一块
-        n_excluded += len(unit_chunk) - len(valid_chunk)
-        if not valid_chunk:
-            continue
-
         # Phase 3: 折叠成单元视图 + 配对 + 证据行 + 块间判定（均用有效单元）
         c_units = unit_correctness_view(valid_chunk, c_per_q)
         candidate_per_q.update(c_per_q)
@@ -668,8 +676,11 @@ async def _run_local_validation(
         if verdict.decision != "continue":
             break
 
+    # verdict 仍为 None ⟺ 全部单元被 INFRA 排除（空 ladder 已在入口拒绝）。
+    # 明确失败，避免落到误导性的"空阶梯"断言而无法定位为 INFRA 原因。
+    if verdict is None:
+        raise RuntimeError("gate 阶梯所有 unit 被判为 INFRA 排除，无法验证（检查推理基础设施）")
     # 最后一块判定即终态（n_remaining=0 → provisional/inertia）
-    assert verdict is not None, "空阶梯应已在 validate_skill_local 入口拒绝"
     return _finalize_outcome(
         verdict=verdict,
         w=w,

tests/unit/test_harness_validate.py

@@ -575,6 +575,144 @@ async def test_baseline_infra_error_not_cached(tmp_path: Path) -> None:
         log.close()
 
 
+@pytest.mark.asyncio
+async def test_all_infra_ladder_raises_clear_error(tmp_path: Path) -> None:
+    """整个阶梯所有 unit 都被判为 INFRA 排除 → 明确 RuntimeError（非误导性空阶梯断言）。"""
+    workspace = _setup_workspace(tmp_path)
+    log = _make_log(workspace)
+    questions = _make_questions(4)
+    cache = BaselineCache(workspace / "baseline_cache.json")
+
+    candidate_calls: list[str] = []
+
+    async def mock_fn(qs, *, run_id, skills_dir):
+        if run_id.endswith("_cand"):
+            candidate_calls.append(run_id)
+        # 基线臂逐题全部 INFRA error（候选臂在修复后不应被空跑）
+        for q in qs:
+            log.insert(
+                "predictions",
+                {
+                    "run_id": run_id,
+                    "video_id": "v0",
+                    "question_id": q.question_id,
+                    "task_type": "temporal",
+                    "prediction": "",
+                    "answer": "A",
+                    "evidence": "",
+                    "reasoning": "",
+                    "steps_used": 1,
+                    "prompt_tokens": 10,
+                    "completion_tokens": 10,
+                    "stop_reason": "error",
+                    "steps_json": "[]",
+                },
+            )
+        total = len(qs)
+        return InferenceResult(
+            run_id=run_id,
+            accuracy=0.0,
+            total=total,
+            correct=0,
+            per_task_type={},
+            steps_mean=1.0,
+            token_usage={"prompt_tokens": 10, "completion_tokens": 10},
+            stop_reason_counts={"error": total},
+        )
+
+    try:
+        with pytest.raises(RuntimeError, match="INFRA"):
+            await validate_skill_local(
+                workspace_dir=workspace,
+                base_skills_version="v1",
+                task_type="temporal",
+                target_file="temporal.md",
+                candidate_content="content",
+                base_skill_content="baseline skill content",
+                ladder_items=questions,
+                gate_params=_DEFAULT_GATE_PARAMS,
+                gate_block=4,
+                gate_n_max=20,
+                gate_guard_err=0.9,  # 高阈值：4 题 <10 分母不触发错误率护栏
+                baseline_cache=cache,
+                prompts_version="p1",
+                run_inference=mock_fn,
+                log=log,
+                gate_run_prefix="step1_gate_test",
+            )
+        # 全 INFRA 块不应触发候选空跑
+        assert candidate_calls == []
+    finally:
+        log.close()
+
+
+@pytest.mark.asyncio
+async def test_parse_error_counts_toward_guard(tmp_path: Path) -> None:
+    """stop_reason=parse_error 也计入护栏错误率（与 INFRA 判定口径一致）→ 超阈值熔断。"""
+    workspace = _setup_workspace(tmp_path)
+    log = _make_log(workspace)
+    questions = _make_questions(6)
+    cache = BaselineCache(workspace / "baseline_cache.json")
+
+    async def mock_fn(qs, *, run_id, skills_dir):
+        # 逐题 stop_reason 保持 completed（不触发 per-unit INFRA 排除），
+        # 但汇总 stop_reason_counts 报大量 parse_error（应计入护栏）。
+        for q in qs:
+            log.insert(
+                "predictions",
+                {
+                    "run_id": run_id,
+                    "video_id": "v0",
+                    "question_id": q.question_id,
+                    "task_type": "temporal",
+                    "prediction": "Z",
+                    "answer": "A",
+                    "evidence": "",
+                    "reasoning": "",
+                    "steps_used": 1,
+                    "prompt_tokens": 10,
+                    "completion_tokens": 10,
+                    "stop_reason": "completed",
+                    "steps_json": "[]",
+                },
+            )
+        total = len(qs)
+        # 两臂各 5 个 parse_error → 累计 10/12 > 0.5 触发护栏
+        return InferenceResult(
+            run_id=run_id,
+            accuracy=0.0,
+            total=total,
+            correct=0,
+            per_task_type={},
+            steps_mean=1.0,
+            token_usage={"prompt_tokens": 10, "completion_tokens": 10},
+            stop_reason_counts={"completed": 1, "parse_error": 5},
+        )
+
+    try:
+        with pytest.raises(RuntimeError, match="错误率过高"):
+            await validate_skill_local(
+                workspace_dir=workspace,
+                base_skills_version="v1",
+                task_type="temporal",
+                target_file="temporal.md",
+                candidate_content="content",
+                base_skill_content="baseline skill content",
+                ladder_items=questions,
+                gate_params=_DEFAULT_GATE_PARAMS,
+                gate_block=6,
+                gate_n_max=20,
+                gate_guard_err=0.5,
+                baseline_cache=cache,
+                prompts_version="p1",
+                run_inference=mock_fn,
+                log=log,
+                gate_run_prefix="step1_gate_test",
+            )
+    finally:
+        log.close()
+
+
 @pytest.mark.asyncio
 async def test_last_block_terminal(tmp_path: Path) -> None:
     """单块 + n_remaining=0 → 终态判定（provisional 或 inertia），非 continue。"""