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 = {
-        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)