fix: gate orchestrator materialize leak + cancel-drain on abort (algo #6)

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 <noreply@anthropic.com>

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 = {
+    cand_dirs: dict[str, Path] = {}
-        r.spec.task_type: materialize_candidate_skill(
-            workspace_dir, base_skills_version, r.spec.target_file, r.spec.candidate_content
-        )
-        for r in runs
-    }
     slots_gate = _QuestionSlots(concurrency)
     try:
-        coros = [
+        # 成功一个登记一个:第 N 个题型物化抛 OSError 时,已登记的前 N-1 个
-            _run_unit_arm(
+        # 目录仍由 finally 统一清理,不泄漏(Codex 质量审 C001)。
-                r,
+        for r in runs:
-                rank,
+            cand_dirs[r.spec.task_type] = materialize_candidate_skill(
-                arm,
+                workspace_dir, base_skills_version, r.spec.target_file, r.spec.candidate_content
-                slots_gate,
+            )
-                run_inference,
+        tasks = [
-                log,
+            asyncio.ensure_future(
-                baseline_cache,
+                _run_unit_arm(
-                prompts_version,
+                    r,
-                base_skills_dir,
+                    rank,
-                cand_dirs[r.spec.task_type],
+                    arm,
-                gate_params,
+                    slots_gate,
-                gate_guard_err,
+                    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 护栏)即向上传播中止整轮,与现行"护栏
+        # 护栏 raise 中止整轮的语义不变(Codex 质量审 C002):首异常先取消其余
-        # 中止训练"语义一致;finally 仍清理候选目录。
+        # 任务并排水(return_exceptions 吞取消回报),确保外层 finally 删除候选
-        await asyncio.gather(*coros)
+        # 目录时已无在飞任务访问该目录、事件循环收尾无 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)
 

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,
+        )