diff --git a/app/harness/runner.py b/app/harness/runner.py index 0664f55..013ecb2 100644 --- a/app/harness/runner.py +++ b/app/harness/runner.py @@ -12,6 +12,7 @@ from __future__ import annotations +import asyncio import json import math import random @@ -34,6 +35,7 @@ from app.harness.checkpoint import ( ) from app.harness.config import RunConfig # noqa: TC001 — 运行时 _compute_total_steps 使用 from app.harness.gate_ladder import BaselineCache, GatePools, build_or_load_gate_pools +from app.harness.log import HarnessLog from app.harness.observation import ( write_dual_metric, write_epoch_report, @@ -45,7 +47,13 @@ from app.harness.observation import ( ) from app.harness.question_units import build_units, unit_correctness_view from app.harness.store import advance_version -from app.harness.validate import Probation, ValidationOutcome +from app.harness.validate import ( + GateSpec, + Probation, + ValidationOutcome, + _ladder_units, + validate_skills_concurrent, +) from app.harness.workspace import ( ResolvedPaths, archive_workspace, @@ -602,6 +610,32 @@ def _write_skip_report( ) +def _assert_disjoint_target_files(targets_by_type: dict[str, str]) -> None: + """断言本 step 各题型进化目标文件互不相同(设计 v3 §1 fail-fast)。 + + 题型并行进化 + 并行 gate 的前提是 skill 文件不相交;两题型 fallback 到 + 同一 default-strategy.md 时并行会互相覆盖候选与 accept,必须显式中止 + 而非静默串行(当前 12 题型均有专属文件,此断言防未来配置漂移)。 + + 参数: + targets_by_type: {题型: 解析后 skill 文件名}。 + + 返回: + 无。 + + 异常: + RuntimeError: 存在两个题型映射同一文件。 + """ + seen: dict[str, str] = {} + for task_type, target in targets_by_type.items(): + if target in seen: + raise RuntimeError( + f"题型 {seen[target]!r} 与 {task_type!r} 映射同一 skill 文件 {target!r}," + "并行进化/gate 不支持共享目标文件(设计 v3 §1)" + ) + seen[target] = task_type + + def _escape_sql_like(text: str) -> str: """转义 SQL LIKE 模式中的全部特殊字符(`\\`、`%`、`_`)为字面匹配。 @@ -1199,7 +1233,7 @@ class Runner: _guard_infra_failures(result, context="rollout") # ----------------------------------------------------------------------- - # _gate_batch_skills：per task_type gate + # _gate_batch_skills：并行进化 + 连续并发 gate（四阶段） # ----------------------------------------------------------------------- async def _gate_batch_skills( @@ -1211,18 +1245,95 @@ class Runner: pools: Pools, state: _TrainState, ) -> None: - """按 task_type 独立 evolve → 局部验证 → accept/reject。""" - from app.harness.workspace import VersionedSkillStore - from core.evolution import evolve_single_skill + """按 task_type 并行 evolve → 连续并发 gate → 字母序统一落账。 + 四阶段(设计 v3 §2.1):Phase A 并行进化(cooldown/无改动照旧跳过); + Phase B 装配 GateSpec(阶梯出题 + 案例单元排除 + n_max 截断); + Phase C validate_skills_concurrent(共享题槽,统计按阶梯序前缀推进, + 只读 state);Phase D 唯一写 state 阶段——按字母序 accept/reject 落账, + 与原串行语义等价(题型 skill 文件不相交,合并顺序仅为确定性)。 + + 参数: + epoch / step / total_steps: 训练坐标。 + diagnosis: 本 step 诊断结果(skill_case_packs 按题型分组)。 + pools: 冻结三池。 + state: 训练状态(Phase D 唯一写入点)。 + + 返回: + 无。 + """ budget = edit_budget_at( global_step=state.global_step, total_steps=total_steps, start=self._config.edit_budget_start, end=self._config.edit_budget_end, ) + + # ---- Phase A: 并行进化(冷却/无真实改动照旧写 skip 后出清) ---- + records = await self._evolve_types_parallel(epoch, step, diagnosis, budget, pools, state) + if not records: + return + _assert_disjoint_target_files({t: r.target_file for t, r in records.items()}) + + # ---- Phase B: 装配 GateSpec(阶梯出题,收编原 _run_gate_validation 前半) ---- + specs = self._assemble_gate_specs(epoch, step, diagnosis, records, pools, state) + + # ---- Phase C: 连续并发 gate(只读 state) ---- + with HarnessLog(str(self._paths.db_path), f"gate_e{epoch}_s{step}") as gate_log: + outcomes = await validate_skills_concurrent( + workspace_dir=self._config.workspace_dir, + base_skills_version=self._current_version("skills"), + specs=specs, + gate_params=GateParams( + e_confirm=self._config.gate_e_confirm, + e_provisional=self._config.gate_e_provisional, + w_net_min=self._config.gate_w_net_min, + delta_min=self._config.gate_delta_min, + lambda_dir=self._config.gate_lambda_dir, + e_rollback=self._config.gate_e_rollback, + ), + gate_guard_err=self._config.gate_guard_err, + baseline_cache=state.baseline_cache, + prompts_version=self._current_version("prompts"), + run_inference=self._make_validate_run_inference_fn(gate_log), + log=gate_log, + concurrency=self._config.concurrency, + ) + + # ---- Phase D: 唯一写 state 阶段(字母序确定性落账) ---- + self._settle_gate_outcomes(epoch, step, records, outcomes, budget, pools, state) + + async def _evolve_types_parallel( + self, + epoch: int, + step: int, + diagnosis: DiagnosisResult, + budget: int, + pools: Pools, + state: _TrainState, + ) -> dict[str, EvolutionRecord]: + """Phase A:各题型进化 asyncio.gather 并行,冷却/无改动路径写 skip 出队。 + + cooldown 与"进化未产出真实改动"(rejected/skipped/内容未变)两类路径 + 与原串行实现语义一致:写 skip_report 后不进 gate。进化互相独立 + (各题型 skill 文件不相交,VersionedSkillStore 只读基线版本), + gather 并行不改变单题型结果。 + + 参数: + epoch / step: 训练坐标。 + diagnosis: 本 step 诊断结果。 + budget: 当步编辑预算。 + pools: 冻结三池。 + state: 训练状态(只读)。 + + 返回: + {题型: EvolutionRecord},仅含产出真实改动、待 gate 的题型。 + """ + from app.harness.workspace import VersionedSkillStore + from core.evolution import evolve_single_skill + + active_types: list[str] = [] for task_type in sorted(diagnosis.skill_case_packs): - # 冷却 admission control if state.gate_cooldown.get(task_type, 0) > 0: _write_skip_report( self._config.workspace_dir, @@ -1237,22 +1348,40 @@ class Runner: budget=budget, ) continue + active_types.append(task_type) + if not active_types: + return {} + evolve_prompts = self._load_evolve_prompts() + skills_version = self._current_version("skills") + + async def _evolve_one(task_type: str) -> EvolutionRecord: pack = diagnosis.skill_case_packs[task_type] skill_store = VersionedSkillStore(self._paths.skills_dir) - evolve_prompts = self._load_evolve_prompts() - record = await evolve_single_skill( + return await evolve_single_skill( self._evolve_llm, pack, skill_store, evolve_prompts, - self._current_version("skills"), + skills_version, budget, self._config.appendix_consolidate_threshold, skill_update_mode=self._config.skill_update_mode, rejected=state.rejected_buffer.get(task_type, []), ) - # 进化未产出真实改动 + + records = dict( + zip( + active_types, + await asyncio.gather(*[_evolve_one(t) for t in active_types]), + strict=True, + ) + ) + + # 无真实改动的题型照旧写 skipped 后出队 + gated: dict[str, EvolutionRecord] = {} + for task_type in active_types: + record = records[task_type] if record.status in ("rejected", "skipped") or ( record.evolved_content == record.original_content ): @@ -1270,11 +1399,107 @@ class Runner: rank_clip_triggered=bool(record.clip_info.get("triggered", False)), ) continue + gated[task_type] = record + return gated - outcome = await self._run_gate_validation( - epoch, step, task_type, pack, record, pools, state + def _assemble_gate_specs( + self, + epoch: int, + step: int, + diagnosis: DiagnosisResult, + records: dict[str, EvolutionRecord], + pools: Pools, + state: _TrainState, + ) -> list[GateSpec]: + """Phase B:为每个待 gate 题型装配 GateSpec(阶梯出题 + 截断)。 + + 案例包按 unit 排除:把每个 case 的 question_id 映射到其所属 unit_id, + 命中单元整体排除,防止只排 AR pair 半个成员而给 gate 池灌半个 pair + (下游 _ladder_units 会 fail-fast)。base_skill_content 读 step 起点 + 版本(self._paths 在 Phase D accept 前不变),保证所有题型对同一 + 基线版本验证。核心算法保真 #5。 + + 参数: + epoch / step: 训练坐标(拼 gate_run_prefix)。 + diagnosis: 本 step 诊断结果(案例排除来源)。 + records: Phase A 产出的待 gate 进化记录。 + pools: 冻结三池(baseline_run_id)。 + state: 训练状态(只读 gate_pools / gate_epoch_observed)。 + + 返回: + 与 records 键序一致的 GateSpec 列表。 + + 异常: + RuntimeError: 阶梯引用了题库中不存在的 unit_id。 + """ + specs: list[GateSpec] = [] + for task_type, record in records.items(): + pack = diagnosis.skill_case_packs[task_type] + exclude_units = { + self._gate_questions_by_id[c.question_id].unit_id + for c in pack.failure_cases + pack.success_cases + if c.question_id in self._gate_questions_by_id + } + ladder_unit_ids = state.gate_pools.ladder_for( + task_type, + exclude_units, + p_low=self._config.gate_p_low, + p_high=self._config.gate_p_high, + cold=not state.gate_epoch_observed, ) - # 观测落库 + missing = [uid for uid in ladder_unit_ids if uid not in self._gate_units_by_id] + if missing: + raise RuntimeError( + f"gate 阶梯引用未知 unit: {missing[:5]}(gate_pools.json 与题库失配)" + ) + ladder_items = [ + q for uid in ladder_unit_ids for q in self._gate_units_by_id[uid].questions + ] + slug = task_type.lower().replace(" ", "-") + specs.append( + GateSpec( + task_type=task_type, + target_file=record.target_file, + candidate_content=record.evolved_content, + base_skill_content=(self._paths.skills_dir / record.target_file).read_text( + encoding="utf-8" + ), + units=tuple(_ladder_units(ladder_items)[: self._config.gate_n_max]), + gate_run_prefix=f"{pools.baseline_run_id}_e{epoch}_s{step}_gate_{slug}", + ) + ) + return specs + + def _settle_gate_outcomes( + self, + epoch: int, + step: int, + records: dict[str, EvolutionRecord], + outcomes: dict[str, ValidationOutcome], + budget: int, + pools: Pools, + state: _TrainState, + ) -> None: + """Phase D:按字母序统一落账(观测落库 + accept/reject 写 state)。 + + 本阶段是 _gate_batch_skills 唯一写 state 的阶段。字母序仅为确定性 + (题型 skill 文件不相交,accept 串行叠加时 _accept_skill 基于最新 + manifest 版本追加各自 target_file,互不覆盖),与原串行语义等价。 + + 参数: + epoch / step: 训练坐标。 + records: Phase A 产出的进化记录。 + outcomes: Phase C 产出的 gate 判定。 + budget: 当步编辑预算(step_report 落账)。 + pools: 冻结三池。 + state: 训练状态(唯一写入点)。 + + 返回: + 无。 + """ + for task_type in sorted(outcomes): + record = records[task_type] + outcome = outcomes[task_type] write_gate_evidence( str(self._paths.db_path), run_id=pools.baseline_run_id, @@ -1313,88 +1538,6 @@ class Runner: state.rejected_buffer, task_type, record, outcome, state.global_step ) - async def _run_gate_validation( - self, - epoch: int, - step: int, - task_type: str, - pack: Any, - record: EvolutionRecord, - pools: Pools, - state: _TrainState, - ) -> ValidationOutcome: - """CE-Gate 块序贯配对验证：阶梯出题 → 基线/候选逐块配对 → e-process 四出口。 - - 参数: - epoch: 轮次。 - step: epoch 内 step。 - task_type: 待验证题型。 - pack: SkillCasePack。 - record: 进化产物。 - pools: 冻结三池。 - state: 训练状态。 - - 返回: - ValidationOutcome。 - """ - from app.harness.log import HarnessLog - from app.harness.validate import validate_skill_local - - # 案例包按 unit 排除：把每个 case 的 question_id 映射到其所属 unit_id， - # 命中单元整体排除，防止只排 AR pair 半个成员而给 gate 池灌半个 pair - # （下游 _ladder_units 会 fail-fast）。核心算法保真 #5。 - exclude_units = { - self._gate_questions_by_id[c.question_id].unit_id - for c in pack.failure_cases + pack.success_cases - if c.question_id in self._gate_questions_by_id - } - ladder_unit_ids = state.gate_pools.ladder_for( - task_type, - exclude_units, - p_low=self._config.gate_p_low, - p_high=self._config.gate_p_high, - cold=not state.gate_epoch_observed, - ) - missing = [uid for uid in ladder_unit_ids if uid not in self._gate_units_by_id] - if missing: - raise ValueError( - f"gate 阶梯[{task_type}] 含 benchmark 中不存在的单元: " - f"{missing[:5]}（gate_pools.json 与题库失配）" - ) - # 单元展开为逐题（unit 内成员顺序保持），下游 validate 再按阶梯序聚合回单元。 - ladder_items = [q for uid in ladder_unit_ids for q in self._gate_units_by_id[uid].questions] - base_skill_content = (self._paths.skills_dir / record.target_file).read_text( - encoding="utf-8" - ) - slug = task_type.lower().replace(" ", "-") - run_inference_fn = self._make_validate_run_inference_fn() - with HarnessLog(str(self._paths.db_path), f"gate_{slug}") as gate_log: - return await validate_skill_local( - workspace_dir=self._config.workspace_dir, - base_skills_version=self._current_version("skills"), - task_type=task_type, - target_file=record.target_file, - candidate_content=record.evolved_content, - base_skill_content=base_skill_content, - ladder_items=ladder_items, - gate_params=GateParams( - e_confirm=self._config.gate_e_confirm, - e_provisional=self._config.gate_e_provisional, - w_net_min=self._config.gate_w_net_min, - delta_min=self._config.gate_delta_min, - lambda_dir=self._config.gate_lambda_dir, - e_rollback=self._config.gate_e_rollback, - ), - gate_block=self._config.gate_block, - gate_n_max=self._config.gate_n_max, - gate_guard_err=self._config.gate_guard_err, - baseline_cache=state.baseline_cache, - prompts_version=self._current_version("prompts"), - run_inference=run_inference_fn, - log=gate_log, - gate_run_prefix=(f"{pools.baseline_run_id}_e{epoch}_s{step}_gate_{slug}"), - ) - # ----------------------------------------------------------------------- # accept / reject / probation # ----------------------------------------------------------------------- @@ -2475,10 +2618,23 @@ class Runner: return _noop_builder - def _make_validate_run_inference_fn(self): - """构造 validate 用的 RunInferenceFn（绑定共享依赖）。""" + def _make_validate_run_inference_fn(self, gate_log: HarnessLog): + """构造 validate 用的 RunInferenceFn(绑定共享依赖与共享 HarnessLog)。 + + 连续并发 gate 下本函数被逐单元高频并发调用:每次调用新建 HarnessLog + 连接会重现多连接争 SQLite 写锁(遥测同款教训),故复用调用方传入的 + 单一 gate_log(单连接 + threading.Lock 串行化)。_record_run 按 run_id + 去重,避免逐单元重复 upsert。 + + 参数: + gate_log: 本 step gate 阶段共享的 HarnessLog 实例。 + + 返回: + 符合 RunInferenceFn 协议的异步推理函数。 + """ from app.harness.inference import run_inference - from app.harness.log import HarnessLog + + recorded: set[str] = set() async def _run( questions: list[GeneratedQuestion], @@ -2486,21 +2642,22 @@ class Runner: run_id: str, skills_dir: Path, ) -> InferenceResult: - self._record_run(run_id) - with HarnessLog(str(self._paths.db_path), run_id) as log: - return await run_inference( - questions=questions, - llm=self._llm, - tool_dispatch_fn=self._make_tool_dispatch_fn(skills_dir=skills_dir), - prompt_builder=self._make_prompt_builder( - skills_dir=skills_dir, prompts_dir=self._paths.prompts_dir - ), - log=log, - run_id=run_id, - concurrency=self._config.concurrency, - max_steps=self._config.max_steps, - skill_mode=self._config.skill_mode, - ) + if run_id not in recorded: + recorded.add(run_id) + self._record_run(run_id) + return await run_inference( + questions=questions, + llm=self._llm, + tool_dispatch_fn=self._make_tool_dispatch_fn(skills_dir=skills_dir), + prompt_builder=self._make_prompt_builder( + skills_dir=skills_dir, prompts_dir=self._paths.prompts_dir + ), + log=gate_log, + run_id=run_id, + concurrency=self._config.concurrency, + max_steps=self._config.max_steps, + skill_mode=self._config.skill_mode, + ) return _run diff --git a/tests/unit/test_gate_batch_parallel.py b/tests/unit/test_gate_batch_parallel.py new file mode 100644 index 0000000..6ff1725 --- /dev/null +++ b/tests/unit/test_gate_batch_parallel.py @@ -0,0 +1,238 @@ +"""_gate_batch_skills 并行装配的纯逻辑护栏 + runner 级并发编排测试。""" + +from __future__ import annotations + +import asyncio +import time +from types import SimpleNamespace +from typing import TYPE_CHECKING + +import pytest + +from app.harness import runner as runner_mod +from app.harness.question_units import build_units +from app.harness.runner import Runner, _assert_disjoint_target_files +from app.harness.validate import ValidationOutcome +from core.types import GeneratedQuestion + +if TYPE_CHECKING: + from pathlib import Path + + +def test_disjoint_target_files_pass() -> None: + """各题型映射不同文件:通过。""" + _assert_disjoint_target_files( + {"Action Reasoning": "action-reasoning.md", "Counting Problem": "counting-problem.md"} + ) + + +def test_shared_target_file_fails_fast() -> None: + """两题型 fallback 到同一文件:并行进化会互相覆盖,必须 fail-fast。""" + with pytest.raises(RuntimeError, match="default-strategy.md"): + _assert_disjoint_target_files( + {"OCR Problems": "default-strategy.md", "Spatial Reasoning": "default-strategy.md"} + ) + + +# --------------------------------------------------------------------------- +# runner 级并发编排测试(Codex 计划审 I6): +# 用 Runner.__new__ 裸实例 + 假依赖驱动 _gate_batch_skills 四阶段, +# 断言 Phase A gather 并行、Phase D 字母序落账、accept/reject 正确分派。 +# --------------------------------------------------------------------------- + +_TYPE_A = "Action Reasoning" +_TYPE_C = "Counting Problem" +_TARGET_FILES = {_TYPE_A: "action-reasoning.md", _TYPE_C: "counting-problem.md"} + + +def _question(qid: str, task_type: str) -> GeneratedQuestion: + """构造一条真实结构的 single 题目(unit_id 由 __post_init__ 回填)。""" + return GeneratedQuestion( + question_id=qid, + video_id="video-001", + task_type=task_type, + question="视频中主角最先做了什么?", + options=("A. 开门", "B. 关灯", "C. 坐下", "D. 起身"), + answer="A", + source_nodes=("L3_0001",), + difficulty="medium", + ) + + +def _record(task_type: str) -> SimpleNamespace: + """构造 EvolutionRecord 替身(仅含 _gate_batch_skills 消费的属性)。""" + return SimpleNamespace( + status="accepted", + original_content="旧 skill 内容", + evolved_content=f"进化后 skill 内容({task_type})", + target_file=_TARGET_FILES[task_type], + clip_info={}, + ) + + +def _outcome(accepted: bool) -> ValidationOutcome: + """构造真实 ValidationOutcome(一 accept 一 reject 分派用)。""" + return ValidationOutcome( + action="accept_confirmed" if accepted else "reject", + accepted=accepted, + stop_reason="confirmed" if accepted else "futility", + e_value=25.0 if accepted else 0.4, + w=3, + l=0 if accepted else 3, + n_used=4, + delta_hat=0.3 if accepted else -0.2, + delta_shrunk=0.2 if accepted else -0.1, + baseline_acc=0.5, + candidate_acc=0.8 if accepted else 0.3, + evidence_rows=[{"question_id": "q", "stop_reason": "answered"}], + ) + + +class _FakeHarnessLog: + """HarnessLog no-op 替身(上下文管理器协议)。""" + + def __init__(self, *args: object, **kwargs: object) -> None: + self.args = args + + def __enter__(self) -> _FakeHarnessLog: + return self + + def __exit__(self, *exc: object) -> bool: + return False + + +def _build_runner(tmp_path: Path) -> tuple[Runner, SimpleNamespace, SimpleNamespace]: + """构造裸 Runner 实例与 state/pools 替身(不触发真实 __init__)。""" + skills_dir = tmp_path / "skills" / "v1" + skills_dir.mkdir(parents=True) + for target in _TARGET_FILES.values(): + (skills_dir / target).write_text("旧 skill 内容", encoding="utf-8") + + runner = Runner.__new__(Runner) + runner._config = SimpleNamespace( + workspace_dir=tmp_path, + edit_budget_start=4, + edit_budget_end=2, + appendix_consolidate_threshold=3, + skill_update_mode="rewrite", + gate_p_low=0.3, + gate_p_high=0.85, + gate_n_max=8, + gate_e_confirm=20.0, + gate_e_provisional=5.0, + gate_w_net_min=2, + gate_delta_min=0.05, + gate_lambda_dir=0.5, + gate_e_rollback=0.05, + gate_guard_err=0.34, + concurrency=4, + max_steps=10, + skill_mode="live", + ) + runner._paths = SimpleNamespace( + skills_dir=skills_dir, + prompts_dir=tmp_path / "prompts", + db_path=tmp_path / "harness.db", + ) + runner._llm = object() + runner._evolve_llm = object() + runner._load_evolve_prompts = lambda: None + runner._current_version = lambda kind: "v1" + runner._class_baseline_acc = lambda *a, **k: 0.5 + runner._record_run = lambda run_id: None + + questions = {t: _question(f"q-{t[:2].lower()}", t) for t in _TARGET_FILES} + units = {t: build_units([q])[0] for t, q in questions.items()} + runner._gate_questions_by_id = {q.question_id: q for q in questions.values()} + runner._gate_units_by_id = {u.unit_id: u for u in units.values()} + unit_ids_by_type = {t: [u.unit_id] for t, u in units.items()} + + state = SimpleNamespace( + gate_cooldown={}, + rejected_buffer={}, + global_step=0, + correctness={}, + gate_epoch_observed=True, + baseline_cache=object(), + gate_pools=SimpleNamespace( + ladder_for=lambda task_type, exclude, *, p_low, p_high, cold: unit_ids_by_type[ + task_type + ] + ), + ) + pools = SimpleNamespace(baseline_run_id="baseline-run", validation=[]) + return runner, state, pools + + +def test_gate_batch_parallel_evolve_and_alphabetical_settle( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Phase A 两题型进化时间窗重叠(并行),Phase D 按字母序 accept/reject 分派。""" + import core.evolution as core_evolution + + runner, state, pools = _build_runner(tmp_path) + diagnosis = SimpleNamespace( + skill_case_packs={ + # 故意逆字母序插入,验证排序不是插入序的巧合 + _TYPE_C: SimpleNamespace(task_type=_TYPE_C, failure_cases=[], success_cases=[]), + _TYPE_A: SimpleNamespace(task_type=_TYPE_A, failure_cases=[], success_cases=[]), + } + ) + records = {t: _record(t) for t in _TARGET_FILES} + outcomes = {_TYPE_A: _outcome(accepted=True), _TYPE_C: _outcome(accepted=False)} + + evolve_windows: dict[str, tuple[float, float]] = {} + + async def fake_evolve_single_skill( + llm, pack, skill_store, prompts, version, budget, threshold, **kwargs + ): + start = time.monotonic() + await asyncio.sleep(0.05) + evolve_windows[pack.task_type] = (start, time.monotonic()) + return records[pack.task_type] + + captured: dict[str, object] = {} + + async def fake_validate_skills_concurrent(**kwargs): + captured.update(kwargs) + # 逆字母序返回,验证 Phase D 落账顺序来自 sorted 而非 dict 插入序 + return { + _TYPE_C: outcomes[_TYPE_C], + _TYPE_A: outcomes[_TYPE_A], + } + + settle_calls: list[tuple[str, str]] = [] + runner._accept_skill = lambda task_type, *a: settle_calls.append(("accept", task_type)) + runner._record_rejected_skill = lambda buf, task_type, *a: settle_calls.append( + ("reject", task_type) + ) + + monkeypatch.setattr(core_evolution, "evolve_single_skill", fake_evolve_single_skill) + monkeypatch.setattr(runner_mod, "validate_skills_concurrent", fake_validate_skills_concurrent) + monkeypatch.setattr(runner_mod, "HarnessLog", _FakeHarnessLog) + monkeypatch.setattr(runner_mod, "write_gate_evidence", lambda *a, **k: None) + monkeypatch.setattr(runner_mod, "write_step_report", lambda *a, **k: None) + monkeypatch.setattr(runner_mod, "write_quadrant_pairs", lambda *a, **k: None) + monkeypatch.setattr(runner_mod, "_outcome_to_quadrant_pairs", lambda t, o: []) + monkeypatch.setattr(runner_mod, "_write_skip_report", lambda *a, **k: None) + + asyncio.run(runner._gate_batch_skills(1, 0, diagnosis, 3, pools, state)) + + # (a) 进化时间窗重叠 = gather 真并行(串行时前者 end <= 后者 start) + win_a, win_c = evolve_windows[_TYPE_A], evolve_windows[_TYPE_C] + assert win_a[0] < win_c[1] and win_c[0] < win_a[1], f"进化未并行: {evolve_windows}" + + # (b) Phase D 落账顺序 == sorted(题型),且 (c) accept/reject 分派与 outcome 一致 + assert settle_calls == [("accept", _TYPE_A), ("reject", _TYPE_C)] + + # Phase B 装配的 GateSpec 与 Phase C 共享 log 抽查 + specs = captured["specs"] + assert [s.task_type for s in specs] == sorted(_TARGET_FILES) + for spec in specs: + assert spec.target_file == _TARGET_FILES[spec.task_type] + assert spec.base_skill_content == "旧 skill 内容" + assert spec.candidate_content == records[spec.task_type].evolved_content + assert len(spec.units) == 1 + assert "_gate_" in spec.gate_run_prefix + assert isinstance(captured["log"], _FakeHarnessLog) + assert callable(captured["run_inference"])