diff --git a/app/harness/observation.py b/app/harness/observation.py index a10ef0b..7b1b573 100644 --- a/app/harness/observation.py +++ b/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 准确率。 diff --git a/app/harness/validate.py b/app/harness/validate.py index f4c6be4..304d124 100644 --- a/app/harness/validate.py +++ b/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, diff --git a/tests/unit/test_harness_validate.py b/tests/unit/test_harness_validate.py index bcd1ef6..f93b641 100644 --- a/tests/unit/test_harness_validate.py +++ b/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。"""