feat: gate unit-arm tasks with question-slot gate (algo #6)

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

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)