diff --git a/app/harness/validate.py b/app/harness/validate.py index 2ba2191..f4c6be4 100644 --- a/app/harness/validate.py +++ b/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 -> 基线单元对错、 - 本块新增的 INFRA error 计数与推理题次分母增量(全命中时为 0, 0)。 + (b_units, valid_units, errors_inc, denom_inc):块内有效 unit_id -> 基线单元 + 对错、剔除 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(): - baseline_cache.put(task_type, s_hash, prompts_version, uid, correct) + # 只回写非 INFRA 单元;INFRA 单元不入缓存(不永久污染基线快照) + 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 新鲜跑)+ 候选侧(全块新鲜跑) - b_units, err_b, den_b = await _resolve_baseline_block( + # Phase 1: 基线侧(缓存优先,miss 新鲜跑,INFRA 单元剔除) + 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: 折叠成单元视图 + 配对 + 证据行 + 块间判定 - c_units = unit_correctness_view(unit_chunk, c_per_q) + # 本块全 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) - 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) - verdict = gate_decision(w, l, n_used, n_plan - n_used, params=gate_params) + n_used += len(valid_chunk) + # 阶梯剩余按扣除 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 diff --git a/tests/unit/test_harness_validate.py b/tests/unit/test_harness_validate.py index 2734408..bcd1ef6 100644 --- a/tests/unit/test_harness_validate.py +++ b/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。"""