fix: isolate gate baseline-arm INFRA errors from BaselineCache (algo #6)

app/harness/validate.py

@@ -209,7 +209,8 @@ def _load_run_rows(
         _correct、steps 等字段。
     """
     rows = log.query(
-        "SELECT question_id, prediction, answer, steps_json FROM predictions WHERE run_id=?",
+        "SELECT question_id, prediction, answer, stop_reason, steps_json "
+        "FROM predictions WHERE run_id=?",
         (run_id,),
     )
     normalized: dict[str, dict[str, Any]] = {}
@@ -230,6 +231,34 @@ def _load_run_rows(
     return normalized
 
 
+# INFRA 故障 stop_reason（推理侧基础设施错误，非模型答错）：这些题的对错无信号意义，
+# 基线臂遇到时不得写入 BaselineCache（否则一次瞬时故障永久污染基线快照）。
+_INFRA_STOP_REASONS = frozenset({"error", "parse_error"})
+
+
+def _infra_question_ids_from_db(
+    log: HarnessLog,
+    run_id: str,
+    chunk: list[GeneratedQuestion],
+) -> set[str]:
+    """从 db 读取一个 run 中 stop_reason 属 INFRA 故障族的 question_id 集合。
+
+    参数:
+        log: HarnessLog 共享实例。
+        run_id: 推理 run_id。
+        chunk: 题目列表。
+
+    返回:
+        stop_reason ∈ {"error", "parse_error"} 的 question_id 集合。
+    """
+    rows = _load_run_rows(log, run_id)
+    return {
+        q.question_id
+        for q in chunk
+        if rows.get(q.question_id, {}).get("stop_reason") in _INFRA_STOP_REASONS
+    }
+
+
 def _candidate_correctness_from_db(
     log: HarnessLog,
     run_id: str,
@@ -264,13 +293,18 @@ async def _resolve_baseline_block(
     run_inference: RunInferenceFn,
     log: HarnessLog,
     run_id: str,
-) -> tuple[dict[str, bool], int, int]:
+) -> tuple[dict[str, bool], list[QuestionUnit], int, int]:
     """基线侧处理一个块：缓存优先（unit 键），miss 的单元新鲜跑基线版本并回写缓存。
 
     缓存以 unit_id 为键、存单元级对错（AR pair 双向 AND 折叠后一个布尔）。
     miss 的单元展开为逐题送推理，读回逐题预测后经 unit_correctness_view 折叠成
     单元级对错再写缓存（核心算法保真 #5）。逐题 predictions 仍逐题落库溯源。
 
+    INFRA 隔离（算法 #6）：miss 单元内**任一题** stop_reason ∈ {error, parse_error}
+    即判定该单元为 INFRA 故障——**不写 BaselineCache**（否则瞬时故障永久污染基线
+    快照）、**不入 b_units**、并从返回的有效单元集中剔除，避免污染 W/L 翻转与配对。
+    命中缓存的单元恒为有效（此前已成功验证过）。
+
     参数:
         units: 当前块的单元列表（single 或 AR pair）。
         task_type: 当前验证题型（缓存键成分）。
@@ -283,8 +317,9 @@ async def _resolve_baseline_block(
         run_id: 本块基线 run_id。
 
     返回:
-        (b_units, errors_inc, denom_inc)：块内 unit_id -> 基线单元对错、
+        (b_units, valid_units, errors_inc, denom_inc)：块内有效 unit_id -> 基线单元
-        本块新增的 INFRA error 计数与推理题次分母增量（全命中时为 0, 0）。
+        对错、剔除 INFRA 后的有效单元列表、本块新增的 INFRA error 计数与推理题次
+        分母增量（全命中时为 0, 0）。
     """
     miss_units = [
         u
@@ -293,22 +328,31 @@ async def _resolve_baseline_block(
     ]
     errors_inc = 0
     denom_inc = 0
+    infra_qids: set[str] = set()
     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)
         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)
         fresh_units = unit_correctness_view(miss_units, fresh_per_q)
-        for uid, correct in fresh_units.items():
+        # 只回写非 INFRA 单元；INFRA 单元不入缓存（不永久污染基线快照）
-            baseline_cache.put(task_type, s_hash, prompts_version, uid, correct)
+        for u in miss_units:
+            if any(q.question_id in infra_qids for q in u.questions):
+                continue
+            baseline_cache.put(task_type, s_hash, prompts_version, u.unit_id, fresh_units[u.unit_id])
+
+    valid_units = [
+        u for u in units if not any(q.question_id in infra_qids for q in u.questions)
+    ]
 
     b_units: dict[str, bool] = {}
-    for u in units:
+    for u in valid_units:
         val = baseline_cache.get(task_type, s_hash, prompts_version, u.unit_id)
         assert val is not None, f"基线缓存补齐后仍有 miss: unit={u.unit_id} run_id={run_id}"
         b_units[u.unit_id] = val
-    return b_units, errors_inc, denom_inc
+    return b_units, valid_units, errors_inc, denom_inc
 
 
 async def _run_candidate_block(
@@ -554,6 +598,7 @@ async def _run_local_validation(
     w = 0
     l = 0  # noqa: E741
     n_used = 0
+    n_excluded = 0  # 累计被 INFRA 隔离剔除的单元数（从阶梯分母扣除）
     errors = 0
     infra_denom = 0
     evidence_rows: list[dict] = []
@@ -567,8 +612,8 @@ async def _run_local_validation(
     verdict: GateVerdict | None = None
 
     for block_idx, unit_chunk in enumerate(unit_chunks):
-        # Phase 1: 基线侧（缓存优先，miss 新鲜跑）+ 候选侧（全块新鲜跑）
+        # Phase 1: 基线侧（缓存优先，miss 新鲜跑，INFRA 单元剔除）
-        b_units, err_b, den_b = await _resolve_baseline_block(
+        b_units, valid_chunk, err_b, den_b = await _resolve_baseline_block(
             units=unit_chunk,
             task_type=task_type,
             s_hash=s_hash,
@@ -579,34 +624,42 @@ async def _run_local_validation(
             log=log,
             run_id=f"{gate_run_prefix}_b{block_idx}_base",
         )
+        # 候选侧只跑基线侧判定有效（非 INFRA）的单元，保证配对 unit_ids 两侧一致
         c_per_q, err_c, den_c = await _run_candidate_block(
-            units=unit_chunk,
+            units=valid_chunk,
             cand_dir=cand_dir,
             run_inference=run_inference,
             log=log,
             run_id=f"{gate_run_prefix}_b{block_idx}_cand",
         )
 
-        # Phase 2: INFRA 护栏（跨块累计，分母 >=10 才触发）
+        # Phase 2: INFRA 护栏（跨块累计，分母 >=10 才触发）——写缓存前置于此已由
+        # _resolve_baseline_block 保证 INFRA 单元不落缓存，此处仅做整轮错误率熔断。
         errors += err_b + err_c
         infra_denom += den_b + den_c
         _check_infra_guard(errors, infra_denom, gate_guard_err)
 
-        # Phase 3: 折叠成单元视图 + 配对 + 证据行 + 块间判定
+        # 本块全 INFRA：无有效单元可配对，累计剔除数后跳过判定进入下一块
-        c_units = unit_correctness_view(unit_chunk, c_per_q)
+        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)
-        unit_ids = [u.unit_id for u in unit_chunk]
+        unit_ids = [u.unit_id for u in valid_chunk]
         pair_result = pair_block(b_units, c_units, unit_ids)
         for uid, (b, c) in pair_result.observed.items():
             base_obs[uid] = b
             cand_obs[uid] = c
 
-        block_rows = _build_evidence_rows(unit_chunk, b_units, c_units, task_type, block_idx)
+        block_rows = _build_evidence_rows(valid_chunk, b_units, c_units, task_type, block_idx)
 
         w += pair_result.w
         l += pair_result.l  # noqa: E741
-        n_used += len(unit_chunk)
+        n_used += len(valid_chunk)
-        verdict = gate_decision(w, l, n_used, n_plan - n_used, params=gate_params)
+        # 阶梯剩余按扣除 INFRA 后的有效分母计：n_remaining = (n_plan - n_excluded) - n_used
+        verdict = gate_decision(w, l, n_used, (n_plan - n_excluded) - n_used, params=gate_params)
 
         for row in block_rows:
             row["e_value"] = verdict.e_value

tests/unit/test_harness_validate.py

@@ -505,6 +505,76 @@ async def test_baseline_cache_hit(tmp_path: Path) -> None:
         log.close()
 
 
+@pytest.mark.asyncio
+async def test_baseline_infra_error_not_cached(tmp_path: Path) -> None:
+    """基线臂 INFRA error 的 unit 不写入 BaselineCache（不永久污染），且从有效单元排除。"""
+    from app.harness.gate_ladder import skill_hash
+    from app.harness.question_units import build_units
+    from app.harness.validate import _resolve_baseline_block
+
+    workspace = _setup_workspace(tmp_path)
+    log = _make_log(workspace)
+    questions = _make_questions(2)  # q0 干净, q1 INFRA error
+    units = build_units(questions)
+    cache = BaselineCache(workspace / "baseline_cache.json")
+    s_hash = skill_hash("baseline skill content")
+
+    async def mock_fn(qs, *, run_id, skills_dir):
+        for q in qs:
+            is_err = q.question_id == "q1"
+            log.insert(
+                "predictions",
+                {
+                    "run_id": run_id,
+                    "video_id": "v0",
+                    "question_id": q.question_id,
+                    "task_type": "temporal",
+                    "prediction": "" if is_err else "A",
+                    "answer": "A",
+                    "evidence": "",
+                    "reasoning": "",
+                    "steps_used": 1,
+                    "prompt_tokens": 10,
+                    "completion_tokens": 10,
+                    "stop_reason": "error" if is_err else "completed",
+                    "steps_json": "[]",
+                },
+            )
+        return InferenceResult(
+            run_id=run_id,
+            accuracy=0.5,
+            total=2,
+            correct=1,
+            per_task_type={},
+            steps_mean=1.0,
+            token_usage={"prompt_tokens": 10, "completion_tokens": 10},
+            stop_reason_counts={"completed": 1, "error": 1},
+        )
+
+    try:
+        b_units, valid_units, _errors_inc, _denom_inc = await _resolve_baseline_block(
+            units=units,
+            task_type="temporal",
+            s_hash=s_hash,
+            prompts_version="p1",
+            baseline_cache=cache,
+            base_skills_dir=workspace / "skills" / "v1",
+            run_inference=mock_fn,
+            log=log,
+            run_id="step1_gate_b0_base",
+        )
+        # q1 是 INFRA：不写缓存、不入 b_units、不在有效单元里
+        assert cache.get("temporal", s_hash, "p1", "q1") is None
+        assert "q1" not in b_units
+        assert all(u.unit_id != "q1" for u in valid_units)
+        # q0 干净：正常缓存并入 b_units/valid_units
+        assert cache.get("temporal", s_hash, "p1", "q0") is True
+        assert b_units["q0"] is True
+        assert any(u.unit_id == "q0" for u in valid_units)
+    finally:
+        log.close()
+
+
 @pytest.mark.asyncio
 async def test_last_block_terminal(tmp_path: Path) -> None:
     """单块 + n_remaining=0 → 终态判定（provisional 或 inertia），非 continue。"""