From 232afd525b7860dcd5bbb7d711e35f61f251a39c Mon Sep 17 00:00:00 2001 From: iomgaa Date: Fri, 17 Jul 2026 00:13:51 -0400 Subject: [PATCH] feat: gate unit-arm tasks with question-slot gate (algo #6) --- app/harness/validate.py | 187 +++++++++++++++++++++++++++++ tests/unit/test_gate_unit_arm.py | 198 +++++++++++++++++++++++++++++++ 2 files changed, 385 insertions(+) create mode 100644 tests/unit/test_gate_unit_arm.py diff --git a/app/harness/validate.py b/app/harness/validate.py index d3d016d..0235b13 100644 --- a/app/harness/validate.py +++ b/app/harness/validate.py @@ -15,6 +15,7 @@ from __future__ import annotations +import asyncio import json import shutil import tempfile @@ -957,3 +958,189 @@ def _advance_prefix(run: _GateRun, params: GateParams) -> None: ) if run.verdict.decision != "continue": run.frozen = True + + +class _QuestionSlots: + """按题数计数的共享并发闸:峰值在飞请求恒 ≤ width(设计 v3 §2.4)。 + + 多槽获取(AR pair 一单元两题)经内部锁串行化,防多任务半持有交错死锁。 + asyncio.Semaphore 等待队列 FIFO,任务按创建序(题型 round-robin)获得槽, + 即公平调度的实现载体(Codex I2)。 + """ + + def __init__(self, width: int) -> None: + """初始化题槽闸。 + + 参数: + width: 并发宽度(全 gate 同时在飞的题数上限),必须为正。 + """ + assert width > 0, f"并发宽度必须为正: {width}" + self._width = width + self._sem = asyncio.Semaphore(width) + self._acquire_lock = asyncio.Lock() + + async def acquire(self, n: int) -> None: + """原子获取 n 个题槽。 + + fail-fast:n > 宽度时任务持锁等待永不满足的槽位 → 自死锁 + (AR pair 单元 2 题 + width=1 的病态配置,Codex plan 审 C2),直接报错。 + + 参数: + n: 申请的题槽数(单元内题目数,single=1 / AR pair=2)。 + + 返回: + 无(成功返回即持有 n 个槽,须与 release(n) 配对)。 + + 异常: + ValueError: n 超过并发宽度(否则自死锁)。 + """ + if n > self._width: + raise ValueError(f"单次申请题槽 {n} 超过并发宽度 {self._width},将自死锁") + async with self._acquire_lock: + for _ in range(n): + await self._sem.acquire() + + def release(self, n: int) -> None: + """归还 n 个题槽。 + + 参数: + n: 与 acquire 对应的题槽数。 + + 返回: + 无。 + """ + for _ in range(n): + self._sem.release() + + +async def _run_unit_arm( + run: _GateRun, + slot_idx: int, + arm: str, + slots: _QuestionSlots, + run_inference: RunInferenceFn, + log: HarnessLog, + baseline_cache: BaselineCache, + prompts_version: str, + base_skills_dir: Path, + cand_dir: Path, + gate_params: GateParams, + gate_guard_err: float, +) -> None: + """执行一个 (单元, 臂) 任务:缓存/推理 → 到达登记 → 前缀消费推进。 + + 冻结检查两次:启动时(排队任务撤销点)与获得题槽后(获槽期间被冻结)。 + base 臂缓存命中不占题槽(零推理);INFRA 单元不写缓存(不永久污染基线快照)。 + 护栏在每次臂完成时检查(等价迁移自跨块累计,设计 v3 §2.3),超阈值 raise + 中止整轮(与现行行为一致)。 + + 参数: + run: 该题型的 gate 运行时状态。 + slot_idx: 单元在阶梯中的下标。 + arm: "base" 或 "cand"。 + slots: 全 gate 共享题槽闸。 + run_inference: 注入推理函数。 + log: HarnessLog 共享实例(推理后读预测)。 + baseline_cache / prompts_version: 基线缓存及键成分。 + base_skills_dir / cand_dir: 两臂各自的 skills 目录。 + gate_params: e-process 判据(前缀消费用)。 + gate_guard_err: INFRA 错误率护栏阈值。 + + 返回: + 无(结果写入 run.slots[slot_idx] 并触发 _advance_prefix)。 + + 异常: + RuntimeError: 累计 INFRA 错误率超护栏阈值(经 _check_infra_guard)。 + """ + assert arm in ("base", "cand") + if run.frozen: + return + slot = run.slots[slot_idx] + spec = run.spec + + if arm == "base": + cached = baseline_cache.get(spec.task_type, run.s_hash, prompts_version, slot.unit.unit_id) + if cached is not None: + slot.base = cached + _advance_prefix(run, gate_params) + return + + questions = list(slot.unit.questions) + await slots.acquire(len(questions)) + try: + if run.frozen: + return + run_id = f"{spec.gate_run_prefix}_{arm}" + skills_dir = base_skills_dir if arm == "base" else cand_dir + r = await run_inference(questions, run_id=run_id, skills_dir=skills_dir) + _register_arm_arrival( + run=run, + slot=slot, + arm=arm, + questions=questions, + inference_run_id=r.run_id, + inference_total=r.total, + log=log, + baseline_cache=baseline_cache, + prompts_version=prompts_version, + ) + _check_infra_guard(run.errors, run.infra_denom, gate_guard_err) + finally: + slots.release(len(questions)) + _advance_prefix(run, gate_params) + + +def _register_arm_arrival( + run: _GateRun, + slot: _UnitSlot, + arm: str, + questions: list[GeneratedQuestion], + inference_run_id: str, + inference_total: int, + log: HarnessLog, + baseline_cache: BaselineCache, + prompts_version: str, +) -> None: + """把一次臂推理结果登记进 slot 与 run 计数器(INFRA 判定 + 对错折叠 + 回写缓存)。 + + INFRA 臂只标记不写缓存(不永久污染基线快照);正常 base 臂折叠为单元级对错并 + 回写 BaselineCache,正常 cand 臂保留逐题对错(折叠交给前缀消费,保留逐题溯源)。 + + 参数: + run: 该题型的 gate 运行时状态(errors / infra_denom 原地累加)。 + slot: 本单元的双臂到达状态(结果或 INFRA 标志原地写入)。 + arm: "base" 或 "cand"。 + questions: 本单元展开后的题目列表。 + inference_run_id: 本次推理的 run_id(DB 回读键)。 + inference_total: 本次推理的题次数(护栏分母增量)。 + log: HarnessLog 共享实例(推理后读预测)。 + baseline_cache / prompts_version: 基线缓存及键成分。 + + 返回: + 无(所有效果原地写入 run 与 slot)。 + + 关键实现细节: + errors 按单元级去重(Codex plan 审 I3):同一单元双臂都 INFRA 只计 1 个 + error,与设计 §2.3"分子=INFRA 单元数(任一臂)"及旧块实现口径一致 + (旧实现 cand 不跑 base-INFRA 单元,天然无双计)。 + """ + spec = run.spec + infra_qids = _infra_question_ids_from_db(log, inference_run_id, questions) + run.infra_denom += inference_total + if infra_qids: + if not slot.excluded(): + run.errors += 1 + if arm == "base": + slot.base_infra = True + else: + slot.cand_infra = True + return + per_q = _candidate_correctness_from_db(log, inference_run_id, questions) + if arm == "base": + folded = unit_correctness_view([slot.unit], per_q) + slot.base = folded[slot.unit.unit_id] + baseline_cache.put( + spec.task_type, run.s_hash, prompts_version, slot.unit.unit_id, slot.base + ) + else: + slot.cand_per_q = per_q diff --git a/tests/unit/test_gate_unit_arm.py b/tests/unit/test_gate_unit_arm.py new file mode 100644 index 0000000..ddb9b8b --- /dev/null +++ b/tests/unit/test_gate_unit_arm.py @@ -0,0 +1,198 @@ +"""单元臂执行任务测试:缓存命中/新鲜跑/INFRA/冻结跳过/题槽并发上限。""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +import pytest + +from app.harness.gate_ladder import BaselineCache +from app.harness.validate import ( + GateSpec, + _GateRun, + _QuestionSlots, + _run_unit_arm, +) +from tests.unit.test_gate_prefix import _PARAMS, _mk_unit # 复用 fixture + + +class _FakeLog: + """假 HarnessLog:query 返回预置 predictions 行。""" + + def __init__(self) -> None: + self.rows: list[dict] = [] + + def query(self, sql: str, params: tuple = ()) -> list[dict]: + """按 run_id（params[0]）过滤预置行,模拟只读 SELECT。""" + run_id = params[0] + return [r for r in self.rows if r["run_id"] == run_id] + + +def _mk_gate_run(n: int, tmp_path: Path) -> tuple[_GateRun, BaselineCache]: + """构造 n 个 single 单元的 gate 运行时状态与空基线缓存。""" + spec = GateSpec( + task_type="Action Reasoning", + target_file="action-reasoning.md", + candidate_content="cand", + base_skill_content="base", + units=tuple(_mk_unit(f"q{i}") for i in range(n)), + gate_run_prefix="r_e1_s0_gate_action-reasoning", + ) + return _GateRun.from_spec(spec), BaselineCache(tmp_path / "bc.json") + + +def _fake_run_inference(log: _FakeLog, correct: bool, stop_reason: str = "finished"): + """构造假推理:把每题结果写进 _FakeLog 并返回带 total 的结果对象。""" + + 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): + 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": stop_reason, + "steps_json": "[]", + } + ) + return _R(run_id, len(questions)) + + return _run + + +@pytest.mark.asyncio +async def test_base_arm_cache_hit_skips_inference(tmp_path) -> None: + """base 臂缓存命中:不调推理,slot.base 直接就位,infra_denom 不增。""" + run, cache = _mk_gate_run(1, tmp_path) + cache.put("Action Reasoning", run.s_hash, "v1", "q0", True) + called = {"n": 0} + + async def _boom(questions, *, run_id, skills_dir): + called["n"] += 1 + raise AssertionError("缓存命中不应触发推理") + + slots = _QuestionSlots(4) + await _run_unit_arm( + run, + 0, + "base", + slots, + _boom, + _FakeLog(), + cache, + "v1", + Path("/nonexistent"), + Path("/nonexistent"), + _PARAMS, + 0.10, + ) + assert called["n"] == 0 and run.slots[0].base is True and run.infra_denom == 0 + + +@pytest.mark.asyncio +async def test_base_arm_fresh_run_writes_cache(tmp_path) -> None: + """base 臂 miss 新鲜跑:结果折叠入 slot 并回写缓存。""" + run, cache = _mk_gate_run(1, tmp_path) + log = _FakeLog() + slots = _QuestionSlots(4) + await _run_unit_arm( + run, + 0, + "base", + slots, + _fake_run_inference(log, correct=True), + log, + cache, + "v1", + tmp_path, + tmp_path, + _PARAMS, + 0.10, + ) + assert run.slots[0].base is True + assert cache.get("Action Reasoning", run.s_hash, "v1", "q0") is True + assert run.infra_denom == 1 + + +@pytest.mark.asyncio +async def test_infra_arm_marks_excluded_and_no_cache(tmp_path) -> None: + """INFRA 臂:标记 infra、errors+1、不写缓存。""" + run, cache = _mk_gate_run(1, tmp_path) + log = _FakeLog() + slots = _QuestionSlots(4) + await _run_unit_arm( + run, + 0, + "base", + slots, + _fake_run_inference(log, correct=False, stop_reason="error"), + log, + cache, + "v1", + tmp_path, + tmp_path, + _PARAMS, + 0.10, + ) + assert run.slots[0].base_infra and run.errors == 1 + assert cache.get("Action Reasoning", run.s_hash, "v1", "q0") is None + + +@pytest.mark.asyncio +async def test_frozen_run_skips_launch(tmp_path) -> None: + """已冻结题型的排队臂:直接返回,不占槽不推理。""" + run, cache = _mk_gate_run(1, tmp_path) + run.frozen = True + called = {"n": 0} + + async def _boom(questions, *, run_id, skills_dir): + called["n"] += 1 + + await _run_unit_arm( + run, + 0, + "cand", + _QuestionSlots(4), + _boom, + _FakeLog(), + cache, + "v1", + tmp_path, + tmp_path, + _PARAMS, + 0.10, + ) + assert called["n"] == 0 + + +@pytest.mark.asyncio +async def test_question_slots_caps_inflight() -> None: + """题槽闸:峰值在飞数严格 ≤ 宽度(多槽获取不交错死锁)。""" + slots = _QuestionSlots(2) + peak = {"cur": 0, "max": 0} + + async def _job(n: int) -> None: + await slots.acquire(n) + peak["cur"] += n + peak["max"] = max(peak["max"], peak["cur"]) + await asyncio.sleep(0.01) + peak["cur"] -= n + slots.release(n) + + await asyncio.gather(*[_job(1) for _ in range(6)], *[_job(2) for _ in range(3)]) + assert peak["max"] <= 2 + + +@pytest.mark.asyncio +async def test_question_slots_rejects_oversized_request() -> None: + """申请槽数超宽度:fail-fast ValueError 而非自死锁(Codex C2 回归锁)。""" + slots = _QuestionSlots(1) + with pytest.raises(ValueError, match="自死锁"): + await slots.acquire(2)