From e8b66f85ab962710261b9993a4fe0175a03181dd Mon Sep 17 00:00:00 2001 From: iomgaa Date: Fri, 17 Jul 2026 00:36:01 -0400 Subject: [PATCH] feat: continuous concurrent gate orchestrator (algo #6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - validate_skills_concurrent: 多题型全部臂共享题槽并发编排,发射序 = 题型 round-robin × 阶梯序(base 先 cand 后),终态统一组装 outcome, verdict None(全 INFRA)保留 RuntimeError 语义 - gate_evidence 列 block_idx → ladder_rank(阶梯序号,0-based);旧块路径 _build_evidence_rows 仅键名同步(值仍为块号)保持落库兼容 - 新增 3 项编排测试:乱序到达前缀有序性/双题型隔离/全 INFRA raise Co-Authored-By: Claude Fable 5 --- app/harness/observation.py | 7 +- app/harness/validate.py | 152 +++++++++++++++++++++- tests/unit/test_gate_concurrent.py | 167 +++++++++++++++++++++++++ tests/unit/test_harness_observation.py | 4 +- 4 files changed, 324 insertions(+), 6 deletions(-) create mode 100644 tests/unit/test_gate_concurrent.py diff --git a/app/harness/observation.py b/app/harness/observation.py index 7b1b573..b1b9d83 100644 --- a/app/harness/observation.py +++ b/app/harness/observation.py @@ -103,7 +103,7 @@ _GATE_EVIDENCE_COLS: dict[str, str] = { # question_id 列承载 unit_id（single=question_id，pair=pair_id）； # 逐题明细在 predictions 表溯源，按 pair_id join 真实 question 表会 join 不上。 "question_id": "TEXT", - "block_idx": "INTEGER", + "ladder_rank": "INTEGER", "baseline_correct": "INTEGER", "candidate_correct": "INTEGER", "e_value": "REAL", @@ -341,8 +341,9 @@ def write_gate_evidence( run_id: 训练 run ID。 epoch: 该 gate 所属的轮次（1-based）。 step: epoch 内 step 序号（0-based）。 - rows: 每 **单元** 一行，含 question_id/task_type/block_idx/baseline_correct/ - candidate_correct/e_value（该单元所在块判定后的累计 e 值）/ + rows: 每 **单元** 一行，含 question_id/task_type/ladder_rank（阶梯序号， + 0-based）/baseline_correct/ + candidate_correct/e_value（该单元判定后的累计 e 值）/ stop_reason（仅最后一单元携带最终 stop_reason，其余空串）。 question_id 字段承载 **unit_id**（single=question_id，pair=pair_id）—— 逐题明细在 predictions 表溯源，按 pair_id join 真实 question 表会 join 不上。 diff --git a/app/harness/validate.py b/app/harness/validate.py index 54810ad..bd0c1d1 100644 --- a/app/harness/validate.py +++ b/app/harness/validate.py @@ -435,7 +435,9 @@ def _build_evidence_rows( { "question_id": u.unit_id, "task_type": task_type, - "block_idx": block_idx, + # 落库列已更名 ladder_rank(阶梯序号);旧块路径此处值仍为块号, + # 仅键名对齐 gate_evidence 表结构以保持落库兼容。 + "ladder_rank": block_idx, "baseline_correct": b_units[u.unit_id], "candidate_correct": c_units[u.unit_id], "e_value": None, @@ -1172,3 +1174,151 @@ def _register_arm_arrival( ) else: slot.cand_per_q = per_q + + +def _validate_gate_specs(specs: list[GateSpec]) -> None: + """校验各题型 gate 规格,不合法直接报错(不兜底)。 + + 参数: + specs: 各题型 gate 规格。 + + 异常: + ValueError: 阶梯为空,或 gate_run_prefix 缺 "_gate_"(防泄露过滤依赖 + 该标记识别 gate run)。 + """ + for spec in specs: + if "_gate_" not in spec.gate_run_prefix: + raise ValueError(f"gate_run_prefix 必须含 '_gate_': {spec.gate_run_prefix!r}") + if not spec.units: + raise ValueError(f"task_type={spec.task_type} 阶梯为空,无法验证") + + +def _cleanup_candidate_dirs(cand_dirs: dict[str, Path]) -> None: + """尽力清理全部候选临时目录,单个失败只记 warning 不中断其余清理。 + + 参数: + cand_dirs: task_type -> 候选临时目录路径。 + + 返回: + 无。 + """ + for d in cand_dirs.values(): + try: + shutil.rmtree(d) + except OSError as e: + logger.warning("候选临时目录清理失败 {}: {}", d, e) + + +def _build_launch_order(runs: list[_GateRun]) -> list[tuple[_GateRun, int, str]]: + """构建 (run, rank, arm) 发射队列:题型 round-robin × 题型内阶梯序。 + + 交错顺序 = rank 0 各题型 → rank 1 各题型 → ...;同一 (题型, rank) 内 + base 先 cand 后。round-robin 让各题型的阶梯头部同批起跑,配合前缀消费 + 使统计推进不因某题型阶梯过长而饿死其他题型。 + + 参数: + runs: 各题型 gate 运行时状态(slots 已按阶梯序初始化)。 + + 返回: + (run, rank, arm) 三元组列表,即任务创建顺序。 + """ + order: list[tuple[_GateRun, int, str]] = [] + max_rank = max((len(r.slots) for r in runs), default=0) + for rank in range(max_rank): + for r in runs: + if rank < len(r.slots): + for arm in ("base", "cand"): + order.append((r, rank, arm)) + return order + + +async def validate_skills_concurrent( + workspace_dir: Path, + base_skills_version: str, + specs: list[GateSpec], + gate_params: GateParams, + gate_guard_err: float, + baseline_cache: BaselineCache, + prompts_version: str, + run_inference: RunInferenceFn, + log: HarnessLog, + concurrency: int, +) -> dict[str, ValidationOutcome]: + """连续并发 gate:多题型全部臂共享题槽并发,统计按阶梯序前缀有序推进。 + + 发射顺序 = 题型 round-robin × 题型内阶梯序(base 先 cand 后);题型过线即 + 冻结,其排队任务启动时自查冻结标志撤销,in-flight 结果不计入(τ 之后样本, + 合法丢弃)。全部题型判定后统一组装 ValidationOutcome。 + + 参数: + workspace_dir: workspace 根目录(候选物化用)。 + base_skills_version: 基线 skills 版本名。 + specs: 各题型 gate 规格(units 已阶梯序 + 截断 n_max)。 + gate_params: e-process 判据阈值组。 + gate_guard_err: INFRA 错误率护栏阈值。 + baseline_cache: 基线侧单元级对错缓存。 + prompts_version: 当前 prompts 版本(缓存键成分)。 + run_inference: 注入推理函数(调用方须绑定共享 HarnessLog)。 + log: HarnessLog 共享实例(推理后读预测,与 run_inference 同库)。 + concurrency: 题槽宽度(峰值在飞题数上限)。 + + 返回: + {task_type: ValidationOutcome}。 + + 异常: + RuntimeError: INFRA 护栏超阈值,或某题型全部单元被 INFRA 排除。 + ValueError: spec 校验失败(空阶梯 / run_prefix 缺 "_gate_")。 + """ + _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 + } + 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, + ) + for r, rank, arm in _build_launch_order(runs) + ] + # gather 任一任务 raise(INFRA 护栏)即向上传播中止整轮,与现行"护栏 + # 中止训练"语义一致;finally 仍清理候选目录。 + await asyncio.gather(*coros) + finally: + _cleanup_candidate_dirs(cand_dirs) + + outcomes: dict[str, ValidationOutcome] = {} + for r in runs: + if r.verdict is None: + raise RuntimeError( + f"gate[{r.spec.task_type}] 全部 unit 被判为 INFRA 排除,无法验证(检查推理基础设施)" + ) + outcomes[r.spec.task_type] = _finalize_outcome( + verdict=r.verdict, + w=r.w, + l=r.l, + n_used=r.n_used, + n_plan=len(r.slots), + base_obs=r.base_obs, + cand_obs=r.cand_obs, + candidate_per_q=r.candidate_per_q, + evidence_rows=r.evidence_rows, + task_type=r.spec.task_type, + ) + return outcomes diff --git a/tests/unit/test_gate_concurrent.py b/tests/unit/test_gate_concurrent.py new file mode 100644 index 0000000..24bdc72 --- /dev/null +++ b/tests/unit/test_gate_concurrent.py @@ -0,0 +1,167 @@ +"""连续并发 gate 编排测试:乱序到达/多题型隔离/终态组装/全 INFRA。""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from pathlib import Path + +from app.harness.gate_ladder import BaselineCache +from app.harness.validate import GateSpec, validate_skills_concurrent +from tests.unit.test_gate_prefix import _PARAMS, _mk_unit +from tests.unit.test_gate_unit_arm import _FakeLog + + +def _mk_spec(task_type: str, slug: str, n: int) -> GateSpec: + """构造 n 个 single 单元的 gate 规格(unit_id 形如 -q)。""" + return GateSpec( + task_type=task_type, + target_file=f"{slug}.md", + candidate_content=f"cand-{slug}", + base_skill_content=f"base-{slug}", + units=tuple(_mk_unit(f"{slug}-q{i}", task_type) for i in range(n)), + gate_run_prefix=f"r_e1_s0_gate_{slug}", + ) + + +def _scripted_inference(log: _FakeLog, script: dict[str, tuple[bool, float]]): + """脚本化假推理:按 question_id+臂 决定 (对错, 延迟秒),制造乱序到达。""" + + class _R: + def __init__(self, run_id: str, total: int) -> None: + self.run_id = run_id + self.total = total + + async def _run(questions, *, run_id: str, skills_dir: Path): + arm = "cand" if run_id.endswith("_cand") else "base" + correct, delay = script[f"{questions[0].question_id}|{arm}"] + await asyncio.sleep(delay) + for q in questions: + log.rows.append( + { + "run_id": run_id, + "question_id": q.question_id, + "prediction": "A" if correct else "B", + "answer": "A", + "stop_reason": "finished", + "steps_json": "[]", + } + ) + return _R(run_id, len(questions)) + + return _run + + +@pytest.mark.asyncio +async def test_out_of_order_arrival_still_ladder_order(tmp_path, monkeypatch) -> None: + """尾部先到、头部后到:判定结果与顺序到达完全相同(前缀有序性端到端)。""" + spec = _mk_spec("Action Reasoning", "action-reasoning", 4) + log = _FakeLog() + script = {} + for i in range(4): # 头部 q0 最慢;全部翻转为 W(base 错 cand 对) + script[f"action-reasoning-q{i}|base"] = (False, 0.05 if i == 0 else 0.0) + script[f"action-reasoning-q{i}|cand"] = (True, 0.05 if i == 0 else 0.0) + monkeypatch.setattr( + "app.harness.validate.materialize_candidate_skill", + lambda *a, **k: tmp_path / "cand", + ) + outcomes = await validate_skills_concurrent( + workspace_dir=tmp_path, + base_skills_version="v1", + specs=[spec], + gate_params=_PARAMS, + gate_guard_err=0.10, + baseline_cache=BaselineCache(tmp_path / "bc.json"), + prompts_version="v1", + run_inference=_scripted_inference(log, script), + log=log, + concurrency=8, + ) + o = outcomes["Action Reasoning"] + assert o.w == 4 and o.l == 0 + assert [r["ladder_rank"] for r in o.evidence_rows] == [0, 1, 2, 3] + + +@pytest.mark.asyncio +async def test_two_types_isolated(tmp_path, monkeypatch) -> None: + """两题型并行:计数互不污染,各自独立判定。 + + A 型 4 单元全 W(题尽 accept_provisional);B 型 2 单元全平 + (futility 早停,W=L=0)——两型结果都不受对方污染。 + """ + spec_a = _mk_spec("Action Reasoning", "action-reasoning", 4) + spec_b = _mk_spec("Counting Problem", "counting-problem", 2) + log = _FakeLog() + script = {} + for i in range(4): + script[f"action-reasoning-q{i}|base"] = (False, 0.0) + script[f"action-reasoning-q{i}|cand"] = (True, 0.0) + for i in range(2): + script[f"counting-problem-q{i}|base"] = (True, 0.0) + script[f"counting-problem-q{i}|cand"] = (True, 0.0) + monkeypatch.setattr( + "app.harness.validate.materialize_candidate_skill", + lambda *a, **k: tmp_path / "cand", + ) + outcomes = 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=_scripted_inference(log, script), + log=log, + concurrency=8, + ) + assert outcomes["Action Reasoning"].w == 4 + assert outcomes["Counting Problem"].w == 0 + assert outcomes["Counting Problem"].l == 0 + + +@pytest.mark.asyncio +async def test_all_infra_raises(tmp_path, monkeypatch) -> None: + """全单元 INFRA:保留现行 RuntimeError 语义(检查推理基础设施)。""" + spec = _mk_spec("Action Reasoning", "action-reasoning", 2) + log = _FakeLog() + + class _R: + def __init__(self, run_id, total): + self.run_id, self.total = run_id, total + + async def _infra_run(questions, *, run_id, skills_dir): + 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 validate_skills_concurrent( + workspace_dir=tmp_path, + base_skills_version="v1", + specs=[spec], + gate_params=_PARAMS, + gate_guard_err=0.99, # 护栏放宽,逼出全 INFRA 分支 + baseline_cache=BaselineCache(tmp_path / "bc.json"), + prompts_version="v1", + run_inference=_infra_run, + log=log, + concurrency=8, + ) diff --git a/tests/unit/test_harness_observation.py b/tests/unit/test_harness_observation.py index c6b7fd2..8b816cc 100644 --- a/tests/unit/test_harness_observation.py +++ b/tests/unit/test_harness_observation.py @@ -166,7 +166,7 @@ def test_write_read_gate_evidence(db_path: str, run_id: str) -> None: { "task_type": "temporal", "question_id": "q1", - "block_idx": 0, + "ladder_rank": 0, "baseline_correct": 1, "candidate_correct": 1, "e_value": 1.0, @@ -175,7 +175,7 @@ def test_write_read_gate_evidence(db_path: str, run_id: str) -> None: { "task_type": "temporal", "question_id": "q2", - "block_idx": 0, + "ladder_rank": 0, "baseline_correct": 0, "candidate_correct": 1, "e_value": 2.0,