feat: continuous concurrent gate orchestrator (algo #6)

- 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 <noreply@anthropic.com>

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/
+        rows: 每 **单元** 一行，含 question_id/task_type/ladder_rank（阶梯序号，
-            candidate_correct/e_value（该单元所在块判定后的累计 e 值）/
+            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 不上。

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

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 形如 <slug>-q<i>)。"""
+    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,
+        )

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,