From 9e8a254fbb6c89afab367dbb2e5c24e56192fe89 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Fri, 17 Jul 2026 00:45:47 -0400 Subject: [PATCH] fix: gate orchestrator materialize leak + cancel-drain on abort (algo #6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex 质量审 2 Critical: - C001: 候选物化移入 try、成功一个登记一个,第 N 个题型物化失败时 finally 仍清理前 N-1 个已建目录,不泄漏 - C002: gather 首异常(护栏 raise)后显式取消其余任务并排水,确保 finally 删除候选目录时无在飞任务访问、事件循环无 pending task 警告; 护栏中止整轮语义不变 - 新增 2 测试:部分物化失败清理 / 护栏 raise 取消收束不悬挂(wait_for 5s) Co-Authored-By: Claude Fable 5 --- app/harness/validate.py | 57 ++++++++++-------- tests/unit/test_gate_concurrent.py | 92 ++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+), 23 deletions(-) diff --git a/app/harness/validate.py b/app/harness/validate.py index bd0c1d1..d16dbe6 100644 --- a/app/harness/validate.py +++ b/app/harness/validate.py @@ -1272,34 +1272,45 @@ async def validate_skills_concurrent( _validate_gate_specs(specs) base_skills_dir = workspace_dir / "skills" / base_skills_version runs = [_GateRun.from_spec(s) for s in specs] - cand_dirs = { - r.spec.task_type: materialize_candidate_skill( - workspace_dir, base_skills_version, r.spec.target_file, r.spec.candidate_content - ) - for r in runs - } + cand_dirs: dict[str, Path] = {} slots_gate = _QuestionSlots(concurrency) try: - coros = [ - _run_unit_arm( - r, - rank, - arm, - slots_gate, - run_inference, - log, - baseline_cache, - prompts_version, - base_skills_dir, - cand_dirs[r.spec.task_type], - gate_params, - gate_guard_err, + # 成功一个登记一个:第 N 个题型物化抛 OSError 时,已登记的前 N-1 个 + # 目录仍由 finally 统一清理,不泄漏(Codex 质量审 C001)。 + for r in runs: + cand_dirs[r.spec.task_type] = materialize_candidate_skill( + workspace_dir, base_skills_version, r.spec.target_file, r.spec.candidate_content + ) + tasks = [ + asyncio.ensure_future( + _run_unit_arm( + r, + rank, + arm, + slots_gate, + run_inference, + log, + baseline_cache, + prompts_version, + base_skills_dir, + cand_dirs[r.spec.task_type], + gate_params, + gate_guard_err, + ) ) for r, rank, arm in _build_launch_order(runs) ] - # gather 任一任务 raise(INFRA 护栏)即向上传播中止整轮,与现行"护栏 - # 中止训练"语义一致;finally 仍清理候选目录。 - await asyncio.gather(*coros) + # 护栏 raise 中止整轮的语义不变(Codex 质量审 C002):首异常先取消其余 + # 任务并排水(return_exceptions 吞取消回报),确保外层 finally 删除候选 + # 目录时已无在飞任务访问该目录、事件循环收尾无 pending task 警告; + # _run_unit_arm 的题槽获取自带取消回滚,cancel 安全。 + try: + await asyncio.gather(*tasks) + except BaseException: + for t in tasks: + t.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise finally: _cleanup_candidate_dirs(cand_dirs) diff --git a/tests/unit/test_gate_concurrent.py b/tests/unit/test_gate_concurrent.py index 24bdc72..38acd42 100644 --- a/tests/unit/test_gate_concurrent.py +++ b/tests/unit/test_gate_concurrent.py @@ -165,3 +165,95 @@ async def test_all_infra_raises(tmp_path, monkeypatch) -> None: log=log, concurrency=8, ) + + +@pytest.mark.asyncio +async def test_partial_materialize_failure_cleans_up(tmp_path, monkeypatch) -> None: + """第 2 个题型物化失败:OSError 传播,且第 1 个已物化目录被清理不泄漏。""" + spec_a = _mk_spec("Action Reasoning", "action-reasoning", 1) + spec_b = _mk_spec("Counting Problem", "counting-problem", 1) + made: list[Path] = [] + + def _mat(workspace_dir, base_skills_version, target_file, content): + if made: # 第 2 次调用:模拟磁盘错误 + raise OSError("第 2 个题型物化失败(模拟)") + d = tmp_path / "cand_a" + d.mkdir() + made.append(d) + return d + + monkeypatch.setattr("app.harness.validate.materialize_candidate_skill", _mat) + + async def _never_called(questions, *, run_id, skills_dir): + raise AssertionError("物化失败后不应发起任何推理") + + with pytest.raises(OSError): + await validate_skills_concurrent( + workspace_dir=tmp_path, + base_skills_version="v1", + specs=[spec_a, spec_b], + gate_params=_PARAMS, + gate_guard_err=0.10, + baseline_cache=BaselineCache(tmp_path / "bc.json"), + prompts_version="v1", + run_inference=_never_called, + log=_FakeLog(), + concurrency=8, + ) + assert len(made) == 1 + assert not made[0].exists() + + +@pytest.mark.asyncio +async def test_guard_raise_cancels_remaining_tasks(tmp_path, monkeypatch) -> None: + """护栏 raise 后其余在飞任务被取消收束:整体在超时内返回,不悬挂。 + + A 型 12 单元推理全 INFRA(stop_reason="error"),分母 ≥10 后错误率 1.0 + 超护栏 0.01 → RuntimeError;B 型推理挂在永不 set 的 Event 上,若无 + 取消收束,validate 将悬挂,wait_for 超时即为回归。 + """ + spec_a = _mk_spec("Action Reasoning", "action-reasoning", 12) + spec_b = _mk_spec("Counting Problem", "counting-problem", 2) + log = _FakeLog() + hang = asyncio.Event() # 永不 set:B 型推理只能靠取消收束 + + class _R: + def __init__(self, run_id, total): + self.run_id, self.total = run_id, total + + async def _run(questions, *, run_id, skills_dir): + if "counting-problem" in run_id: + await hang.wait() + for q in questions: + log.rows.append( + { + "run_id": run_id, + "question_id": q.question_id, + "prediction": "", + "answer": "A", + "stop_reason": "error", + "steps_json": "[]", + } + ) + return _R(run_id, len(questions)) + + monkeypatch.setattr( + "app.harness.validate.materialize_candidate_skill", + lambda *a, **k: tmp_path / "cand", + ) + with pytest.raises(RuntimeError): + await asyncio.wait_for( + validate_skills_concurrent( + workspace_dir=tmp_path, + base_skills_version="v1", + specs=[spec_a, spec_b], + gate_params=_PARAMS, + gate_guard_err=0.01, + baseline_cache=BaselineCache(tmp_path / "bc.json"), + prompts_version="v1", + run_inference=_run, + log=log, + concurrency=8, + ), + timeout=5, + )